state_store/fix the prefix bugs for scaling policies documented in 1a…

…9318

CHANGELOG.md

@@ -8,6 +8,7 @@ IMPROVEMENTS:
 BUG FIXES:
 
  * core: Fixed a bug where unpromoted job versions are used when rescheduling failed allocations [[GH-8691](https://github.com/hashicorp/nomad/issues/8691)]
+ * core: Fixed bugs where scaling policies could be matched against incorrect jobs with a similar prefix [[GH-8753](https://github.com/hashicorp/nomad/issues/8753)]
 
 ## 0.12.3 (August 13, 2020)
 

nomad/scaling_endpoint_test.go

@@ -136,6 +136,7 @@ func TestScalingEndpoint_ListPolicies(t *testing.T) {
 	get := &structs.ScalingPolicyListRequest{
 		QueryOptions: structs.QueryOptions{
 			Region: "global",
+			Namespace: "default",
 		},
 	}
 	var resp structs.ACLPolicyListResponse
@@ -170,6 +171,7 @@ func TestScalingEndpoint_ListPolicies_ACL(t *testing.T) {
 	get := &structs.ScalingPolicyListRequest{
 		QueryOptions: structs.QueryOptions{
 			Region: "global",
+			Namespace: "default",
 		},
 	}
 
@@ -261,6 +263,7 @@ func TestScalingEndpoint_ListPolicies_Blocking(t *testing.T) {
 	req := &structs.ScalingPolicyListRequest{
 		QueryOptions: structs.QueryOptions{
 			Region:        "global",
+			Namespace: "default",
 			MinQueryIndex: 150,
 		},
 	}

nomad/state/state_store.go

@@ -1539,11 +1539,30 @@ func (s *StateStore) DeleteJobTxn(index uint64, namespace, jobID string, txn Txn
 
 // deleteJobScalingPolicies deletes any scaling policies associated with the job
 func (s *StateStore) deleteJobScalingPolicies(index uint64, job *structs.Job, txn *memdb.Txn) error {
-	numDeletedScalingPolicies, err := txn.DeleteAll("scaling_policy", "target_prefix", job.Namespace, job.ID)
+	iter, err := s.ScalingPoliciesByJobTxn(nil, job.Namespace, job.ID, txn)
 	if err != nil {
-		return fmt.Errorf("deleting job scaling policies failed: %v", err)
+		return fmt.Errorf("getting job scaling policies for deletion failed: %v", err)
+	}
+
+	// Put them into a slice so there are no safety concerns while actually
+	// performing the deletes
+	policies := []interface{}{}
+	for {
+		raw := iter.Next()
+		if raw == nil {
+			break
+		}
+		policies = append(policies, raw)
+	}
+
+	// Do the deletes
+	for _, p := range policies {
+		if err := txn.Delete("scaling_policy", p); err != nil {
+			return fmt.Errorf("deleting scaling policy failed: %v", err)
+		}
 	}
-	if numDeletedScalingPolicies > 0 {
+
+	if len(policies) > 0 {
 		if err := txn.Insert("index", &IndexEntry{"scaling_policy", index}); err != nil {
 			return fmt.Errorf("index update failed: %v", err)
 		}
@@ -5332,7 +5351,19 @@ func (s *StateStore) ScalingPoliciesByNamespace(ws memdb.WatchSet, namespace str
 	}
 
 	ws.Add(iter.WatchCh())
-	return iter, nil
+
+	filter := func(raw interface{}) bool {
+		d, ok := raw.(*structs.ScalingPolicy)
+		if !ok {
+			return true
+		}
+
+		return d.Target[structs.ScalingTargetNamespace] != namespace
+	}
+
+	// Wrap the iterator in a filter
+	wrap := memdb.NewFilterIterator(iter, filter)
+	return wrap, nil
 }
 
 func (s *StateStore) ScalingPoliciesByJob(ws memdb.WatchSet, namespace, jobID string) (memdb.ResultIterator, error) {
@@ -5349,7 +5380,19 @@ func (s *StateStore) ScalingPoliciesByJobTxn(ws memdb.WatchSet, namespace, jobID
 	}
 
 	ws.Add(iter.WatchCh())
-	return iter, nil
+
+	filter := func(raw interface{}) bool {
+		d, ok := raw.(*structs.ScalingPolicy)
+		if !ok {
+			return true
+		}
+
+		return d.Target[structs.ScalingTargetJob] != jobID
+	}
+
+	// Wrap the iterator in a filter
+	wrap := memdb.NewFilterIterator(iter, filter)
+	return wrap, nil
 }
 
 func (s *StateStore) ScalingPolicyByID(ws memdb.WatchSet, id string) (*structs.ScalingPolicy, error) {