Flowlogs: make calico-node to fetch flow logs from a node

felix/collector/goldmane/client.go

@@ -38,6 +38,11 @@ type GoldmaneReporter struct {
 	address string
 	client  *client.FlowClient
 	once    sync.Once
+
+	// Fields related to goldmane node socket
+	mayReportToNodeSocket bool
+	nodeClient            *client.FlowClient
+	nodeClientLock        sync.RWMutex
 }
 
 func NewReporter(addr, cert, key, ca string) (*GoldmaneReporter, error) {
@@ -48,6 +53,9 @@ func NewReporter(addr, cert, key, ca string) (*GoldmaneReporter, error) {
 	return &GoldmaneReporter{
 		address: addr,
 		client:  cli,
+
+		// Do not send flowlogs to node socket, if goldmane address set via FelixConfig is equal to node socket
+		mayReportToNodeSocket: addr != NodeSocketAddress,
 	}, nil
 }
 
@@ -56,18 +64,79 @@ func (g *GoldmaneReporter) Start() error {
 	g.once.Do(func() {
 		// We don't wait for the initial connection to start so we don't block the caller.
 		g.client.Connect(context.Background())
+
+		if g.mayReportToNodeSocket {
+			go g.nodeSocketReporter()
+		}
 	})
 	return err
 }
 
+func (g *GoldmaneReporter) nodeSocketReporter() {
+	for {
+		if NodeSocketExists() {
+			g.mayStartNodeSocketReporter()
+		} else {
+			g.mayStopNodeSocketReporter()
+		}
+		time.Sleep(time.Second * 10)
+	}
+}
+
+func (g *GoldmaneReporter) nodeClientIsNil() bool {
+	g.nodeClientLock.RLock()
+	defer g.nodeClientLock.RUnlock()
+	return g.nodeClient == nil
+}
+
+func (g *GoldmaneReporter) mayStartNodeSocketReporter() {
+	// If node socket is already setup, do not try to set it up again.
+	if !g.nodeClientIsNil() {
+		return
+	}
+
+	var err error
+	g.nodeClientLock.Lock()
+	defer g.nodeClientLock.Unlock()
+	g.nodeClient, err = client.NewFlowClient(NodeSocketAddress, "", "", "")
+	if err != nil {
+		logrus.WithError(err).Warn("Failed to create goldmane unix client")
+		return
+	}
+	logrus.Info("Created goldmane node client")
+	g.nodeClient.Connect(context.Background())
+}
+
+func (g *GoldmaneReporter) mayStopNodeSocketReporter() {
+	// If node socket is already closed, do not try to close it again.
+	if g.nodeClientIsNil() {
+		return
+	}
+
+	g.nodeClientLock.Lock()
+	defer g.nodeClientLock.Unlock()
+	g.nodeClient = nil
+	logrus.Info("Destroyed goldmane node client")
+}
+
 func (g *GoldmaneReporter) Report(logSlice any) error {
 	switch logs := logSlice.(type) {
 	case []*flowlog.FlowLog:
 		if logrus.IsLevelEnabled(logrus.DebugLevel) {
 			logrus.WithField("num", len(logs)).Debug("Dispatching flow logs to goldmane")
 		}
 		for _, l := range logs {
-			g.client.Push(convertFlowlogToGoldmane(l))
+			goldmaneLog := convertFlowlogToGoldmane(l)
+			g.client.Push(goldmaneLog)
+
+			if g.mayReportToNodeSocket {
+				// If goldmane node server exists, also send flowlogs to it.
+				g.nodeClientLock.RLock()
+				if g.nodeClient != nil {
+					g.nodeClient.Push(goldmaneLog)
+				}
+				g.nodeClientLock.RUnlock()
+			}
 		}
 	default:
 		logrus.Panic("Unexpected kind of log dispatcher")

felix/collector/goldmane/node_server.go

@@ -0,0 +1,194 @@
+// Copyright (c) 2025 Tigera, Inc. All rights reserved.
+//
+// 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 goldmane
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"path"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+
+	"github.com/projectcalico/calico/goldmane/pkg/server"
+	"github.com/projectcalico/calico/goldmane/pkg/types"
+)
+
+const (
+	NodeSocketDir  = "/var/log/calico/flowlogs"
+	NodeSocketName = "goldmane.sock"
+)
+
+var (
+	NodeSocketPath    = path.Join(NodeSocketDir, NodeSocketName)
+	NodeSocketAddress = fmt.Sprintf("unix://%v", NodeSocketPath)
+)
+
+type flowStore struct {
+	lock  sync.RWMutex
+	flows []*types.Flow
+}
+
+func newFlowStore() *flowStore {
+	return &flowStore{}
+}
+
+func (s *flowStore) Receive(f *types.Flow) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.flows = append(s.flows, f)
+}
+
+func (s *flowStore) List() []*types.Flow {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.flows
+}
+
+func (s *flowStore) Flush() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.flows = nil
+}
+
+func (s *flowStore) ListAndFlush() []*types.Flow {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	flows := s.flows
+	s.flows = nil
+	return flows
+}
+
+type NodeServer struct {
+	store      *flowStore
+	grpcServer *grpc.Server
+	once       sync.Once
+
+	// In Calico node, the address of unix socket is always /var/log/calico/flowlogs/goldmane.sock.
+	// However, NodeServer is also used by Felix FVs, where the code is executed outside of Calico Node. In this case,
+	// unix socket is created in host filesystem, and instead mounted at the mentioned path in Felix containers.
+	dir string
+}
+
+func NewNodeServer(dir string) *NodeServer {
+	nodeServer := NodeServer{
+		dir:        dir,
+		grpcServer: grpc.NewServer(),
+		store:      newFlowStore(),
+	}
+	col := server.NewFlowCollector(nodeServer.store)
+	col.RegisterWith(nodeServer.grpcServer)
+	return &nodeServer
+}
+
+func (s *NodeServer) Run() error {
+	var err error
+	err = ensureNodeSocketDirExists(s.dir)
+	if err != nil {
+		logrus.WithError(err).Error("Failed to create goldmane node socket")
+		return err
+	}
+
+	s.once.Do(func() {
+		var l net.Listener
+		sockAddr := s.Address()
+		l, err = net.Listen("unix", sockAddr)
+		if err != nil {
+			return
+		}
+		logrus.WithField("address", sockAddr).Info("Running goldmane node server")
+		go func() {
+			err = s.grpcServer.Serve(l)
+			if err != nil {
+				return
+			}
+		}()
+	})
+	return nil
+}
+
+func (s *NodeServer) Watch(ctx context.Context, num int, processFlow func(*types.Flow)) {
+	infinitLoop := num < 0
+	var count int
+	logrus.Debug("Starting to watch goldmane node socket")
+	for {
+		if ctx.Err() != nil ||
+			(!infinitLoop && count >= num) {
+			logrus.Debug("Stopped watching goldmane node socket")
+			return
+		}
+
+		flows := s.ListAndFlush()
+		for _, f := range flows {
+			processFlow(f)
+		}
+		count = count + len(flows)
+		time.Sleep(time.Second)
+	}
+}
+
+func (s *NodeServer) Stop() {
+	cleanupNodeSocket(s.Address())
+	s.grpcServer.Stop()
+}
+
+func (s *NodeServer) List() []*types.Flow {
+	return s.store.List()
+}
+
+func (s *NodeServer) Flush() {
+	s.store.Flush()
+}
+
+func (s *NodeServer) ListAndFlush() []*types.Flow {
+	return s.store.ListAndFlush()
+}
+
+func (s *NodeServer) Address() string {
+	return path.Join(s.dir, NodeSocketName)
+}
+
+func ensureNodeSocketDirExists(dir string) error {
+	logrus.Debug("Checking if goldmane node socket exists.")
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		logrus.WithField("directory", dir).Debug("Goldmane node socket directory does not exist")
+		err := os.MkdirAll(dir, 0o600)
+		if err != nil {
+			logrus.WithError(err).WithField("directory", dir).Error("Failed to create node socket directory")
+			return err
+		}
+		logrus.WithField("directory", dir).Debug("Created goldmane node socket directory")
+	}
+	return nil
+}
+
+func cleanupNodeSocket(addr string) {
+	if NodeSocketExists() {
+		err := os.Remove(addr)
+		if err != nil {
+			logrus.WithError(err).WithField("address", addr).Errorf("Failed to remove goldmane node socket")
+		}
+	}
+}
+
+func NodeSocketExists() bool {
+	_, err := os.Stat(NodeSocketPath)
+	// In case of any error, return false
+	return err == nil
+}

felix/fv/flow_logs_goldmane_staged_test.go

@@ -34,6 +34,7 @@ import (
 
 	"github.com/projectcalico/calico/felix/bpf/conntrack"
 	"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"
@@ -100,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"] = flowlogs.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)
@@ -823,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"] = flowlogs.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)
@@ -1572,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"] = flowlogs.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		// Start felix instances.
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)

felix/fv/flow_logs_goldmane_test.go

@@ -32,6 +32,7 @@ import (
 
 	"github.com/projectcalico/calico/felix/bpf/conntrack"
 	"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"
@@ -105,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"] = flowlogs.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 	})
 
 	JustBeforeEach(func() {
@@ -591,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"] = flowlogs.LocalGoldmaneServer
+		opts.ExtraEnvVars["FELIX_FLOWLOGSGOLDMANESERVER"] = goldmane.NodeSocketAddress
 
 		tc, client = infrastructure.StartNNodeTopology(2, opts, infra)