Skip to content

Commit

Permalink
add pod store for flow-exporter and flow-aggregator
Browse files Browse the repository at this point in the history 
1. Add pod store which can be used to store current and previous pod information.
2. Modify unit test

Signed-off-by: Yun-Tang Hsu <[email protected]>
  • Loading branch information
@yuntanghsu
yuntanghsu committed Jul 18, 2023
1 parent c372130 commit acc0be2
Show file tree
Hide file tree
Showing 15 changed files with 777 additions and 269 deletions.
7 changes: 5 additions & 2 deletions cmd/antrea-agent/agent.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,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"
)

Expand Down Expand Up @@ -312,7 +313,8 @@ func run(o *Options) error {
var localPodInformer cache.SharedIndexInformer
if enableNodePortLocal || enableBridgingMode || enableMulticlusterNP ||
features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) ||
features.DefaultFeatureGate.Enabled(features.TrafficControl) {
features.DefaultFeatureGate.Enabled(features.TrafficControl) ||
(features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable) {
listOptions := func(options *metav1.ListOptions) {
options.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", nodeConfig.Name).String()
}
Expand Down Expand Up @@ -595,6 +597,7 @@ func run(o *Options) error {

var flowExporter *exporter.FlowExporter
if features.DefaultFeatureGate.Enabled(features.FlowExporter) && o.config.FlowExporter.Enable {
podStore := podstore.NewPodStorage(localPodInformer)
flowExporterOptions := &flowexporter.FlowExporterOptions{
FlowCollectorAddr: o.flowCollectorAddr,
FlowCollectorProto: o.flowCollectorProto,
Expand All @@ -604,7 +607,7 @@ func run(o *Options) error {
PollInterval: o.pollInterval,
ConnectUplinkToBridge: connectUplinkToBridge}
flowExporter, err = exporter.NewFlowExporter(
ifaceStore,
podStore,
proxier,
k8sClient,
nodeRouteController,
Expand Down
21 changes: 15 additions & 6 deletions cmd/flow-aggregator/flow-aggregator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,19 @@ import (
"sync"
"time"

"k8s.io/client-go/informers"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
coreinformers "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
"k8s.io/klog/v2"

aggregator "antrea.io/antrea/pkg/flowaggregator"
"antrea.io/antrea/pkg/flowaggregator/apiserver"
"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
Expand All @@ -47,12 +50,19 @@ func run(configFile string) error {
return fmt.Errorf("error when creating K8s client: %v", err)
}

informerFactory := informers.NewSharedInformerFactory(k8sClient, informerDefaultResync)
podInformer := informerFactory.Core().V1().Pods()
podInformer := coreinformers.NewFilteredPodInformer(
k8sClient,
metav1.NamespaceAll,
informerDefaultResync,
cache.Indexers{},
nil,
)

podStore := podstore.NewPodStorage(podInformer)

flowAggregator, err := aggregator.NewFlowAggregator(
k8sClient,
podInformer,
podStore,
configFile,
)

Expand All @@ -79,8 +89,7 @@ func run(configFile string) error {
return fmt.Errorf("error when creating flow aggregator API server: %v", err)
}
go apiServer.Run(stopCh)

informerFactory.Start(stopCh)
go podInformer.Run(stopCh)

<-stopCh
klog.InfoS("Stopping flow aggregator")
Expand Down
28 changes: 14 additions & 14 deletions pkg/agent/flowexporter/connections/connections.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -34,20 +34,20 @@ const (

type connectionStore struct {
connections map[flowexporter.ConnectionKey]*flowexporter.Connection
ifaceStore interfacestore.InterfaceStore
podStore *podstore.PodStorage
antreaProxier proxy.Proxier
expirePriorityQueue *priorityqueue.ExpirePriorityQueue
staleConnectionTimeout time.Duration
mutex sync.Mutex
}

func NewConnectionStore(
ifaceStore interfacestore.InterfaceStore,
podStore *podstore.PodStorage,
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,
Expand Down Expand Up @@ -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
}
}

Expand Down
23 changes: 16 additions & 7 deletions pkg/agent/flowexporter/connections/connections_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,18 +21,24 @@ import (

"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
coreinformers "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/tools/cache"

"k8s.io/client-go/kubernetes/fake"

"antrea.io/antrea/pkg/agent/flowexporter"
connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
"antrea.io/antrea/pkg/agent/metrics"
"antrea.io/antrea/pkg/util/podstore"
)

const (
testActiveFlowTimeout = 3 * time.Second
testIdleFlowTimeout = 1 * time.Second
testPollInterval = 0 // Not used in these tests, hence 0.
testStaleConnectionTimeout = 5 * time.Minute
informerDefaultResync = 12 * time.Hour
)

var testFlowExporterOptions = &flowexporter.FlowExporterOptions{
Expand All @@ -45,7 +51,6 @@ var testFlowExporterOptions = &flowexporter.FlowExporterOptions{
}

func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
ctrl := gomock.NewController(t)
// Create two flows; one is already in connectionStore and other one is new
testFlows := make([]*flowexporter.Connection, 2)
testFlowKeys := make([]*flowexporter.ConnectionKey, 2)
Expand Down Expand Up @@ -79,8 +84,10 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
testFlowKeys[i] = &connKey
}
// Create connectionStore
mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
connStore := NewConnectionStore(mockIfaceStore, nil, testFlowExporterOptions)
k8sClient := fake.NewSimpleClientset()
podInformer := coreinformers.NewFilteredPodInformer(k8sClient, metav1.NamespaceAll, informerDefaultResync, cache.Indexers{}, nil)
podStore := podstore.NewPodStorage(podInformer)
connStore := NewConnectionStore(podStore, nil, testFlowExporterOptions)
// Add flows to the Connection store
for i, flow := range testFlows {
connStore.connections[*testFlowKeys[i]] = flow
Expand All @@ -105,8 +112,10 @@ 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)
k8sClient := fake.NewSimpleClientset()
podInformer := coreinformers.NewFilteredPodInformer(k8sClient, metav1.NamespaceAll, informerDefaultResync, cache.Indexers{}, nil)
podStore := podstore.NewPodStorage(podInformer)
denyConnStore := NewDenyConnectionStore(podStore, 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,
Expand All @@ -123,7 +132,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, podStore, nil, testFlowExporterOptions)
conntrackConnStore.connections[connKey] = conn

metrics.TotalAntreaConnectionsInConnTrackTable.Set(1)
Expand Down
6 changes: 3 additions & 3 deletions pkg/agent/flowexporter/connections/conntrack_connections.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand All @@ -53,7 +53,7 @@ func NewConntrackConnectionStore(
v4Enabled bool,
v6Enabled bool,
npQuerier querier.AgentNetworkPolicyInfoQuerier,
ifaceStore interfacestore.InterfaceStore,
podStore *podstore.PodStorage,
proxier proxy.Proxier,
o *flowexporter.FlowExporterOptions,
) *ConntrackConnectionStore {
Expand All @@ -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,
}
}
Expand Down
39 changes: 27 additions & 12 deletions pkg/agent/flowexporter/connections/conntrack_connections_perf_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,16 +28,19 @@ 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"
coreinformers "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/tools/cache"
"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"
"antrea.io/antrea/pkg/util/podstore"
k8sproxy "antrea.io/antrea/third_party/proxy"
)

Expand All @@ -62,8 +65,8 @@ Sample output (10000 init connections, 1000 new connections, 1000 deleted connec

func BenchmarkPoll(b *testing.B) {
disableLogToStderr()
connStore, mockConnDumper := setupConntrackConnStore(b)
conns := generateConns()
connStore, mockConnDumper := setupConntrackConnStore(b, conns)
b.ResetTimer()
for n := 0; n < b.N; n++ {
mockConnDumper.EXPECT().DumpFlows(uint16(openflow.CtZone)).Return(conns, testNumOfConns, nil)
Expand All @@ -75,18 +78,30 @@ func BenchmarkPoll(b *testing.B) {
b.Logf("\nSummary:\nNumber of initial connections: %d\nNumber of new connections/poll: %d\nNumber of deleted connections/poll: %d\n", testNumOfConns, testNumOfNewConns, testNumOfDeletedConns)
}

func setupConntrackConnStore(b *testing.B) (*ConntrackConnectionStore, *connectionstest.MockConnTrackDumper) {
func setupConntrackConnStore(b *testing.B, connections []*flowexporter.Connection) (*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",
k8sClient := fake.NewSimpleClientset()
podInformer := coreinformers.NewFilteredPodInformer(k8sClient, metav1.NamespaceAll, informerDefaultResync, cache.Indexers{}, nil)
podStore := podstore.NewPodStorage(podInformer)
stopCh := make(chan struct{})
defer close(stopCh)
go podInformer.Run(stopCh)
cache.WaitForCacheSync(stopCh, podInformer.HasSynced)
pod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: "pod-ns",
Name: "pod",
CreationTimestamp: metav1.Time{Time: time.Now().Add(-255 * time.Second)},
},
Status: v1.PodStatus{
Phase: v1.PodPending,
},
}
for _, conn := range connections {
pod.Status.PodIPs = append(pod.Status.PodIPs, v1.PodIP{IP: conn.FlowKey.SourceAddress.String()}, v1.PodIP{IP: conn.FlowKey.DestinationAddress.String()})
}
mockIfaceStore.EXPECT().GetInterfaceByIP(gomock.Any()).Return(testInterface, true).AnyTimes()
podInformer.GetIndexer().Add(pod)

mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
mockConnDumper.EXPECT().GetMaxConnections().Return(100000, nil).AnyTimes()
Expand All @@ -104,7 +119,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, podStore, nil, testFlowExporterOptions), mockConnDumper
}

func generateConns() []*flowexporter.Connection {
Expand Down
Loading

0 comments on commit acc0be2

Please sign in to comment.