hubble: Support --cel-expression filter in hubble observe

hubble/cmd/observe/flows.go

@@ -523,6 +523,9 @@ func newFlowsCmdHelper(usage cmdUsage, vp *viper.Viper, ofilter *flowFilter) *co
 	filterFlags.Var(filterVar(
 		"traffic-direction", ofilter,
 		"Show all flows in the given traffic direction (either ingress or egress)"))
+	filterFlags.Var(filterVar(
+		"cel-expression", ofilter,
+		"Filter flows using the given CEL expression"))
 
 	rawFilterFlags.StringArray(allowlistFlag, []string{}, "Specify allowlist as JSON encoded FlowFilters")
 	rawFilterFlags.StringArray(denylistFlag, []string{}, "Specify denylist as JSON encoded FlowFilters")

hubble/cmd/observe/flows_filter.go

@@ -212,6 +212,7 @@ func newFlowFilter() *flowFilter {
 			{"tcp-flags"},
 			{"uuid"},
 			{"traffic-direction"},
+			{"cel-expression"},
 		},
 	}
 }
@@ -719,6 +720,14 @@ func (of *flowFilter) set(f *filterTracker, name, val string, track bool) error
 		default:
 			return fmt.Errorf("%s: invalid traffic direction, expected ingress or egress", td)
 		}
+
+	case "cel-expression":
+		f.apply(func(f *flowpb.FlowFilter) {
+			if f.GetExperimental() == nil {
+				f.Experimental = &flowpb.FlowFilter_Experimental{}
+			}
+			f.Experimental.CelExpression = append(f.Experimental.CelExpression, val)
+		})
 	}
 
 	if err := f.checkNamespaceConflicts(f.left); err != nil {

hubble/cmd/observe/flows_filter_test.go

@@ -1003,3 +1003,45 @@ func TestCluster(t *testing.T) {
 		})
 	}
 }
+
+func TestCELExpression(t *testing.T) {
+	tt := []struct {
+		name    string
+		flags   []string
+		filters []*flowpb.FlowFilter
+		err     string
+	}{
+		{
+			name:  "Single CEL expression filter",
+			flags: []string{"--cel-expression", "_flow.verdict == Verdict.FORWARDED || _flow.verdict == Verdict.TRANSLATED"},
+			filters: []*flowpb.FlowFilter{
+				{Experimental: &flowpb.FlowFilter_Experimental{CelExpression: []string{"_flow.verdict == Verdict.FORWARDED || _flow.verdict == Verdict.TRANSLATED"}}},
+			},
+		},
+		{
+			name:  "Multiple CEL expression filter",
+			flags: []string{"--cel-expression", "_flow.verdict == Verdict.FORWARDED", "--cel-expression", "_flow.verdict == Verdict.TRANSLATED"},
+			filters: []*flowpb.FlowFilter{
+				{Experimental: &flowpb.FlowFilter_Experimental{CelExpression: []string{"_flow.verdict == Verdict.FORWARDED", "_flow.verdict == Verdict.TRANSLATED"}}},
+			},
+		},
+	}
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newFlowFilter()
+			cmd := newFlowsCmdWithFilter(viper.New(), f)
+			err := cmd.Flags().Parse(tc.flags)
+			if tc.err != "" {
+				require.Errorf(t, err, tc.err)
+				return
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Nil(t, f.blacklist)
+			diff := cmp.Diff(tc.filters, f.whitelist.flowFilters(), cmpopts.IgnoreUnexported(flowpb.FlowFilter{}, flowpb.FlowFilter_Experimental{}))
+			if diff != "" {
+				t.Errorf("mismatch (-want +got):\n%s", diff)
+			}
+		})
+	}
+}

hubble/cmd/observe_help.txt

@@ -39,6 +39,7 @@ Selectors Flags:
 
 Filters Flags:
   -A, --all-namespaces filter[=true]        Show all flows in any Kubernetes namespace.
+      --cel-expression filter               Filter flows using the given CEL expression
       --cluster filter                      Show all flows which match the cluster names (e.g. "test-cluster", "prod-*")
       --drop-reason-desc filter             Show only flows which match this drop reason describe (e.g. "POLICY_DENIED", "UNSUPPORTED_L3_PROTOCOL")
       --fqdn filter                         Show all flows related to the given fully qualified domain name (e.g. "*.cilium.io").