Add endpoint workload filters

This adds support to Hubble CLI for filtering against endpoints workloads
The server side of this was implemented in cilium/cilium#21296

Signed-off-by: Chance Zibolski <chance.zibolski@gmail.com>

cmd/observe/flows.go

@@ -406,6 +406,16 @@ more.`,
 		"to-port", ofilter,
 		"Show only flows with the given destination port (e.g. 8080)"))
 
+	filterFlags.Var(filterVar(
+		"from-workload", ofilter,
+		"Show all flows originating at an endpoint with the given workload"))
+	filterFlags.Var(filterVar(
+		"workload", ofilter,
+		"Show all flows related to an endpoint with the given workload"))
+	filterFlags.Var(filterVar(
+		"to-workload", ofilter,
+		"Show all flows terminating at an endpoint with the given workload"))
+
 	filterFlags.Var(filterVar(
 		"from-identity", ofilter,
 		"Show all flows originating at an endpoint with the given security identity"))

cmd/observe/flows_filter.go

@@ -135,6 +135,8 @@ func newFlowFilter() *flowFilter {
 			{"port", "from-port"},
 			{"identity", "to-identity"},
 			{"identity", "from-identity"},
+			{"workload", "to-workload"},
+			{"workload", "from-workload"},
 			{"node-name"},
 			{"tcp-flags"},
 		},
@@ -489,6 +491,26 @@ func (of *flowFilter) set(f *filterTracker, name, val string, track bool) error
 			f.Protocol = append(f.Protocol, val)
 		})
 
+	// workload filters
+	case "workload":
+		workload := parseWorkload(val)
+		f.applyLeft(func(f *flowpb.FlowFilter) {
+			f.SourceWorkload = append(f.SourceWorkload, workload)
+		})
+		f.applyRight(func(f *flowpb.FlowFilter) {
+			f.DestinationWorkload = append(f.DestinationWorkload, workload)
+		})
+	case "from-workload":
+		workload := parseWorkload(val)
+		f.applyLeft(func(f *flowpb.FlowFilter) {
+			f.SourceWorkload = append(f.SourceWorkload, workload)
+		})
+	case "to-workload":
+		workload := parseWorkload(val)
+		f.applyRight(func(f *flowpb.FlowFilter) {
+			f.DestinationWorkload = append(f.DestinationWorkload, workload)
+		})
+
 	// identity filters
 	case "identity":
 		identity, err := parseIdentity(val)

cmd/observe/workload.go

@@ -0,0 +1,36 @@
+// Copyright 2022 Authors of Hubble
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package observe
+
+import (
+	"strings"
+
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+)
+
+// parseWorkload parse and returns workloads
+func parseWorkload(s string) *flowpb.Workload {
+	if s == "" {
+		return &flowpb.Workload{}
+	}
+	var kind, name string
+	elements := strings.SplitN(s, "/", 2)
+	if len(elements) == 1 { // foo-deploy
+		name = elements[0]
+	} else { // Deployment/foo-deploy and Deployment/
+		kind, name = elements[0], elements[1]
+	}
+	return &flowpb.Workload{Kind: kind, Name: name}
+}

cmd/observe/workload_test.go

@@ -0,0 +1,62 @@
+// Copyright 2022 Authors of Hubble
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package observe
+
+import (
+	"testing"
+
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseWorkload(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected *flowpb.Workload
+	}{
+		{
+			name:     "empty",
+			expected: &flowpb.Workload{},
+		},
+		{
+			name:     "kind and name",
+			input:    "Deployment/foo-deploy",
+			expected: &flowpb.Workload{Kind: "Deployment", Name: "foo-deploy"},
+		},
+		{
+			name:     "kind only",
+			input:    "Deployment/",
+			expected: &flowpb.Workload{Kind: "Deployment"},
+		},
+		{
+			name:     "name only", // no trailing /
+			input:    "foo-deploy",
+			expected: &flowpb.Workload{Name: "foo-deploy"},
+		},
+		{
+			name:  "multiple slashes",
+			input: "Deployment/foo/bar/",
+			// this isn't a valid resource name, but we don't validate that extensively
+			expected: &flowpb.Workload{Kind: "Deployment", Name: "foo/bar/"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := parseWorkload(tt.input)
+			assert.Equal(t, tt.expected, got)
+		})
+	}
+}