Flowlogs: make calico-node to fetch flow logs from a node (#10144)

Author: mazdakn

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

api/pkg/apis/projectcalico/v3/zz_generated.deepcopy.go

@@ -20,6 +20,7 @@ import (
 	"github.com/projectcalico/calico/felix/calc"
 	"github.com/projectcalico/calico/felix/collector/flowlog"
 	"github.com/projectcalico/calico/felix/collector/goldmane"
+	"github.com/projectcalico/calico/felix/collector/local"
 	"github.com/projectcalico/calico/felix/collector/types"
 	"github.com/projectcalico/calico/felix/config"
 	"github.com/projectcalico/calico/felix/rules"
@@ -29,6 +30,7 @@ import (
 const (
 	// Log dispatcher names
 	FlowLogsGoldmaneReporterName = "goldmane"
+	FlowLogsLocalReporterName    = "socket"
 )
 
 // New creates the required dataplane stats collector, reporters and aggregators.
@@ -72,6 +74,12 @@ func New(
 			dispatchers[FlowLogsGoldmaneReporterName] = gd
 		}
 	}
+	if configParams.FlowLogsLocalReporterEnabled() {
+		log.Infof("Creating local Flow Logs Reporter with address %v", local.SocketAddress)
+		nd := local.NewReporter()
+		dispatchers[FlowLogsLocalReporterName] = nd
+	}
+
 	if len(dispatchers) > 0 {
 		log.Info("Creating Flow Logs Reporter")
 		cw := flowlog.NewReporter(dispatchers, configParams.FlowLogsFlushInterval, healthAggregator)
@@ -84,24 +92,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 := defaultFlowAggregator(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 := defaultFlowAggregator(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 local socket Aggregator for allowed")
+		gaa := defaultFlowAggregator(rules.RuleActionAllow, configParams.FlowLogsCollectorDebugTrace)
+		log.Info("Adding Flow Logs Aggregator (allowed) for local socket")
+		fr.AddAggregator(gaa, []string{FlowLogsLocalReporterName})
+		log.Info("Creating local socket Aggregator for denied")
+		gad := defaultFlowAggregator(rules.RuleActionDeny, configParams.FlowLogsCollectorDebugTrace)
+		log.Info("Adding Flow Logs Aggregator (denied) for local socket")
+		fr.AddAggregator(gad, []string{FlowLogsLocalReporterName})
+	}
+}
+
+func defaultFlowAggregator(forAction rules.RuleAction, traceEnabled bool) *flowlog.Aggregator {
+	return flowlog.NewAggregator().
+		DisplayDebugTraceLogs(traceEnabled).
+		IncludeLabels(true).
+		IncludePolicies(true).
+		IncludeService(true).
+		ForAction(forAction)
 }

api/pkg/openapi/generated.openapi.go

@@ -67,7 +67,7 @@ func (g *GoldmaneReporter) Report(logSlice any) error {
 			logrus.WithField("num", len(logs)).Debug("Dispatching flow logs to goldmane")
 		}
 		for _, l := range logs {
-			g.client.Push(convertFlowlogToGoldmane(l))
+			g.client.Push(ConvertFlowlogToGoldmane(l))
 		}
 	default:
 		logrus.Panic("Unexpected kind of log dispatcher")
@@ -115,7 +115,7 @@ func convertAction(a flowlog.Action) proto.Action {
 	return proto.Action_ActionUnspecified
 }
 
-func convertFlowlogToGoldmane(fl *flowlog.FlowLog) *types.Flow {
+func ConvertFlowlogToGoldmane(fl *flowlog.FlowLog) *types.Flow {
 	return &types.Flow{
 		Key: types.NewFlowKey(
 			&types.FlowKeySource{

felix/collector/dpstatshelper.go

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