Goldmane API updates (#9784)

felix/collector/goldmane/client.go

@@ -106,9 +106,11 @@ func convertFlowlogToGoldmane(fl *flowlog.FlowLog) *proto.Flow {
 			Proto:    utils.ProtoToString(fl.Tuple.Proto),
 			Reporter: string(fl.Reporter),
 			Action:   string(fl.Action),
-			Policies: &proto.FlowLogPolicy{
-				// TODO (mazdak): need to add other policysets
-				AllPolicies: ensurePolicies(fl.FlowAllPolicySet),
+			Policies: &proto.PolicyTrace{
+				// TODO: Right now, Goldmane only supports Pending/Enforced policies, but
+				// Felix uses AllPolicies. Use EnforcedPolicies as the transmissiong
+				// mechanism for now, until Felix is updated to use Pending/Enforced.
+				EnforcedPolicies: toPolicyHits(fl.FlowAllPolicySet),
 			},
 		},
 	}
@@ -129,7 +131,7 @@ func ConvertGoldmaneToFlowlog(gl *proto.Flow) flowlog.FlowLog {
 
 	fl.SrcLabels = ensureFlowLogLabels(gl.SourceLabels)
 	fl.DstLabels = ensureFlowLogLabels(gl.DestLabels)
-	fl.FlowAllPolicySet = ensureFlowLogPolicies(gl.Key.Policies.AllPolicies)
+	fl.FlowAllPolicySet = toFlowPolicySet(gl.Key.Policies.EnforcedPolicies)
 
 	fl.SrcMeta = endpoint.Metadata{
 		Type:           endpoint.Type(gl.Key.SourceType),
@@ -158,22 +160,28 @@ func ConvertGoldmaneToFlowlog(gl *proto.Flow) flowlog.FlowLog {
 	return fl
 }
 
-func ensurePolicies(labels flowlog.FlowPolicySet) []string {
-	var policies []string
+// toPolicyHits converts a FlowPolicySet to a slice of policy hits in Goldmane protobuf format.
+func toPolicyHits(labels flowlog.FlowPolicySet) []*proto.PolicyHit {
+	var hits []*proto.PolicyHit
 	for p := range labels {
-		policies = append(policies, p)
+		h, err := proto.HitFromString(p)
+		if err != nil {
+			logrus.WithError(err).WithField("label", p).Panic("Failed to parse policy hit")
+		}
+		hits = append(hits, h)
 	}
-	return policies
+	return hits
 }
 
-func ensureFlowLogPolicies(policies []string) flowlog.FlowPolicySet {
+// toFlowPolicySet converts a slice of policy hits in Goldmane protobuf format to a FlowPolicySet.
+func toFlowPolicySet(policies []*proto.PolicyHit) flowlog.FlowPolicySet {
 	if policies == nil {
 		return nil
 	}
 
 	policySet := make(flowlog.FlowPolicySet)
 	for _, pol := range policies {
-		policySet[pol] = struct{}{}
+		policySet[pol.ToString()] = struct{}{}
 	}
 	return policySet
 }

felix/fv/flow_logs_goldmane_staged_test.go

@@ -70,7 +70,7 @@ import (
 
 // These tests include tests of Kubernetes policies as well as other policy types. To ensure we have the correct
 // behavior, run using the Kubernetes infrastructure only.
-var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ flow log with staged policy tests", []apiconfig.DatastoreType{apiconfig.Kubernetes}, func(getInfra infrastructure.InfraFactory) {
+var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ goldmane flow log with staged policy tests", []apiconfig.DatastoreType{apiconfig.Kubernetes}, func(getInfra infrastructure.InfraFactory) {
 	const (
 		wepPort = 8055
 		svcPort = 8066

felix/fv/flowlogs/goldmane_mock.go

@@ -21,7 +21,7 @@ import (
 
 	"google.golang.org/grpc"
 
-	"github.com/projectcalico/calico/goldmane/pkg/collector"
+	"github.com/projectcalico/calico/goldmane/pkg/server"
 	"github.com/projectcalico/calico/goldmane/proto"
 )
 
@@ -69,7 +69,7 @@ func (g *GoldmaneMock) Run() {
 	g.once.Do(func() {
 		g.grpcServer = grpc.NewServer()
 		g.store = newFlowStore()
-		col := collector.NewFlowCollector(g.store)
+		col := server.NewFlowCollector(g.store)
 		col.RegisterWith(g.grpcServer)
 
 		l, err := net.Listen("unix", g.sockAddr)

goldmane/.gitignore

@@ -0,0 +1,2 @@
+coverage.profile
+*.log

goldmane/Makefile

@@ -59,7 +59,7 @@ gen-files: $(GENERATED_FILES)
 ###############################################################################
 ci: static-checks ut
 ut:
-	$(DOCKER_GO_BUILD) go test ./... -cover -count 1
+	$(DOCKER_GO_BUILD) go test ./... -coverprofile coverage.profile -count 1
 
 ###############################################################################
 # Release