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

api/pkg/apis/projectcalico/v3/felixconfig.go

@@ -835,6 +835,10 @@ type FelixConfigurationSpec struct {
 	// FlowLogGoldmaneServer is the flow server endpoint to which flow data should be published.
 	FlowLogsGoldmaneServer *string `json:"flowLogsGoldmaneServer,omitempty"`
 
+	// FlowLogsLocalReporter configures local unix socket for reporting flow data from each node. [Default: Disabled]
+	// +kubebuilder:validation:Enum=Disabled;Enabled
+	FlowLogsLocalReporter *string `json:"flowLogsLocalReporter,omitempty"`
+
 	// BPFProfiling controls profiling of BPF programs. At the monent, it can be
 	// Disabled or Enabled. [Default: Disabled]
 	//+kubebuilder:validation:Enum=Enabled;Disabled

felix/collector/dpstatshelper.go

@@ -29,6 +29,7 @@ import (
 const (
 	// Log dispatcher names
 	FlowLogsGoldmaneReporterName = "goldmane"
+	FlowLogsLocalReporterName    = "node socket"
 )
 
 // New creates the required dataplane stats collector, reporters and aggregators.
@@ -72,6 +73,12 @@ func New(
 			dispatchers[FlowLogsGoldmaneReporterName] = gd
 		}
 	}
+	if configParams.FlowLogsLocalReporterEnabled() {
+		log.Infof("Creating local Flow Logs Reporter with address %v", goldmane.NodeSocketAddress)
+		nd := goldmane.NewNodeSocketReporter()
+		dispatchers[FlowLogsLocalReporterName] = nd
+	}
+
 	if len(dispatchers) > 0 {
 		log.Info("Creating Flow Logs Reporter")
 		cw := flowlog.NewReporter(dispatchers, configParams.FlowLogsFlushInterval, healthAggregator)
@@ -84,24 +91,35 @@ func New(
 
 // configureFlowAggregation adds appropriate aggregators to the FlowLogReporter, depending on configuration.
 func configureFlowAggregation(configParams *config.Config, fr *flowlog.FlowLogReporter) {
+	// Set up aggregator for goldmane reporter.
 	if configParams.FlowLogsGoldmaneServer != "" {
-		log.Info("Creating golemane Aggregator for allowed")
-		gaa := flowlog.NewAggregator().
-			DisplayDebugTraceLogs(configParams.FlowLogsCollectorDebugTrace).
-			IncludeLabels(true).
-			IncludePolicies(true).
-			IncludeService(true).
-			ForAction(rules.RuleActionAllow)
+		log.Info("Creating goldmane Aggregator for allowed")
+		gaa := flowAggregatorForGoldmane(rules.RuleActionAllow, configParams.FlowLogsCollectorDebugTrace)
 		log.Info("Adding Flow Logs Aggregator (allowed) for goldmane")
 		fr.AddAggregator(gaa, []string{FlowLogsGoldmaneReporterName})
 		log.Info("Creating goldmane Aggregator for denied")
-		gad := flowlog.NewAggregator().
-			DisplayDebugTraceLogs(configParams.FlowLogsCollectorDebugTrace).
-			IncludeLabels(true).
-			IncludePolicies(true).
-			IncludeService(true).
-			ForAction(rules.RuleActionDeny)
+		gad := flowAggregatorForGoldmane(rules.RuleActionDeny, configParams.FlowLogsCollectorDebugTrace)
 		log.Info("Adding Flow Logs Aggregator (denied) for goldmane")
 		fr.AddAggregator(gad, []string{FlowLogsGoldmaneReporterName})
 	}
+	// Set up aggregator for local socket reporter.
+	if configParams.FlowLogsLocalReporterEnabled() {
+		log.Info("Creating node socket Aggregator for allowed")
+		gaa := flowAggregatorForGoldmane(rules.RuleActionAllow, configParams.FlowLogsCollectorDebugTrace)
+		log.Info("Adding Flow Logs Aggregator (allowed) for node socket")
+		fr.AddAggregator(gaa, []string{FlowLogsLocalReporterName})
+		log.Info("Creating node socket Aggregator for denied")
+		gad := flowAggregatorForGoldmane(rules.RuleActionDeny, configParams.FlowLogsCollectorDebugTrace)
+		log.Info("Adding Flow Logs Aggregator (denied) for node socket")
+		fr.AddAggregator(gad, []string{FlowLogsLocalReporterName})
+	}
+}
+
+func flowAggregatorForGoldmane(forAction rules.RuleAction, traceEnabled bool) *flowlog.Aggregator {
+	return flowlog.NewAggregator().
+		DisplayDebugTraceLogs(traceEnabled).
+		IncludeLabels(true).
+		IncludePolicies(true).
+		IncludeService(true).
+		ForAction(forAction)
 }

felix/collector/goldmane/node_reporter.go

@@ -0,0 +1,119 @@
+// 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"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/projectcalico/calico/felix/collector/flowlog"
+	"github.com/projectcalico/calico/goldmane/pkg/client"
+)
+
+const (
+	checkNodeSocketTimer = time.Second * 10
+)
+
+type NodeSocketReporter struct {
+	client       *client.FlowClient
+	clientLock   sync.RWMutex
+	clientCancel context.CancelFunc
+	once         sync.Once
+}
+
+func NewNodeSocketReporter() *NodeSocketReporter {
+	return &NodeSocketReporter{}
+}
+
+func (n *NodeSocketReporter) Start() error {
+	var err error
+	n.once.Do(func() {
+		go n.nodeSocketReporter()
+	})
+	return err
+}
+
+func (n *NodeSocketReporter) nodeSocketReporter() {
+	for {
+		if _, err := os.Stat(NodeSocketPath); err == nil {
+			n.mayStartNodeSocketReporter()
+		} else {
+			n.mayStopNodeSocketReporter()
+		}
+		time.Sleep(checkNodeSocketTimer)
+	}
+}
+
+func (n *NodeSocketReporter) nodeClientIsNil() bool {
+	n.clientLock.RLock()
+	defer n.clientLock.RUnlock()
+	return n.client == nil
+}
+
+func (n *NodeSocketReporter) mayStartNodeSocketReporter() {
+	// If node socket is already setup, do not try to set it up again.
+	if !n.nodeClientIsNil() {
+		return
+	}
+
+	var err error
+	n.clientLock.Lock()
+	defer n.clientLock.Unlock()
+	n.client, err = client.NewFlowClient(NodeSocketAddress, "", "", "")
+	if err != nil {
+		logrus.WithError(err).Warn("Failed to create goldmane node socket client")
+		return
+	}
+	logrus.Info("Created node socket client")
+	ctx, cancel := context.WithCancel(context.Background())
+	n.clientCancel = cancel
+	n.client.Connect(ctx)
+}
+
+func (n *NodeSocketReporter) mayStopNodeSocketReporter() {
+	// If node socket is already closed, do not try to close it again.
+	if n.nodeClientIsNil() {
+		return
+	}
+
+	n.clientLock.Lock()
+	defer n.clientLock.Unlock()
+	n.clientCancel()
+	n.client = nil
+	logrus.Info("Destroyed node socket client")
+}
+
+func (n *NodeSocketReporter) 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 node socket")
+		}
+		for _, l := range logs {
+			n.clientLock.RLock()
+			if n.client != nil {
+				n.client.Push(convertFlowlogToGoldmane(l))
+			}
+			n.clientLock.RUnlock()
+		}
+	default:
+		logrus.Panic("Unexpected kind of log dispatcher")
+	}
+	return nil
+}