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

felix/collector/goldmane/client.go

@@ -17,6 +17,7 @@ package goldmane
 import (
 	"context"
 	"fmt"
+	"os"
 	"sort"
 	"strings"
 	"sync"
@@ -34,10 +35,20 @@ import (
 	"github.com/projectcalico/calico/goldmane/proto"
 )
 
+const (
+	checkNodeSocketTimer = time.Second * 10
+)
+
 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
+	nodeClientCancel      context.CancelFunc
 }
 
 func NewReporter(addr, cert, key, ca string) (*GoldmaneReporter, error) {
@@ -48,6 +59,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: !strings.Contains(addr, NodeSocketPath),
 	}, nil
 }
 
@@ -56,18 +70,82 @@ 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 _, err := os.Stat(NodeSocketPath); err == nil {
+			g.mayStartNodeSocketReporter()
+		} else {
+			g.mayStopNodeSocketReporter()
+		}
+		time.Sleep(checkNodeSocketTimer)
+	}
+}
+
+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")
+	ctx, cancel := context.WithCancel(context.Background())
+	g.nodeClientCancel = cancel
+	g.nodeClient.Connect(ctx)
+}
+
+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.nodeClientCancel()
+	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,197 @@
+// 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 {
+			logrus.WithError(err).Error("Failed to listen on goldmane node socket")
+			return
+		}
+		logrus.WithField("address", sockAddr).Info("Running goldmane node server")
+		go func() {
+			err = s.grpcServer.Serve(l)
+			if err != nil {
+				logrus.WithError(err).Error("Failed to start listening on goldmane node socket")
+				return
+			}
+		}()
+	})
+	return err
+}
+
+func (s *NodeServer) Watch(
+	ctx context.Context,
+	num int,
+	period time.Duration,
+	processFlow func(*types.Flow),
+) {
+	infiniteLoop := num < 0
+	var count int
+	logrus.Debug("Starting to watch goldmane node socket")
+	for {
+		if ctx.Err() != nil ||
+			(!infiniteLoop && 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(period)
+	}
+}
+
+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) {
+	_, err := os.Stat(addr)
+	if err != nil && os.IsNotExist(err) {
+		return
+	}
+	err = os.Remove(addr)
+	if err != nil {
+		logrus.WithError(err).WithField("address", addr).Errorf("Failed to remove goldmane node socket")
+	}
+}