Support antctl in the flow aggregator

Signed-off-by: Yanjun Zhou <[email protected]>

build/images/flow-aggregator/Dockerfile

@@ -1,4 +1,5 @@
 ARG GO_VERSION
+ARG OVS_VERSION
 FROM golang:${GO_VERSION} as flow-aggregator-build
 
 WORKDIR /antrea
@@ -7,14 +8,16 @@ COPY . /antrea
 
 # Make sure the flow-aggregator binary is statically linked.
 RUN CGO_ENABLED=0 make flow-aggregator
+RUN make antctl-ubuntu
 
-FROM scratch
+FROM antrea/base-ubuntu:${OVS_VERSION}
 
 LABEL maintainer="Antrea <[email protected]>"
 LABEL description="The docker image for the flow aggregator"
 
 ENV USER root
 
 COPY --from=flow-aggregator-build /antrea/bin/flow-aggregator /
+COPY --from=flow-aggregator-build /antrea/bin/antctl /usr/local/bin/
 
 ENTRYPOINT ["/flow-aggregator"]

build/yamls/flow-aggregator.yml

@@ -68,6 +68,9 @@ rules:
   - configmaps
   verbs:
   - create
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resourceNames:
@@ -169,6 +172,20 @@ data:
     # Provide format for records sent to the configured flow collector.
     # Supported formats are IPFIX and JSON.
     #recordFormat: "IPFIX"
+    
+    # The port for the flow-aggregator APIServer to serve on.
+    # Note that if it's set to another value, the `containerPort` of the `api` port of the
+    # `flow-aggregator` container must be set to the same value.
+    #apiPort: 10350
+
+    # Comma-separated list of Cipher Suites. If omitted, the default Go Cipher Suites will be used.
+    # https://golang.org/pkg/crypto/tls/#pkg-constants
+    # Note that TLS1.3 Cipher Suites cannot be added to the list. But the apiserver will always
+    # prefer TLS1.3 Cipher Suites whenever possible.
+    #tlsCipherSuites:
+
+    # TLS min version from: VersionTLS10, VersionTLS11, VersionTLS12, VersionTLS13.
+    #tlsMinVersion:
 kind: ConfigMap
 metadata:
   annotations: {}
@@ -224,6 +241,11 @@ spec:
         - --log_file_max_size=100
         - --log_file_max_num=4
         - --v=0
+        env:
+        - name: POD_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
         image: projects.registry.vmware.com/antrea/flow-aggregator:latest
         imagePullPolicy: IfNotPresent
         name: flow-aggregator

build/yamls/flow-aggregator/base/conf/flow-aggregator.conf

@@ -34,3 +34,17 @@
 # Provide format for records sent to the configured flow collector.
 # Supported formats are IPFIX and JSON.
 #recordFormat: "IPFIX"
+
+# The port for the flow-aggregator APIServer to serve on.
+# Note that if it's set to another value, the `containerPort` of the `api` port of the
+# `flow-aggregator` container must be set to the same value.
+#apiPort: 10350
+
+# Comma-separated list of Cipher Suites. If omitted, the default Go Cipher Suites will be used.
+# https://golang.org/pkg/crypto/tls/#pkg-constants
+# Note that TLS1.3 Cipher Suites cannot be added to the list. But the apiserver will always
+# prefer TLS1.3 Cipher Suites whenever possible.
+#tlsCipherSuites:
+
+# TLS min version from: VersionTLS10, VersionTLS11, VersionTLS12, VersionTLS13.
+#tlsMinVersion:

build/yamls/flow-aggregator/base/flow-aggregator.yml

@@ -24,7 +24,7 @@ rules:
     verbs: ["get", "list", "watch"]
   - apiGroups: [""]
     resources: ["configmaps"]
-    verbs: ["create"]
+    verbs: ["create", "get", "list", "watch"]
   - apiGroups: [""]
     resources: ["secrets"]
     resourceNames: ["flow-aggregator-client-tls"]
@@ -130,6 +130,11 @@ spec:
         - --v=0
         name: flow-aggregator
         image: flow-aggregator
+        env:
+          - name: POD_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.name
         ports:
           - containerPort: 4739
         volumeMounts:

cmd/flow-aggregator/config.go

@@ -51,4 +51,11 @@ type FlowAggregatorConfig struct {
 	// Provide format for records sent to the configured flow collector. Supported formats are IPFIX and JSON.
 	// Defaults to "IPFIX"
 	RecordFormat string `yaml:"recordFormat,omitempty"`
+	// APIPort is the port for the antrea-agent APIServer to serve on.
+	// Defaults to 10350.
+	APIPort int `yaml:"apiPort,omitempty"`
+	// Cipher suites to use.
+	TLSCipherSuites string `yaml:"tlsCipherSuites,omitempty"`
+	// TLS min version.
+	TLSMinVersion string `yaml:"tlsMinVersion,omitempty"`
 }

cmd/flow-aggregator/flow-aggregator.go

@@ -29,8 +29,10 @@ import (
 
 	"antrea.io/antrea/pkg/clusteridentity"
 	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"
 )
 
 const informerDefaultResync = 12 * time.Hour
@@ -128,6 +130,20 @@ func run(o *Options) error {
 	wg.Add(1)
 	go flowAggregator.Run(stopCh, &wg)
 
+	cipherSuites, err := cipher.GenerateCipherSuitesList(o.config.TLSCipherSuites)
+	if err != nil {
+		return fmt.Errorf("error generating Cipher Suite list: %v", err)
+	}
+	apiServer, err := apiserver.New(
+		flowAggregator,
+		o.config.APIPort,
+		cipherSuites,
+		cipher.TLSVersionMap[o.config.TLSMinVersion])
+	if err != nil {
+		return fmt.Errorf("error when creating flow aggregator API server: %v", err)
+	}
+	go apiServer.Run(stopCh)
+
 	informerFactory.Start(stopCh)
 
 	<-stopCh

cmd/flow-aggregator/options.go

@@ -24,6 +24,7 @@ import (
 	"github.com/spf13/pflag"
 	"gopkg.in/yaml.v2"
 
+	"antrea.io/antrea/pkg/apis"
 	"antrea.io/antrea/pkg/flowaggregator"
 	"antrea.io/antrea/pkg/util/flowexport"
 )
@@ -133,6 +134,9 @@ func (o *Options) validate(args []string) error {
 	} else {
 		o.format = o.config.RecordFormat
 	}
+	if o.config.APIPort == 0 {
+		o.config.APIPort = apis.AntreaAgentAPIPort
+	}
 	return nil
 }
 

docs/antctl.md

@@ -10,6 +10,9 @@ running in two different modes:
   the Antrea Agent and query information local to that Agent (e.g. the set of
   computed NetworkPolicies received by that Agent from the Antrea Controller, as
   opposed to the entire set of computed policies).
+* "flowaggregator mode": when run from within a Flow Aggregator Pod, antctl can
+  connect to the Flow Aggregator and query information from it(e.g. flow records
+  related statistics).
 
 ## Table of Contents
 
@@ -27,6 +30,8 @@ running in two different modes:
   - [OVS packet tracing](#ovs-packet-tracing)
   - [Traceflow](#traceflow)
   - [Antctl Proxy](#antctl-proxy)
+  - [Dumping flow records](#dumping-flow-records)
+  - [Flow record metrics](#flow-record-metrics)
 <!-- /toc -->
 
 ## Installation
@@ -82,9 +87,9 @@ troubleshooting the Antrea system.
 ### Showing or changing log verbosity level
 
 Starting from version 0.10.0, Antrea supports showing or changing the log
-verbosity level of Antrea Controller or Agent using the `antctl log-level`
-command. The command can only run locally inside the `antrea-controller` or
-`antrea-agent` container.
+verbosity level of Antrea Controller, Antrea Agent or Flow Aggregator using the
+`antctl log-level` command. The command can only run locally inside the
+`antrea-controller`, `antrea-agent` or `flow-aggregator` container.
 
 The following command prints the current log verbosity level:
 
@@ -500,3 +505,97 @@ This feature is useful if one wants to use the Go
 [pprof](https://golang.org/pkg/net/http/pprof/) tool to collect runtime
 profiling data about the Antrea components. Please refer to this
 [document](troubleshooting.md#profiling-antrea-components) for more information.
+
+### Dumping flow records
+
+Flow Aggregator supports dumping flow records stored in the flow aggregator.
+The `antctl get flowrecords` can dump all flow records, records with a complete
+flow key, or records matching a partial flow key. A complete flow key is 5-tuple
+of a network connection containing Source IP Address, Destination IP Address,
+Source Port, Destination Port and Transport protocol. A partitial flow key
+consists of one or several elements from a complete flow key.
+`antctl get flowrecords --help` shows the usage of the command. This section
+lists a few dumping flor records command examples.
+
+```bash
+# Get the list of all flow records
+antctl get flowrecords
+# Get the list of flow records with a complete filter
+antctl get flowrecords --srcaddr 10.0.0.1 --dstaddr 10.0.0.2 --proto 6 --srcport 1234 --dstport 5678
+# Get the list flow records with a partial fiter, e.g. source address and source port
+antctl get flowrecords --srcaddr 10.0.0.1 --srcport 1234
+```
+
+Example outputs of dumping flow records:
+
+```bash
+$ antctl get flowrecords --srcaddr 10.10.0.1 --srcport 47340
+Flow Records
+    flowStartSeconds: 1633653734 
+    flowEndSeconds: 1633653796 
+    flowEndReason: 2 
+    sourceTransportPort: 47340 
+    destinationTransportPort: 4739 
+    protocolIdentifier: 6 
+    packetTotalCount: 14 
+    octetTotalCount: 2472 
+    packetDeltaCount: 14 
+    octetDeltaCount: 2472 
+    sourceIPv4Address: 10.10.0.1 
+    destinationIPv4Address: 10.10.1.9 
+    reversePacketTotalCount: 13 
+    reverseOctetTotalCount: 2169 
+    reversePacketDeltaCount: 13 
+    reverseOctetDeltaCount: 2169 
+    sourcePodName:  
+    sourcePodNamespace:  
+    sourceNodeName:  
+    destinationPodName: flow-aggregator-6d56fbf78d-7p5zm 
+    destinationPodNamespace: flow-aggregator 
+    destinationNodeName: k8s-node-worker-1 
+    destinationServicePort: 0 
+    destinationServicePortName:  
+    ingressNetworkPolicyName:  
+    ingressNetworkPolicyNamespace:  
+    ingressNetworkPolicyType: 0 
+    ingressNetworkPolicyRuleName:  
+    ingressNetworkPolicyRuleAction: 0 
+    egressNetworkPolicyName:  
+    egressNetworkPolicyNamespace:  
+    egressNetworkPolicyType: 0 
+    egressNetworkPolicyRuleName:  
+    egressNetworkPolicyRuleAction: 0 
+    tcpState: ESTABLISHED 
+    flowType: 2 
+    destinationClusterIPv4: 0.0.0.0 
+    octetDeltaCountFromSourceNode: 0 
+    octetDeltaCountFromDestinationNode: 2472 
+    octetTotalCountFromSourceNode: 0 
+    octetTotalCountFromDestinationNode: 2472 
+    packetDeltaCountFromSourceNode: 0 
+    packetDeltaCountFromDestinationNode: 14 
+    packetTotalCountFromSourceNode: 0 
+    packetTotalCountFromDestinationNode: 14 
+    reverseOctetDeltaCountFromSourceNode: 0 
+    reverseOctetDeltaCountFromDestinationNode: 2169 
+    reverseOctetTotalCountFromSourceNode: 0 
+    reverseOctetTotalCountFromDestinationNode: 2169 
+    reversePacketDeltaCountFromSourceNode: 0 
+    reversePacketDeltaCountFromDestinationNode: 13 
+    reversePacketTotalCountFromSourceNode: 0 
+    reversePacketTotalCountFromDestinationNode: 13 
+```
+
+### Flow record metrics
+
+Flow Aggregator supports printing record metrics. The `antctl get recordmetrics`
+can print all metrics related to the flow aggregator. It includes number of
+records received, number of records exported, number of flows stored in the
+Flow Aggregator and number of exporters connected to controller.
+
+Example outputs of record metrics:
+
+```bash
+RECORDS-EXPORTED RECORDS-RECEIVED FLOWS EXPORTERS-CONNECTED
+46               118              7     2      
+```

docs/network-flow-visibility.md

@@ -354,6 +354,12 @@ the [new fields](#ies-from-antrea-ie-registry) in Antrea Enterprise IPFIX regist
 corresponding to the Source Node and Destination Node, so that flow statistics from
 different Nodes can be preserved.
 
+### Antctl support
+
+Flow Aggregator supports antctl which is the command-line tool to access Flow
+Aggregator API. It supports dumping flow records and record metrics. Refer to
+the [antctl documentation](antctl.md#dumping-flow-records) for more information.
+
 ## Quick deployment
 
 If you would like to quickly try Network Flow Visibility feature, you can deploy

docs/troubleshooting.md

@@ -13,6 +13,9 @@
   - [Using antctl](#using-antctl-1)
   - [Using antctl proxy](#using-antctl-proxy-1)
   - [Directly accessing the antrea-agent API](#directly-accessing-the-antrea-agent-api)
+- [Accessing the flow-aggregator API](#accessing-the-flow-aggregator-api)
+  - [Using antctl](#using-antctl-2)
+  - [Directly accessing the flow-aggregator API](#directly-accessing-the-flow-aggregator-api)
 - [Troubleshooting Open vSwitch](#troubleshooting-open-vswitch)
 - [Troubleshooting with antctl](#troubleshooting-with-antctl)
 - [Profiling Antrea components](#profiling-antrea-components)
@@ -181,6 +184,38 @@ curl --insecure --header "Authorization: Bearer $TOKEN" https://<Node IP address
 However, in this case you will be limited to the endpoints that `antctl` is
 allowed to access, as defined [here](/build/yamls/base/antctl.yml).
 
+## Accessing the flow-aggregator API
+
+flow-aggregator runs as a Deployment and exposes its API via a
+local endpoint. There are two ways you can access it:
+
+### Using antctl
+
+To use `antctl` to access the flow-aggregator API, you need to exec into the
+flow-aggregator container first. `antctl` is embedded in the image so it can be
+used directly.
+
+For example, you can dump the flow records with this command:
+
+```bash
+# Get into the flow-aggregator container
+kubectl exec -it <flow-aggregator Pod name> -n flow-aggregator -- bash
+# View the flow records
+antctl get flowrecords
+```
+
+### Directly accessing the flow-aggregator API
+
+If you want to directly access the flow-aggregator API, you need to log into
+the Node that the flow-aggregator runs on or exec into the antrea-agent
+container. Then access the local endpoint directly using the Bearer Token
+stored in the file system:
+
+```bash
+TOKEN=$(cat /var/run/antrea/apiserver/loopback-client-token)
+curl --insecure --header "Authorization: Bearer $TOKEN" https://127.0.0.1:10350/
+```
+
 ## Troubleshooting Open vSwitch
 
 OVS daemons (`ovsdb-server` and `ovs-vswitchd`) run inside the `antrea-ovs`

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/ti-mo/conntrack v0.4.0
 	github.com/vishvananda/netlink v1.1.1-0.20210510164352-d17758a128bf
-	github.com/vmware/go-ipfix v0.5.8
+	github.com/vmware/go-ipfix v0.2.1-0.20211005220332-fb8df2e846c1
 	golang.org/x/crypto v0.0.0-20210503195802-e9a32991a82e
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6
 	golang.org/x/mod v0.4.2
@@ -151,3 +151,5 @@ require (
 // hcshim repo is modifed to add "AdditionalParams" field to HNSEndpoint struct.
 // We will use this replace before pushing the change to hcshim upstream repo.
 replace github.com/Microsoft/hcsshim v0.8.9 => github.com/ruicao93/hcsshim v0.8.10-0.20210114035434-63fe00c1b9aa
+
+replace github.com/vmware/go-ipfix v0.5.8 => github.com/vmware/go-ipfix v0.2.1-0.20211005220332-fb8df2e846c1

go.sum

@@ -653,8 +653,8 @@ github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc/go.mod h1:ZjcWmF
 github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae h1:4hwBBUfQCFe3Cym0ZtKyq7L16eZUtYKs+BaHDN6mAns=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
-github.com/vmware/go-ipfix v0.5.8 h1:/npDq+uGp9AoESNJvjsT6f3vT2rQc1Oq1RhHNN+N06k=
-github.com/vmware/go-ipfix v0.5.8/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
+github.com/vmware/go-ipfix v0.2.1-0.20211005220332-fb8df2e846c1 h1:hPCwabBFM3Wj0L5YecN7QUliMUc/n1W/C8d0TrCn4rA=
+github.com/vmware/go-ipfix v0.2.1-0.20211005220332-fb8df2e846c1/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=

pkg/agent/flowexporter/exporter/exporter.go

@@ -138,7 +138,6 @@ func prepareExporterInputArgs(collectorAddr, collectorProto, nodeName string) ex
 		expInput.IsEncrypted = false
 		expInput.CollectorProtocol = collectorProto
 	}
-	expInput.PathMTU = 0
 
 	return expInput
 }