Add Pod store for Flow Exporter and Flow Aggregator

1. Add Pod store which can be used to store current and previous Pods information.
2. Modify unit test

Signed-off-by: Yun-Tang Hsu <[email protected]>

cmd/antrea-agent/agent.go

@@ -77,6 +77,7 @@ import (
 	"antrea.io/antrea/pkg/signals"
 	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/k8s"
+	"antrea.io/antrea/pkg/util/podstore"
 	"antrea.io/antrea/pkg/version"
 )
 
@@ -139,6 +140,7 @@ func run(o *Options) error {
 	l7NetworkPolicyEnabled := features.DefaultFeatureGate.Enabled(features.L7NetworkPolicy)
 	enableMulticlusterGW := features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.EnableGateway
 	enableMulticlusterNP := features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.EnableStretchedNetworkPolicy
+	enableFLowExporter := features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable
 
 	nodeIPTracker := nodeip.NewTracker(nodeInformer)
 	// Bridging mode will connect the uplink interface to the OVS bridge.
@@ -156,7 +158,7 @@ func run(o *Options) error {
 		features.DefaultFeatureGate.Enabled(features.AntreaPolicy),
 		l7NetworkPolicyEnabled,
 		o.enableEgress,
-		features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable,
+		enableFLowExporter,
 		o.config.AntreaProxy.ProxyAll,
 		features.DefaultFeatureGate.Enabled(features.LoadBalancerModeDSR),
 		connectUplinkToBridge,
@@ -317,7 +319,7 @@ func run(o *Options) error {
 	// Initialize localPodInformer for NPLAgent, AntreaIPAMController,
 	// StretchedNetworkPolicyController, and secondary network controller.
 	var localPodInformer cache.SharedIndexInformer
-	if enableNodePortLocal || enableBridgingMode || enableMulticlusterNP ||
+	if enableNodePortLocal || enableBridgingMode || enableMulticlusterNP || enableFLowExporter ||
 		features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) ||
 		features.DefaultFeatureGate.Enabled(features.TrafficControl) {
 		listOptions := func(options *metav1.ListOptions) {
@@ -602,7 +604,8 @@ func run(o *Options) error {
 	}
 
 	var flowExporter *exporter.FlowExporter
-	if features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable {
+	if enableFLowExporter {
+		podStore := podstore.NewPodStore(localPodInformer)
 		flowExporterOptions := &flowexporter.FlowExporterOptions{
 			FlowCollectorAddr:      o.flowCollectorAddr,
 			FlowCollectorProto:     o.flowCollectorProto,
@@ -612,7 +615,7 @@ func run(o *Options) error {
 			PollInterval:           o.pollInterval,
 			ConnectUplinkToBridge:  connectUplinkToBridge}
 		flowExporter, err = exporter.NewFlowExporter(
-			ifaceStore,
+			podStore,
 			proxier,
 			k8sClient,
 			nodeRouteController,
@@ -877,7 +880,7 @@ func run(o *Options) error {
 	go ofClient.StartPacketInHandler(stopCh)
 
 	// Start the goroutine to periodically export IPFIX flow records.
-	if features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable {
+	if enableFLowExporter {
 		go flowExporter.Run(stopCh)
 	}
 

cmd/flow-aggregator/flow-aggregator.go

@@ -29,6 +29,7 @@ import (
 	"antrea.io/antrea/pkg/log"
 	"antrea.io/antrea/pkg/signals"
 	"antrea.io/antrea/pkg/util/cipher"
+	"antrea.io/antrea/pkg/util/podstore"
 )
 
 const informerDefaultResync = 12 * time.Hour
@@ -49,10 +50,11 @@ func run(configFile string) error {
 
 	informerFactory := informers.NewSharedInformerFactory(k8sClient, informerDefaultResync)
 	podInformer := informerFactory.Core().V1().Pods()
+	podStore := podstore.NewPodStore(podInformer.Informer())
 
 	flowAggregator, err := aggregator.NewFlowAggregator(
 		k8sClient,
-		podInformer,
+		podStore,
 		configFile,
 	)
 

hack/update-codegen-dockerized.sh

@@ -70,6 +70,7 @@ MOCKGEN_TARGETS=(
   "pkg/querier AgentNetworkPolicyInfoQuerier,AgentMulticastInfoQuerier,EgressQuerier testing"
   "pkg/flowaggregator/querier FlowAggregatorQuerier testing"
   "pkg/flowaggregator/s3uploader S3UploaderAPI testing"
+  "pkg/util/podstore Interface testing"
   "third_party/proxy Provider testing"
 )
 

pkg/agent/flowexporter/connections/connections.go

@@ -24,8 +24,8 @@ import (
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	"antrea.io/antrea/pkg/agent/flowexporter/priorityqueue"
-	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/proxy"
+	"antrea.io/antrea/pkg/util/podstore"
 )
 
 const (
@@ -34,20 +34,20 @@ const (
 
 type connectionStore struct {
 	connections            map[flowexporter.ConnectionKey]*flowexporter.Connection
-	ifaceStore             interfacestore.InterfaceStore
+	podStore               podstore.Interface
 	antreaProxier          proxy.Proxier
 	expirePriorityQueue    *priorityqueue.ExpirePriorityQueue
 	staleConnectionTimeout time.Duration
 	mutex                  sync.Mutex
 }
 
 func NewConnectionStore(
-	ifaceStore interfacestore.InterfaceStore,
+	podStore podstore.Interface,
 	proxier proxy.Proxier,
 	o *flowexporter.FlowExporterOptions) connectionStore {
 	return connectionStore{
 		connections:            make(map[flowexporter.ConnectionKey]*flowexporter.Connection),
-		ifaceStore:             ifaceStore,
+		podStore:               podStore,
 		antreaProxier:          proxier,
 		expirePriorityQueue:    priorityqueue.NewExpirePriorityQueue(o.ActiveFlowTimeout, o.IdleFlowTimeout),
 		staleConnectionTimeout: o.StaleConnectionTimeout,
@@ -98,26 +98,26 @@ func (cs *connectionStore) AddConnToMap(connKey *flowexporter.ConnectionKey, con
 }
 
 func (cs *connectionStore) fillPodInfo(conn *flowexporter.Connection) {
-	if cs.ifaceStore == nil {
-		klog.V(4).Info("Interface store is not available to retrieve local Pods information.")
+	if cs.podStore == nil {
+		klog.V(4).Info("Pod store is not available to retrieve local Pods information.")
 		return
 	}
 	// sourceIP/destinationIP are mapped only to local pods and not remote pods.
 	srcIP := conn.FlowKey.SourceAddress.String()
 	dstIP := conn.FlowKey.DestinationAddress.String()
 
-	sIface, srcFound := cs.ifaceStore.GetInterfaceByIP(srcIP)
-	dIface, dstFound := cs.ifaceStore.GetInterfaceByIP(dstIP)
+	srcPod, srcFound := cs.podStore.GetPodByIPAndTime(srcIP, conn.StartTime)
+	dstPod, dstFound := cs.podStore.GetPodByIPAndTime(dstIP, conn.StartTime)
 	if !srcFound && !dstFound {
 		klog.Warningf("Cannot map any of the IP %s or %s to a local Pod", srcIP, dstIP)
 	}
-	if srcFound && sIface.Type == interfacestore.ContainerInterface {
-		conn.SourcePodName = sIface.ContainerInterfaceConfig.PodName
-		conn.SourcePodNamespace = sIface.ContainerInterfaceConfig.PodNamespace
+	if srcFound {
+		conn.SourcePodName = srcPod.Name
+		conn.SourcePodNamespace = srcPod.Namespace
 	}
-	if dstFound && dIface.Type == interfacestore.ContainerInterface {
-		conn.DestinationPodName = dIface.ContainerInterfaceConfig.PodName
-		conn.DestinationPodNamespace = dIface.ContainerInterfaceConfig.PodNamespace
+	if dstFound {
+		conn.DestinationPodName = dstPod.Name
+		conn.DestinationPodNamespace = dstPod.Namespace
 	}
 }
 

pkg/agent/flowexporter/connections/connections_test.go

@@ -24,7 +24,8 @@ import (
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
-	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
+	podstoretest "antrea.io/antrea/pkg/util/podstore/testing"
+
 	"antrea.io/antrea/pkg/agent/metrics"
 )
 
@@ -79,8 +80,8 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 		testFlowKeys[i] = &connKey
 	}
 	// Create connectionStore
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	connStore := NewConnectionStore(mockIfaceStore, nil, testFlowExporterOptions)
+	mockPodStore := podstoretest.NewMockInterface(ctrl)
+	connStore := NewConnectionStore(mockPodStore, nil, testFlowExporterOptions)
 	// Add flows to the Connection store
 	for i, flow := range testFlows {
 		connStore.connections[*testFlowKeys[i]] = flow
@@ -105,8 +106,8 @@ func TestConnectionStore_DeleteConnWithoutLock(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	metrics.InitializeConnectionMetrics()
 	// test on deny connection store
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	denyConnStore := NewDenyConnectionStore(mockIfaceStore, nil, testFlowExporterOptions)
+	mockPodStore := podstoretest.NewMockInterface(ctrl)
+	denyConnStore := NewDenyConnectionStore(mockPodStore, nil, testFlowExporterOptions)
 	tuple := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
 	conn := &flowexporter.Connection{
 		FlowKey: tuple,
@@ -123,7 +124,7 @@ func TestConnectionStore_DeleteConnWithoutLock(t *testing.T) {
 
 	// test on conntrack connection store
 	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
-	conntrackConnStore := NewConntrackConnectionStore(mockConnDumper, true, false, nil, mockIfaceStore, nil, testFlowExporterOptions)
+	conntrackConnStore := NewConntrackConnectionStore(mockConnDumper, true, false, nil, mockPodStore, nil, testFlowExporterOptions)
 	conntrackConnStore.connections[connKey] = conn
 
 	metrics.TotalAntreaConnectionsInConnTrackTable.Set(1)

pkg/agent/flowexporter/connections/conntrack_connections.go

@@ -25,11 +25,11 @@ import (
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	"antrea.io/antrea/pkg/agent/flowexporter/priorityqueue"
-	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/metrics"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/proxy"
 	"antrea.io/antrea/pkg/querier"
+	"antrea.io/antrea/pkg/util/podstore"
 )
 
 var serviceProtocolMap = map[uint8]corev1.Protocol{
@@ -53,7 +53,7 @@ func NewConntrackConnectionStore(
 	v4Enabled bool,
 	v6Enabled bool,
 	npQuerier querier.AgentNetworkPolicyInfoQuerier,
-	ifaceStore interfacestore.InterfaceStore,
+	podStore podstore.Interface,
 	proxier proxy.Proxier,
 	o *flowexporter.FlowExporterOptions,
 ) *ConntrackConnectionStore {
@@ -63,7 +63,7 @@ func NewConntrackConnectionStore(
 		v6Enabled:             v6Enabled,
 		networkPolicyQuerier:  npQuerier,
 		pollInterval:          o.PollInterval,
-		connectionStore:       NewConnectionStore(ifaceStore, proxier, o),
+		connectionStore:       NewConnectionStore(podStore, proxier, o),
 		connectUplinkToBridge: o.ConnectUplinkToBridge,
 	}
 }

pkg/agent/flowexporter/connections/conntrack_connections_perf_test.go

@@ -28,16 +28,16 @@ import (
 
 	"github.com/golang/mock/gomock"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
-	"antrea.io/antrea/pkg/agent/interfacestore"
-	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
 	"antrea.io/antrea/pkg/agent/openflow"
 	proxytest "antrea.io/antrea/pkg/agent/proxy/testing"
 	queriertest "antrea.io/antrea/pkg/querier/testing"
+	podstoretest "antrea.io/antrea/pkg/util/podstore/testing"
 	k8sproxy "antrea.io/antrea/third_party/proxy"
 )
 
@@ -78,15 +78,18 @@ func BenchmarkPoll(b *testing.B) {
 func setupConntrackConnStore(b *testing.B) (*ConntrackConnectionStore, *connectionstest.MockConnTrackDumper) {
 	ctrl := gomock.NewController(b)
 	defer ctrl.Finish()
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	testInterface := &interfacestore.InterfaceConfig{
-		Type: interfacestore.ContainerInterface,
-		ContainerInterfaceConfig: &interfacestore.ContainerInterfaceConfig{
-			PodName:      "pod",
-			PodNamespace: "pod-ns",
+	mockPodStore := podstoretest.NewMockInterface(ctrl)
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "pod-ns",
+			Name:      "pod",
+			UID:       "pod",
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodPending,
 		},
 	}
-	mockIfaceStore.EXPECT().GetInterfaceByIP(gomock.Any()).Return(testInterface, true).AnyTimes()
+	mockPodStore.EXPECT().GetPodByIPAndTime(gomock.Any(), gomock.Any()).Return(pod, true).AnyTimes()
 
 	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
 	mockConnDumper.EXPECT().GetMaxConnections().Return(100000, nil).AnyTimes()
@@ -104,7 +107,7 @@ func setupConntrackConnStore(b *testing.B) (*ConntrackConnectionStore, *connecti
 	mockProxier.EXPECT().GetServiceByIP(serviceStr).Return(servicePortName, true).AnyTimes()
 
 	npQuerier := queriertest.NewMockAgentNetworkPolicyInfoQuerier(ctrl)
-	return NewConntrackConnectionStore(mockConnDumper, true, false, npQuerier, mockIfaceStore, nil, testFlowExporterOptions), mockConnDumper
+	return NewConntrackConnectionStore(mockConnDumper, true, false, npQuerier, mockPodStore, nil, testFlowExporterOptions), mockConnDumper
 }
 
 func generateConns() []*flowexporter.Connection {