Fix FVs

Author: mazdakn

felix/collector/goldmane/client.go

@@ -56,7 +56,7 @@ func NewReporter(addr, cert, key, ca string) (*GoldmaneReporter, error) {
 		client:  cli,
 
 		// Do not send flowlogs to node socket, if goldmane address set via FelixConfig is equal to node socket
-		mayReportToNodeSocket: addr != LocalGoldmaneServer,
+		mayReportToNodeSocket: addr != NodeSocketAddress,
 	}, nil
 }
 
@@ -85,7 +85,7 @@ func (g *GoldmaneReporter) nodeSocketReporter() {
 }
 
 func NodeSocketExists() bool {
-	_, err := os.Stat(LocalGoldmaneServer)
+	_, err := os.Stat(NodeSocketAddress)
 	// In case of any error, return false
 	return err == nil
 }
@@ -105,8 +105,7 @@ func (g *GoldmaneReporter) mayStartNodeSocketReporter() {
 	var err error
 	g.nodeClientLock.Lock()
 	defer g.nodeClientLock.Unlock()
-	sockAddr := fmt.Sprintf("unix://%v", LocalGoldmaneServer)
-	g.nodeClient, err = client.NewFlowClient(sockAddr, "", "", "")
+	g.nodeClient, err = client.NewFlowClient(NodeSocketAddress, "", "", "")
 	if err != nil {
 		logrus.WithError(err).Warn("Failed to create goldmane unix client")
 		return

felix/collector/goldmane/node_server.go

@@ -26,7 +26,9 @@ import (
 )
 
 const (
-	LocalGoldmaneServer = "/var/log/calico/flowlogs/goldmane.sock"
+	NodeSocketAddress = "unix:///var/log/calico/flowlogs/goldmane.sock"
+	NodeSocketPath    = "/var/log/calico/flowlogs"
+	NodeSocketName    = "goldmane.sock"
 )
 
 type flowStore struct {

felix/fv/flow_logs_goldmane_staged_test.go

@@ -101,7 +101,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ goldmane flow log with stag
 
 		opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "5"
 		opts.ExtraEnvVars["FELIX_FLOWLOGSCOLLECTORDEBUGTRACE"] = "true"
-		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)
@@ -824,7 +824,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ aggregation of flow log wit
 
 		opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "5"
 		opts.ExtraEnvVars["FELIX_FLOWLOGSCOLLECTORDEBUGTRACE"] = "true"
-		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)
@@ -1573,7 +1573,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ goldmane flow log with stag
 
 		opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "3"
 		opts.ExtraEnvVars["FELIX_FLOWLOGSCOLLECTORDEBUGTRACE"] = "true"
-		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)

felix/fv/flow_logs_goldmane_test.go

@@ -106,7 +106,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ goldmane flow log tests", [
 
 		opts.ExtraEnvVars["FELIX_FLOWLOGSCOLLECTORDEBUGTRACE"] = "true"
 		opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "2"
-		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 	})
 
 	JustBeforeEach(func() {
@@ -592,7 +592,7 @@ var _ = infrastructure.DatastoreDescribe("goldmane flow log ipv6 tests", []apico
 		opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "2"
 		opts.ExtraEnvVars["FELIX_IPV6SUPPORT"] = "true"
 		opts.ExtraEnvVars["FELIX_DefaultEndpointToHostAction"] = "RETURN"
-		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)
 

felix/fv/infrastructure/felix.go

@@ -203,11 +203,11 @@ func RunFelix(infra DatastoreInfra, id int, options TopologyOptions) *Felix {
 	// aren't enabled the directory will just stay empty.
 	logDir := path.Join(cwLogDir, uniqueName)
 	Expect(os.MkdirAll(logDir, 0o777)).NotTo(HaveOccurred())
-	args = append(args, "-v", logDir+":/var/log/calico/flowlogs")
+	args = append(args, "-v", fmt.Sprintf("%v:%v", logDir, goldmane.NodeSocketPath))
 
 	var goldmaneServer *goldmane.NodeServer
 	if options.FlowLogSource == FlowLogSourceGoldmane {
-		sockAddr := fmt.Sprintf("%v/goldmane.sock", logDir)
+		sockAddr := path.Join(logDir, goldmane.NodeSocketName)
 		goldmaneServer = goldmane.NewNodeServer(sockAddr)
 		err := goldmaneServer.Run()
 		if err != nil {

felix/fv/tiered_policy_test.go

@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/projectcalico/calico/felix/collector/flowlog"
+	"github.com/projectcalico/calico/felix/collector/goldmane"
 	"github.com/projectcalico/calico/felix/collector/types/endpoint"
 	"github.com/projectcalico/calico/felix/collector/types/tuple"
 	"github.com/projectcalico/calico/felix/fv/connectivity"
@@ -443,7 +444,7 @@ var _ = infrastructure.DatastoreDescribe("connectivity tests and flow logs with
 
 			opts.ExtraEnvVars["FELIX_FLOWLOGSCOLLECTORDEBUGTRACE"] = "true"
 			opts.ExtraEnvVars["FELIX_FLOWLOGSFLUSHINTERVAL"] = "2"
-			opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = flowlogs.LocalGoldmaneServer
+			opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 			testSetup()