Add S3Uploader to Flow Aggregator

build/charts/Makefile

@@ -6,7 +6,7 @@ VERSION := $(shell head -n 1 ../../VERSION | cut -c 2-)
 .PHONY: helm-docs
 helm-docs:
 	docker run --rm --volume "$(CURDIR):/helm-docs" --user=$(USERID):$(GRPID) jnorwood/helm-docs:v1.7.0
-	sed -i.bak "s/0.0.0/$(VERSION)/g" antrea/README.md # replace version placeholder
+	sed -i.bak "s/0\.0\.0/$(VERSION)/g" antrea/README.md # replace version placeholder
 	sed -i.bak "s/-dev-informational/--dev-informational/g" antrea/README.md # fix img.shields.io badge URLs
-	sed -i.bak "s/0.0.0/$(VERSION)/g" flow-aggregator/README.md # replace version placeholder
+	sed -i.bak "s/0\.0\.0/$(VERSION)/g" flow-aggregator/README.md # replace version placeholder
 	sed -i.bak "s/-dev-informational/--dev-informational/g" flow-aggregator/README.md # fix img.shields.io badge URLs

build/charts/flow-aggregator/README.md

@@ -26,7 +26,7 @@ Kubernetes: `>= 1.16.0-0`
 | clickHouse.commitInterval | string | `"8s"` | CommitInterval is the periodical interval between batch commit of flow records to DB. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". |
 | clickHouse.compress | bool | `true` | Compress enables lz4 compression when committing flow records. |
 | clickHouse.connectionSecret | object | `{"password":"clickhouse_operator_password","username":"clickhouse_operator"}` | Credentials to connect to ClickHouse. They will be stored in a Secret. |
-| clickHouse.databaseURL | string | `"tcp://clickhouse-clickhouse.flow-visibility.svc:9000"` |  |
+| clickHouse.databaseURL | string | `"tcp://clickhouse-clickhouse.flow-visibility.svc:9000"` | DatabaseURL is the url to the database. TCP protocol is required. |
 | clickHouse.debug | bool | `false` | Debug enables debug logs from ClickHouse sql driver. |
 | clickHouse.enable | bool | `false` | Determine whether to enable exporting flow records to ClickHouse. |
 | flowAggregatorAddress | string | `"flow-aggregator.flow-aggregator.svc"` | Provide DNS name or IP address of flow aggregator for generating TLS certificate. It must match the flowCollectorAddr parameter in the antrea-agent config. |
@@ -38,6 +38,15 @@ Kubernetes: `>= 1.16.0-0`
 | inactiveFlowRecordTimeout | string | `"90s"` | Provide the inactive flow record timeout as a duration string. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". |
 | logVerbosity | int | `0` |  |
 | recordContents.podLabels | bool | `false` | Determine whether source and destination Pod labels will be included in the flow records. |
+| s3Uploader.awsCredentials | object | `{"aws_access_key_id":"changeme","aws_secret_access_key":"changeme","aws_session_token":""}` | Credentials to authenticate to AWS. They will be stored in a Secret and injected into the Pod as environment variables. |
+| s3Uploader.bucketName | string | `""` | BucketName is the name of the S3 bucket to which flow records will be uploaded. It is required. |
+| s3Uploader.bucketPrefix | string | `""` | BucketPrefix is the prefix ("folder") under which flow records will be uploaded. |
+| s3Uploader.compress | bool | `true` | Compress enables gzip compression when uploading files to S3. |
+| s3Uploader.enable | bool | `false` | Determine whether to enable exporting flow records to AWS S3. |
+| s3Uploader.maxRecordsPerFile | int | `1000000` | MaxRecordsPerFile is the maximum number of records per file uploaded. It is not recommended to change this value. |
+| s3Uploader.recordFormat | string | `"CSV"` | RecordFormat defines the format of the flow records uploaded to S3. Only "CSV" is supported at the moment. |
+| s3Uploader.region | string | `"us-west-2"` | Region is used as a "hint" to get the region in which the provided bucket is located. An error will occur if the bucket does not exist in the AWS partition the region hint belongs to. |
+| s3Uploader.uploadInterval | string | `"60s"` | UploadInterval is the duration between each file upload to S3. |
 | testing.coverage | bool | `false` |  |
 
 ----------------------------------------------

build/charts/flow-aggregator/conf/flow-aggregator.conf

@@ -86,3 +86,40 @@ clickHouse:
   # Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
   # The minimum interval is 1s based on ClickHouse documentation for best performance.
   commitInterval: {{ .Values.clickHouse.commitInterval | quote }}
+
+# s3Uploader contains configuration options for uploading flow records to AWS S3.
+s3Uploader:
+  # Enable is the switch to enable exporting flow records to AWS S3.
+  # At the moment, the flow aggregator will look for the "standard" environment variables to
+  # authenticate to AWS. These can be static credentials (AWS_ACCESS_KEY_ID,
+  # AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN) or a Web Identity Token
+  # (AWS_WEB_IDENTITY_TOKEN_FILE).
+  enable: {{ .Values.s3Uploader.enable }}
+
+  # BucketName is the name of the S3 bucket to which flow records will be uploaded. If this
+  # field is empty, initialization will fail.
+  bucketName: {{ .Values.s3Uploader.bucketName | quote }}
+
+  # BucketPrefix is the prefix ("folder") under which flow records will be uploaded. If this
+  # is omitted, flow records will be uploaded to the root of the bucket.
+  bucketPrefix: {{ .Values.s3Uploader.bucketPrefix | quote }}
+
+  # Region is used as a "hint" to get the region in which the provided bucket is located.
+  # An error will occur if the bucket does not exist in the AWS partition the region hint
+  # belongs to. If region is omitted, the value of the AWS_REGION environment variable will
+  # be used, and if it is missing, we will default to "us-west-2".
+  region: {{ .Values.s3Uploader.region | quote }}
+
+  # RecordFormat defines the format of the flow records uploaded to S3. Only "CSV" is
+  # supported at the moment.
+  recordFormat: {{ .Values.s3Uploader.recordFormat | quote }}
+
+  # Compress enables gzip compression when uploading files to S3. Defaults to true.
+  compress: {{ .Values.s3Uploader.compress }}
+
+  # MaxRecordsPerFile is the maximum number of records per file uploaded. It is not recommended
+  # to change this value.
+  maxRecordsPerFile: {{ .Values.s3Uploader.maxRecordsPerFile }}
+
+  # UploadInterval is the duration between each file upload to S3.
+  uploadInterval: {{ .Values.s3Uploader.uploadInterval | quote }}

build/charts/flow-aggregator/templates/deployment.yaml

@@ -56,6 +56,21 @@ spec:
                 key: password
           - name: FA_CONFIG_MAP_NAME
             value: flow-aggregator-configmap
+          - name: AWS_ACCESS_KEY_ID
+            valueFrom:
+              secretKeyRef:
+                name: flow-aggregator-aws-credentials
+                key: aws_access_key_id
+          - name: AWS_SECRET_ACCESS_KEY
+            valueFrom:
+              secretKeyRef:
+                name: flow-aggregator-aws-credentials
+                key: aws_secret_access_key
+          - name: AWS_SESSION_TOKEN
+            valueFrom:
+              secretKeyRef:
+                name: flow-aggregator-aws-credentials
+                key: aws_session_token
         ports:
           - containerPort: 4739
         volumeMounts:

build/charts/flow-aggregator/templates/secrets.yaml

@@ -0,0 +1,24 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  labels:
+    app: flow-aggregator
+  name: clickhouse-secret
+  namespace: {{ .Release.Namespace }}
+type: Opaque
+stringData:
+  username: {{ .Values.clickHouse.connectionSecret.username }}
+  password: {{ .Values.clickHouse.connectionSecret.password }}
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  labels:
+    app: flow-aggregator
+  name: flow-aggregator-aws-credentials
+  namespace: {{ .Release.Namespace }}
+type: Opaque
+stringData:
+  aws_access_key_id: {{ .Values.s3Uploader.awsCredentials.aws_access_key_id | quote }}
+  aws_secret_access_key: {{ .Values.s3Uploader.awsCredentials.aws_secret_access_key | quote }}
+  aws_session_token: {{ .Values.s3Uploader.awsCredentials.aws_session_token | quote }}

build/charts/flow-aggregator/values.yaml

@@ -46,7 +46,7 @@ flowCollector:
 clickHouse:
   # -- Determine whether to enable exporting flow records to ClickHouse.
   enable: false
-  # DatabaseURL is the url to the database. TCP protocol is required.
+  # -- DatabaseURL is the url to the database. TCP protocol is required.
   databaseURL: "tcp://clickhouse-clickhouse.flow-visibility.svc:9000"
   # -- Debug enables debug logs from ClickHouse sql driver.
   debug: false
@@ -59,6 +59,32 @@ clickHouse:
   connectionSecret:
     username : "clickhouse_operator"
     password: "clickhouse_operator_password"
+# s3Uploader contains configuration options for uploading flow records to AWS S3.
+s3Uploader:
+  # -- Determine whether to enable exporting flow records to AWS S3.
+  enable: false
+  # -- BucketName is the name of the S3 bucket to which flow records will be uploaded. It is required.
+  bucketName: ""
+  # -- BucketPrefix is the prefix ("folder") under which flow records will be uploaded.
+  bucketPrefix: ""
+  # -- Region is used as a "hint" to get the region in which the provided bucket is located.
+  # An error will occur if the bucket does not exist in the AWS partition the region hint belongs to.
+  region: "us-west-2"
+  # -- RecordFormat defines the format of the flow records uploaded to S3. Only "CSV" is supported at the moment.
+  recordFormat: "CSV"
+  # -- Compress enables gzip compression when uploading files to S3.
+  compress: true
+  # -- MaxRecordsPerFile is the maximum number of records per file uploaded. It is not recommended
+  # to change this value.
+  maxRecordsPerFile: 1000000
+  # -- UploadInterval is the duration between each file upload to S3.
+  uploadInterval: "60s"
+  # -- Credentials to authenticate to AWS. They will be stored in a Secret and injected into the Pod
+  # as environment variables.
+  awsCredentials:
+    aws_access_key_id: "changeme"
+    aws_secret_access_key: "changeme"
+    aws_session_token: ""
 testing:
   ## -- Enable code coverage measurement (used when testing Flow Aggregator only).
   coverage: false

build/images/flow-aggregator/Dockerfile

@@ -31,4 +31,9 @@ LABEL description="The docker image for the flow aggregator"
 COPY --from=flow-aggregator-build /antrea/bin/flow-aggregator /
 COPY --from=flow-aggregator-build /antrea/bin/antctl /usr/local/bin/
 
+# install ca-certificates
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
 ENTRYPOINT ["/flow-aggregator"]

build/images/flow-aggregator/Dockerfile.coverage

@@ -32,3 +32,7 @@ USER root
 COPY --from=flow-aggregator-build /antrea/bin/flow-aggregator* /usr/local/bin/
 COPY --from=flow-aggregator-build /antrea/test/e2e/coverage/flow-aggregator-arg-file /
 COPY --from=flow-aggregator-build /antrea/bin/antctl* /usr/local/bin/
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates \
+    && rm -rf /var/lib/apt/lists/*

build/yamls/flow-aggregator.yml

@@ -238,6 +238,43 @@ data:
       # Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
       # The minimum interval is 1s based on ClickHouse documentation for best performance.
       commitInterval: "8s"
+
+    # s3Uploader contains configuration options for uploading flow records to AWS S3.
+    s3Uploader:
+      # Enable is the switch to enable exporting flow records to AWS S3.
+      # At the moment, the flow aggregator will look for the "standard" environment variables to
+      # authenticate to AWS. These can be static credentials (AWS_ACCESS_KEY_ID,
+      # AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN) or a Web Identity Token
+      # (AWS_WEB_IDENTITY_TOKEN_FILE).
+      enable: false
+
+      # BucketName is the name of the S3 bucket to which flow records will be uploaded. If this
+      # field is empty, initialization will fail.
+      bucketName: ""
+
+      # BucketPrefix is the prefix ("folder") under which flow records will be uploaded. If this
+      # is omitted, flow records will be uploaded to the root of the bucket.
+      bucketPrefix: ""
+
+      # Region is used as a "hint" to get the region in which the provided bucket is located.
+      # An error will occur if the bucket does not exist in the AWS partition the region hint
+      # belongs to. If region is omitted, the value of the AWS_REGION environment variable will
+      # be used, and if it is missing, we will default to "us-west-2".
+      region: "us-west-2"
+
+      # RecordFormat defines the format of the flow records uploaded to S3. Only "CSV" is
+      # supported at the moment.
+      recordFormat: "CSV"
+
+      # Compress enables gzip compression when uploading files to S3. Defaults to true.
+      compress: true
+
+      # MaxRecordsPerFile is the maximum number of records per file uploaded. It is not recommended
+      # to change this value.
+      maxRecordsPerFile: 1e+06
+
+      # UploadInterval is the duration between each file upload to S3.
+      uploadInterval: "60s"
 kind: ConfigMap
 metadata:
   labels:
@@ -258,6 +295,19 @@ stringData:
 type: Opaque
 ---
 apiVersion: v1
+kind: Secret
+metadata:
+  labels:
+    app: flow-aggregator
+  name: flow-aggregator-aws-credentials
+  namespace: flow-aggregator
+stringData:
+  aws_access_key_id: changeme
+  aws_secret_access_key: changeme
+  aws_session_token: ""
+type: Opaque
+---
+apiVersion: v1
 kind: Service
 metadata:
   labels:
@@ -324,6 +374,21 @@ spec:
               name: clickhouse-secret
         - name: FA_CONFIG_MAP_NAME
           value: flow-aggregator-configmap
+        - name: AWS_ACCESS_KEY_ID
+          valueFrom:
+            secretKeyRef:
+              key: aws_access_key_id
+              name: flow-aggregator-aws-credentials
+        - name: AWS_SECRET_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              key: aws_secret_access_key
+              name: flow-aggregator-aws-credentials
+        - name: AWS_SESSION_TOKEN
+          valueFrom:
+            secretKeyRef:
+              key: aws_session_token
+              name: flow-aggregator-aws-credentials
         image: projects.registry.vmware.com/antrea/flow-aggregator:latest
         imagePullPolicy: IfNotPresent
         name: flow-aggregator

go.mod

@@ -12,6 +12,10 @@ require (
 	github.com/Microsoft/hcsshim v0.8.9
 	github.com/TomCodeLV/OVSDB-golang-lib v0.0.0-20200116135253-9bbdfadcd881
 	github.com/awalterschulze/gographviz v2.0.1+incompatible
+	github.com/aws/aws-sdk-go-v2 v1.16.10
+	github.com/aws/aws-sdk-go-v2/config v1.16.0
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.23
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.27.4
 	github.com/blang/semver v3.5.1+incompatible
 	github.com/cheggaaa/pb/v3 v3.1.0
 	github.com/confluentinc/bincover v0.1.0
@@ -92,6 +96,20 @@ require (
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/VividCortex/ewma v1.1.1 // indirect
 	github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.4 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.12.12 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.11 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.11 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.18 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.0.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.9.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.12 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.13.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.11.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.16.12 // indirect
+	github.com/aws/smithy-go v1.12.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cenk/hub v1.0.1 // indirect
@@ -115,7 +133,7 @@ require (
 	github.com/go-openapi/swag v0.19.14 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/gnostic v0.5.7-v3refs // indirect
-	github.com/google/go-cmp v0.5.5 // indirect
+	github.com/google/go-cmp v0.5.8 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
@@ -127,6 +145,7 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/josharian/native v0.0.0-20200817173448-b6b71def0850 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect