Allow agents to require labels on workflows (#5633)

server/grpc/filter.go

@@ -29,6 +29,10 @@ func createFilterFunc(agentFilter rpc.Filter) queue.FilterFn {
 		// Create a copy of the labels for filtering to avoid modifying the original task
 		labels := maps.Clone(task.Labels)
 
+		if requiredLabelsMissing(labels, agentFilter.Labels) {
+			return false, 0
+		}
+
 		// ignore internal labels for filtering
 		for k := range labels {
 			if strings.HasPrefix(k, pipelineConsts.InternalLabelPrefix) {
@@ -64,3 +68,15 @@ func createFilterFunc(agentFilter rpc.Filter) queue.FilterFn {
 		return true, score
 	}
 }
+
+func requiredLabelsMissing(taskLabels, agentLabels map[string]string) bool {
+	for label, value := range agentLabels {
+		if len(label) > 0 && label[0] == '!' {
+			val, ok := taskLabels[label[1:]]
+			if !ok || val != value {
+				return true
+			}
+		}
+	}
+	return false
+}

server/grpc/filter_test.go

@@ -131,3 +131,44 @@ func TestCreateFilterFunc(t *testing.T) {
 		})
 	}
 }
+
+func TestMissingRequiredLabels(t *testing.T) {
+	t.Parallel()
+
+	testdata := []struct {
+		taskLabels     map[string]string
+		requiredLabels map[string]string
+		want           bool
+	}{
+		// Required label present and matches
+		{
+			taskLabels:     map[string]string{"os": "linux"},
+			requiredLabels: map[string]string{"!os": "linux", "platform": "arm64"},
+			want:           false,
+		},
+		// Required label present but does not match
+		{
+			taskLabels:     map[string]string{"os": "windows"},
+			requiredLabels: map[string]string{"!os": "linux", "platform": "amd64"},
+			want:           true,
+		},
+		// Required label missing
+		{
+			taskLabels:     map[string]string{"arch": "amd64"},
+			requiredLabels: map[string]string{"!os": "linux"},
+			want:           true,
+		},
+		// No agent labels
+		{
+			taskLabels:     map[string]string{"os": "linux"},
+			requiredLabels: map[string]string{},
+			want:           false,
+		},
+	}
+
+	for _, tt := range testdata {
+		if got := requiredLabelsMissing(tt.taskLabels, tt.requiredLabels); got != tt.want {
+			t.Errorf("requiredLabelsMissing(%v, %v) = %v, want %v", tt.taskLabels, tt.requiredLabels, got, tt.want)
+		}
+	}
+}

server/queue/queue.go

@@ -67,7 +67,7 @@ func (t *InfoT) String() string {
 	return sb.String()
 }
 
-// Filter filters tasks in the queue. If the Filter returns false,
+// FilterFn filters tasks in the queue. If the Filter returns false,
 // the Task is skipped and not returned to the subscriber.
 // The int return value represents the matching score (higher is better).
 type FilterFn func(*model.Task) (bool, int)