feat(runs): accept a "search" ListRuns filter matching run name or task name (#8035)

* runs: accept a "search" ListRuns filter matching run name or task name

The console's Runs page search box sends a single `task_name CONTAINS`
filter, so searching by run ID never worked. The Filter proto has no OR and
ListRuns ANDs every filter, so the console cannot express "run name OR task
name" on its own.

Add a synthetic "search" filter field, resolved by the run service before
the generic column allow-list conversion sees the request:

  search CONTAINS_CASE_INSENSITIVE x -> (LOWER(run_name) LIKE LOWER('%x%')) OR (LOWER(task_name) LIKE LOWER('%x%'))
  search CONTAINS x                  -> (run_name LIKE '%x%') OR (task_name LIKE '%x%')
  search EQUAL x                     -> (run_name = x) OR (task_name = x)

Any other function, or a missing, empty or multi-value term, is an
InvalidArgument. No IDL change: Filter.field is a free-form string validated
server-side. The Union cloud run service accepts the same field, so one
console request works against both backends; the console change follows
separately and must ship after this.

WatchGroups is left alone: the OSS grouped view does not consume request
filters at all.

* make devbox-build

Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com>

---------

Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com>

Author: AdilFayyaz

docker/devbox-bundled/manifests/complete.yaml

@@ -7110,17 +7110,6 @@ metadata:
   namespace: flyte
 ---
 apiVersion: v1
-kind: ServiceAccount
-metadata:
-  labels:
-    app.kubernetes.io/instance: flyte-devbox
-    app.kubernetes.io/managed-by: Helm
-    app.kubernetes.io/name: flyteconnector
-    helm.sh/chart: flyteconnector-v2.0.0
-  name: flyteconnector
-  namespace: flyte
----
-apiVersion: v1
 automountServiceAccountToken: true
 kind: ServiceAccount
 metadata:
@@ -8372,30 +8361,6 @@ spec:
 ---
 apiVersion: v1
 kind: Service
-metadata:
-  annotations:
-    projectcontour.io/upstream-protocol.h2c: grpc
-  labels:
-    app.kubernetes.io/instance: flyte-devbox
-    app.kubernetes.io/managed-by: Helm
-    app.kubernetes.io/name: flyteconnector
-    helm.sh/chart: flyteconnector-v2.0.0
-  name: flyteconnector
-  namespace: flyte
-spec:
-  ports:
-  - appProtocol: TCP
-    name: connector-grpc
-    port: 8000
-    protocol: TCP
-    targetPort: connector-grpc
-  selector:
-    app.kubernetes.io/instance: flyte-devbox
-    app.kubernetes.io/name: flyteconnector
-  type: ClusterIP
----
-apiVersion: v1
-kind: Service
 metadata:
   labels:
     app.kubernetes.io/name: embedded-postgresql
@@ -8877,59 +8842,6 @@ spec:
 ---
 apiVersion: apps/v1
 kind: Deployment
-metadata:
-  labels:
-    app.kubernetes.io/instance: flyte-devbox
-    app.kubernetes.io/managed-by: Helm
-    app.kubernetes.io/name: flyteconnector
-    helm.sh/chart: flyteconnector-v2.0.0
-  name: flyteconnector
-  namespace: flyte
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app.kubernetes.io/instance: flyte-devbox
-      app.kubernetes.io/name: flyteconnector
-  template:
-    metadata:
-      annotations: null
-      labels:
-        app.kubernetes.io/instance: flyte-devbox
-        app.kubernetes.io/managed-by: Helm
-        app.kubernetes.io/name: flyteconnector
-        helm.sh/chart: flyteconnector-v2.0.0
-    spec:
-      containers:
-      - command:
-        - c0
-        image: ghcr.io/flyteorg/flyte-connectors:py3.12-v2.3.6
-        imagePullPolicy: IfNotPresent
-        name: flyteconnector
-        ports:
-        - containerPort: 8000
-          name: connector-grpc
-        readinessProbe:
-          grpc:
-            port: 8000
-          initialDelaySeconds: 1
-          periodSeconds: 3
-        resources:
-          limits:
-            cpu: 500m
-            ephemeral-storage: 200Mi
-            memory: 300Mi
-          requests:
-            cpu: 500m
-            ephemeral-storage: 200Mi
-            memory: 200Mi
-        securityContext:
-          allowPrivilegeEscalation: false
-      securityContext: {}
-      serviceAccountName: flyteconnector
----
-apiVersion: apps/v1
-kind: Deployment
 metadata:
   labels:
     app.kubernetes.io/instance: flyte-devbox

runs/repository/impl/action_test.go

@@ -488,6 +488,57 @@ func TestListRuns_HasPausedActionFilter(t *testing.T) {
 	assert.False(t, runs[0].ParentActionName.Valid, "only the root action should be returned")
 }
 
+// TestListRuns_SearchFilter runs the console's run search predicate against a real
+// database: one term must match either the run name or the task name, case-insensitively,
+// and EQUAL on a full run name must select exactly that run.
+func TestListRuns_SearchFilter(t *testing.T) {
+	db := setupActionDB(t)
+	defer func() { db.Exec("DELETE FROM actions") }()
+	actionRepo, err := NewActionRepo(db, testDbConfig)
+	require.NoError(t, err)
+	ctx := context.Background()
+
+	// Run names are unrelated to task names so a match is attributable to one column.
+	for _, a := range []*models.Action{
+		{Project: "proj1", Domain: "domain1", RunName: "runone-abc", Name: rootActionName,
+			TaskName: sql.NullString{String: "env1.searchfn", Valid: true},
+			Phase:    int32(common.ActionPhase_ACTION_PHASE_RUNNING)},
+		{Project: "proj1", Domain: "domain1", RunName: "runtwo-abc", Name: rootActionName,
+			TaskName: sql.NullString{String: "env2.otherfn", Valid: true},
+			Phase:    int32(common.ActionPhase_ACTION_PHASE_RUNNING)},
+	} {
+		_, err := actionRepo.CreateAction(ctx, a, false)
+		require.NoError(t, err)
+	}
+
+	search := func(t *testing.T, fn common.Filter_Function, term string) []string {
+		t.Helper()
+		filter, err := NewSearchFilter(fn, []string{term})
+		require.NoError(t, err)
+		runs, err := actionRepo.ListActions(ctx, interfaces.ListResourceInput{
+			Filter: NewIsRootActionFilter().And(filter),
+			Limit:  50,
+		})
+		require.NoError(t, err)
+		names := make([]string, 0, len(runs))
+		for _, r := range runs {
+			names = append(names, r.RunName)
+		}
+		return names
+	}
+
+	assert.ElementsMatch(t, []string{"runone-abc"}, search(t, common.Filter_CONTAINS_CASE_INSENSITIVE, "RUNONE"),
+		"matches the run name case-insensitively")
+	assert.ElementsMatch(t, []string{"runone-abc"}, search(t, common.Filter_CONTAINS_CASE_INSENSITIVE, "searchfn"),
+		"matches the task name")
+	assert.ElementsMatch(t, []string{"runone-abc", "runtwo-abc"}, search(t, common.Filter_CONTAINS_CASE_INSENSITIVE, "abc"),
+		"matches either column across runs")
+	assert.ElementsMatch(t, []string{"runtwo-abc"}, search(t, common.Filter_EQUAL, "runtwo-abc"),
+		"EQUAL selects exactly the named run")
+	assert.Empty(t, search(t, common.Filter_CONTAINS_CASE_INSENSITIVE, "nomatch"),
+		"a term matching neither column returns nothing")
+}
+
 // TestListActions_KeysetPagination covers the O(n) keyset paging used by the
 // WatchActions snapshot: pages continue after the previous page's (created_at, name)
 // instead of by OFFSET. It forces tied created_at (the bulk-created map-task case) so

runs/repository/impl/filters.go

@@ -263,6 +263,69 @@ func NewDeployedByFilter(deployedBy string) interfaces.Filter {
 	return NewEqualFilter("deployed_by", deployedBy)
 }
 
+// SearchField is the synthetic filter field behind the console's run search box. A filter
+// on it matches runs whose run name OR task name satisfies the predicate, which the flat,
+// AND-ed common.Filter list cannot express on its own. It is not a column: the run service
+// resolves it with SplitSearchFilters before the generic allow-list conversion
+// (ConvertProtoFilters) sees the request. The Union cloud run service accepts the same
+// field, so one console request works against both backends.
+const SearchField = "search"
+
+// searchFunctions maps the proto filter functions accepted on SearchField to their SQL
+// expression. Contains (case-sensitive or not) is what the search box sends; EQUAL lets a
+// caller holding a full run name match it exactly instead of with a wildcard scan.
+var searchFunctions = map[common.Filter_Function]interfaces.FilterExpression{
+	common.Filter_EQUAL:                     interfaces.FilterExpressionEqual,
+	common.Filter_CONTAINS:                  interfaces.FilterExpressionContains,
+	common.Filter_CONTAINS_CASE_INSENSITIVE: interfaces.FilterExpressionContainsCaseInsensitive,
+}
+
+// NewSearchFilter builds the `run_name <op> ? OR task_name <op> ?` predicate for a
+// SearchField filter. The OR is parenthesized by compositeFilter, so it composes safely
+// with the AND-ed scope filters callers add around it.
+func NewSearchFilter(fn common.Filter_Function, values []string) (interfaces.Filter, error) {
+	expression, ok := searchFunctions[fn]
+	if !ok {
+		return nil, fmt.Errorf("unsupported filter function %s for field %q; expected EQUAL, CONTAINS or CONTAINS_CASE_INSENSITIVE", fn, SearchField)
+	}
+	if len(values) != 1 {
+		return nil, fmt.Errorf("filter on field %q expects a single value, got %d", SearchField, len(values))
+	}
+	term := values[0]
+	if term == "" {
+		return nil, fmt.Errorf("filter on field %q requires a non-empty value", SearchField)
+	}
+	runName := &basicFilter{field: "run_name", expression: expression, value: term}
+	taskName := &basicFilter{field: "task_name", expression: expression, value: term}
+	return runName.Or(taskName), nil
+}
+
+// SplitSearchFilters partitions proto filters into a single combined SearchField predicate
+// (nil when there is none; several are AND-ed) and the remaining plain column filters.
+// Callers pass the remaining filters to ConvertProtoFilters, which validates fields against
+// the column allow-list and would otherwise reject the synthetic field, and AND the returned
+// predicate in separately.
+func SplitSearchFilters(filters []*common.Filter) (interfaces.Filter, []*common.Filter, error) {
+	var search interfaces.Filter
+	remaining := make([]*common.Filter, 0, len(filters))
+	for _, f := range filters {
+		if f.GetField() != SearchField {
+			remaining = append(remaining, f)
+			continue
+		}
+		sf, err := NewSearchFilter(f.GetFunction(), f.GetValues())
+		if err != nil {
+			return nil, nil, err
+		}
+		if search == nil {
+			search = sf
+		} else {
+			search = search.And(sf)
+		}
+	}
+	return search, remaining, nil
+}
+
 // ConvertProtoFilters converts proto filters to our Filter interfaces.
 // allowedColumns is checked to prevent SQL injection via user-supplied field names.
 func ConvertProtoFilters(protoFilters []*common.Filter, allowedColumns sets.Set[string]) (interfaces.Filter, error) {

runs/repository/impl/filters_test.go

@@ -185,3 +185,127 @@ func TestParseStringFilters_ValueInState(t *testing.T) {
 	assert.Contains(t, fmt.Sprintf("%v", expr.Args[0]), "1")
 	assert.Contains(t, fmt.Sprintf("%v", expr.Args[0]), "2")
 }
+
+// TestNewSearchFilter verifies the run search predicate matches the term against the run
+// name OR the task name, using the SQL operator selected by the proto function.
+func TestNewSearchFilter(t *testing.T) {
+	cases := []struct {
+		name     string
+		fn       common.Filter_Function
+		query    string
+		wantArgs []interface{}
+	}{
+		{
+			name:     "contains case insensitive",
+			fn:       common.Filter_CONTAINS_CASE_INSENSITIVE,
+			query:    "(LOWER(run_name) LIKE LOWER(?)) OR (LOWER(task_name) LIKE LOWER(?))",
+			wantArgs: []interface{}{"%abc%", "%abc%"},
+		},
+		{
+			name:     "contains case sensitive",
+			fn:       common.Filter_CONTAINS,
+			query:    "(run_name LIKE ?) OR (task_name LIKE ?)",
+			wantArgs: []interface{}{"%abc%", "%abc%"},
+		},
+		{
+			name:     "equal",
+			fn:       common.Filter_EQUAL,
+			query:    "(run_name = ?) OR (task_name = ?)",
+			wantArgs: []interface{}{"abc", "abc"},
+		},
+	}
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			filter, err := NewSearchFilter(tt.fn, []string{"abc"})
+			require.NoError(t, err)
+
+			expr, err := filter.QueryExpression("")
+			require.NoError(t, err)
+			assert.Equal(t, tt.query, expr.Query)
+			assert.Equal(t, tt.wantArgs, expr.Args)
+		})
+	}
+}
+
+// TestNewSearchFilter_WithTablePrefix verifies both columns are table-qualified when the
+// predicate is rendered for a query that aliases the actions table.
+func TestNewSearchFilter_WithTablePrefix(t *testing.T) {
+	filter, err := NewSearchFilter(common.Filter_EQUAL, []string{"abc"})
+	require.NoError(t, err)
+
+	expr, err := filter.QueryExpression("actions")
+	require.NoError(t, err)
+	assert.Equal(t, "(actions.run_name = ?) OR (actions.task_name = ?)", expr.Query)
+}
+
+// TestNewSearchFilter_Errors verifies the search field rejects operators that make no sense
+// for a free-text search and malformed value lists, so a bad request fails loudly instead
+// of silently matching everything.
+func TestNewSearchFilter_Errors(t *testing.T) {
+	cases := []struct {
+		name   string
+		fn     common.Filter_Function
+		values []string
+	}{
+		{name: "unsupported function", fn: common.Filter_GREATER_THAN, values: []string{"abc"}},
+		{name: "value in is not a search", fn: common.Filter_VALUE_IN, values: []string{"a", "b"}},
+		{name: "no values", fn: common.Filter_CONTAINS_CASE_INSENSITIVE, values: nil},
+		{name: "multiple values", fn: common.Filter_CONTAINS_CASE_INSENSITIVE, values: []string{"a", "b"}},
+		{name: "empty term", fn: common.Filter_CONTAINS_CASE_INSENSITIVE, values: []string{""}},
+	}
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			filter, err := NewSearchFilter(tt.fn, tt.values)
+			require.Error(t, err)
+			assert.Nil(t, filter)
+		})
+	}
+}
+
+// TestSplitSearchFilters verifies the "search" field is pulled out of the filter list and
+// resolved to the run-name-or-task-name predicate, while plain column filters pass through
+// for ConvertProtoFilters.
+func TestSplitSearchFilters(t *testing.T) {
+	filters := []*common.Filter{
+		{Function: common.Filter_EQUAL, Field: "phase", Values: []string{"2"}},
+		{Function: common.Filter_CONTAINS_CASE_INSENSITIVE, Field: SearchField, Values: []string{"abc"}},
+	}
+
+	search, remaining, err := SplitSearchFilters(filters)
+	require.NoError(t, err)
+
+	require.Len(t, remaining, 1)
+	assert.Equal(t, "phase", remaining[0].GetField())
+
+	require.NotNil(t, search)
+	expr, err := search.QueryExpression("")
+	require.NoError(t, err)
+	assert.Equal(t, "(LOWER(run_name) LIKE LOWER(?)) OR (LOWER(task_name) LIKE LOWER(?))", expr.Query)
+	assert.Equal(t, []interface{}{"%abc%", "%abc%"}, expr.Args)
+}
+
+// TestSplitSearchFilters_NoSearch verifies a nil predicate is returned when there is no
+// search filter, so the caller skips the extra AND and leaves the request untouched.
+func TestSplitSearchFilters_NoSearch(t *testing.T) {
+	filters := []*common.Filter{
+		{Function: common.Filter_EQUAL, Field: "phase", Values: []string{"2"}},
+	}
+
+	search, remaining, err := SplitSearchFilters(filters)
+	require.NoError(t, err)
+	assert.Nil(t, search)
+	assert.Len(t, remaining, 1)
+}
+
+// TestSplitSearchFilters_Error verifies a malformed search filter fails the whole split, so
+// the handler returns InvalidArgument instead of dropping the predicate.
+func TestSplitSearchFilters_Error(t *testing.T) {
+	filters := []*common.Filter{
+		{Function: common.Filter_GREATER_THAN, Field: SearchField, Values: []string{"abc"}},
+	}
+
+	search, remaining, err := SplitSearchFilters(filters)
+	require.Error(t, err)
+	assert.Nil(t, search)
+	assert.Nil(t, remaining)
+}

runs/service/run_service.go

@@ -947,8 +947,24 @@ func (s *RunService) ListRuns(
 		scopeFilter = scopeFilter.And(impl.NewHasPausedActionFilter())
 	}
 
+	// The console's run search box arrives as a filter on the synthetic "search" field
+	// (run name OR task name). Resolve it here and AND it onto the scope filter: the
+	// generic parser below validates fields against the action column allow-list and
+	// would reject it.
+	listReq := req.Msg.Request
+	searchFilter, remainingFilters, err := impl.SplitSearchFilters(listReq.GetFilters())
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInvalidArgument, err)
+	}
+	if searchFilter != nil {
+		scopeFilter = scopeFilter.And(searchFilter)
+		// Clone so the caller's request is not mutated when the search filter is stripped.
+		listReq = proto.Clone(listReq).(*common.ListRequest)
+		listReq.Filters = remainingFilters
+	}
+
 	// Parse pagination, sort, and user-supplied filters from the common ListRequest.
-	listInput, err := impl.NewListResourceInputFromProto(req.Msg.Request, models.ActionColumnsSet)
+	listInput, err := impl.NewListResourceInputFromProto(listReq, models.ActionColumnsSet)
 	if err != nil {
 		return nil, connect.NewError(connect.CodeInvalidArgument, err)
 	}