Support Kafka flow export in Flow Aggregator #2466

srikartati · 2021-07-24T05:03:25Z

This commit supports Kafka flow export in Flow Aggregator
by adding Kafka producer. Given a Kafka broker, Kafka producer
converts the aggregated flow record to a Kafka flow message
and send it to the Kafka broker, which will inturn be sent to
the Kafka consumer, where it could get decoded. Our Kafka producer
and consumer are dependent on sarama Kafka library.

Our Kafka producer in Flow Aggregator has the capability to support
multiple proto schema. However, we are currently supporting one proto
schema, which is named as "AntreaFlowMsg".

We add e2e tests by utilizing the Kafka flow collector provided
in go-ipfix repo. Kafka flow collector is a combination of a lean
Kafka broker and Kafka consumer.

Signed-off-by: Srikar Tati [email protected]

lgtm-com · 2021-07-24T05:38:40Z

This pull request introduces 1 alert when merging 308835b into fb28c8e - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

lgtm-com · 2021-07-26T21:50:04Z

This pull request introduces 1 alert when merging 6dcc491 into cdc8453 - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

lgtm-com · 2021-07-27T04:39:39Z

This pull request introduces 1 alert when merging 33988d7 into cdc8453 - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

lgtm-com · 2021-07-28T05:16:14Z

This pull request introduces 1 alert when merging 93a6667 into 6f33da3 - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

lgtm-com · 2021-08-18T05:45:36Z

This pull request introduces 1 alert when merging 26e399f into 7f90fc0 - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

srikartati · 2021-08-18T19:36:20Z

/test-all

lgtm-com · 2021-08-18T20:01:19Z

This pull request introduces 1 alert when merging 0d86a27 into ea26e6a - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

codecov-commenter · 2021-08-18T20:13:37Z

Codecov Report

Merging #2466 (dd94b3d) into main (8d228c1) will decrease coverage by 20.58%.
The diff coverage is 4.76%.

@@             Coverage Diff             @@
##             main    #2466       +/-   ##
===========================================
- Coverage   61.43%   40.84%   -20.59%     
===========================================
  Files         284      158      -126     
  Lines       23470    19557     -3913     
===========================================
- Hits        14418     7989     -6429     
- Misses       7504    10818     +3314     
+ Partials     1548      750      -798

Flag	Coverage Δ
kind-e2e-tests	`?`
unit-tests	`40.84% <4.76%> (-0.05%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
pkg/flowaggregator/flowaggregator.go	`19.46% <4.76%> (-49.83%)`	⬇️
pkg/ovs/openflow/default.go	`0.00% <0.00%> (-100.00%)`	⬇️
pkg/util/runtime/runtime.go	`0.00% <0.00%> (-100.00%)`	⬇️
pkg/agent/cniserver/pod_configuration_linux.go	`0.00% <0.00%> (-100.00%)`	⬇️
pkg/ovs/openflow/logs.go	`9.52% <0.00%> (-90.48%)`	⬇️
pkg/apis/controlplane/register.go	`0.00% <0.00%> (-90.00%)`	⬇️
pkg/agent/nodeportlocal/k8s/annotations.go	`0.00% <0.00%> (-83.88%)`	⬇️
pkg/agent/agent_linux.go	`0.00% <0.00%> (-80.00%)`	⬇️
pkg/agent/client.go	`0.00% <0.00%> (-77.42%)`	⬇️
pkg/ovs/ovsconfig/ovs_client_linux.go	`0.00% <0.00%> (-76.93%)`	⬇️
... and 231 more

zyiou · 2021-08-18T22:50:29Z

build/yamls/flow-aggregator.yml

+    # Provide the Kafka broker address to send the aggregated flow records in the
+    # given protobuf format. Kafka flow export is only enabled if the broker address
+    # is provided.
+    #kafkaBrokerAddress: ""


Do we support multiple broker addresses here? if so, we can specify the format (e.g. string divided by comma)

Currently, the plan is to support a single broker. In the future, when we have the requirement of multiple brokers, we can look at enhancing this configmap parameter.

zyiou · 2021-08-18T22:53:08Z

build/yamls/flow-aggregator.yml

+    # of multiple proto formats, this config parameter could take in multiple values.
+    #kafkaProtoSchema: "AntreaFlowMsg"
+
+    # Provide the topic name on Kafka Broker, where the flow messages are intended


Broker -> broker to keep consistent with other usage in description.

zyiou · 2021-08-18T23:02:49Z

hack/update-codegen.sh

@@ -22,7 +22,7 @@ ANTREA_ROOT="$( cd "$( dirname "${BASH_SOURCE[0]}" )/../" && pwd )"
 IMAGE_NAME="antrea/codegen:kubernetes-1.21.0"

 function docker_run() {
-  docker pull ${IMAGE_NAME}
+  #docker pull ${IMAGE_NAME}


comment by mistake?

I am using a local image for codegen with new verison of protoc-gen-go. Therefore, commenting this when running codegen on local machine. Once the image on antrea repo on docker registry is updated (before merging this PR), I will remove this.

zyiou · 2021-08-18T23:05:40Z

pkg/apis/cni/v1beta1/cni.pb.go

@@ -1,29 +1,43 @@
+// Copyright 2019 Antrea Authors


Are file changes of cni related to Kafka flow export support for flow aggregator? If not, I feel this should be put in a separate PR.

This is because of version change in protoc-gen-go. All this is generated code.

I agree with @zyiou (left another comment above). The Protobuf version upgrade should be handled in a separate PR, even if the upgrade is only motivated by this change.

pkg/flowaggregator/flowaggregator.go

pkg/flowaggregator/kafka/protobuf/antrea.proto

srikartati · 2021-08-23T20:58:28Z

Tagging this as v1.3. Will revisit the milestone after one round of reviews if needed.

lgtm-com · 2021-08-23T21:23:29Z

This pull request introduces 1 alert when merging 6266f94 into 4aaa90e - view on LGTM.com

new alerts:

1 for Size computation for allocation may overflow

build/yamls/flow-aggregator.yml

antoninbas · 2021-08-24T23:05:50Z

cmd/flow-aggregator/flow-aggregator.go

+	CACertFile            = "ca.crt"
+	TLSCertFile           = "tls.crt"
+	TLSKeyFile            = "tls.key"


just to clarify, ca.crt is to authenticate the Kafka broker, while tls.crt and tls.key are to authenticate the aggregator to the broker?

maybe this should be covered by comments

There are two different options. Added comments.

pkg/agent/cniserver/server_test.go

antoninbas · 2021-08-24T23:11:34Z

pkg/apis/cni/v1beta1/cni.pb.go

@@ -1,29 +1,43 @@
+// Copyright 2019 Antrea Authors


I agree with @zyiou (left another comment above). The Protobuf version upgrade should be handled in a separate PR, even if the upgrade is only motivated by this change.

antoninbas · 2021-08-24T23:12:37Z

pkg/flowaggregator/flowaggregator.go

+						klog.Errorf("Error when initializing kafka producer: %v, will retry in %s", err, fa.activeFlowRecordTimeout)
+					} else {
+						klog.Info("Initialized kafka producer")


please use structured logging for all new log messages

antoninbas · 2021-08-25T18:09:43Z

Switching milestone to v1.4 after talking with Srikar.

srikartati · 2021-09-22T23:07:39Z

Thanks @antoninbas for comments. Addressed them. Please take a look at this PR again.

antoninbas · 2021-09-23T19:48:46Z

build/yamls/flow-aggregator.yml

+      # Provide the Kafka broker address to send the aggregated flow records in the
+      # given protobuf format. Kafka flow export is only enabled if the broker address
+      # is provided.
+      #kafkaBrokerAddress: ""
+
+      # Provide the Kafka proto schema name. Currently, Flow Aggregator supports only
+      # one Antrea proto format called AntreaFlowMsg. In the future, with support
+      # of multiple proto formats, this config parameter could take in multiple values.
+      #kafkaProtoSchema: "AntreaFlowMsg"
+
+      # Provide the topic name on Kafka Broker, where the flow messages are intended
+      # to be received.
+      #kafkaBrokerTopic: "AntreaTopic"
+
+      # Enable TLS communication between the Kafka producer and the Kafka broker.
+      #kafkaTLSEnable: false
+
+      # Enable or disable the TLS certificate verification. This is to support the
+      # Kafka TLS certs that are either self-signed, mis-configured or expired.
+      #kafkaTLSSkipVerify: false


I don't know if we need to repeat the kafka prefix for all sub-parameters

antoninbas · 2021-09-23T19:49:46Z

build/yamls/flow-aggregator.yml

+    # KafkaParams contains all configuration parameters required to export flow
+    # records to a Kafka broker.
+    kafkaParams:
+      # Provide the Kafka broker address to send the aggregated flow records in the


Provide the Kafka broker address to which the aggregated flow records will be sent, in the given Protobuf format

antoninbas · 2021-09-23T19:49:57Z

build/yamls/flow-aggregator.yml

+      # Provide the Kafka proto schema name. Currently, Flow Aggregator supports only
+      # one Antrea proto format called AntreaFlowMsg. In the future, with support
+      # of multiple proto formats, this config parameter could take in multiple values.
+      #kafkaProtoSchema: "AntreaFlowMsg"


s/proto/Protobuf

antoninbas · 2021-09-23T19:51:26Z

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

+      - name: flow-aggregator-kafka-tls
+        secret:
+          secretName: flow-aggregator-kafka-tls
+          defaultMode: 0400


I notice we use 256 for the antrea controller, which is the same value but in decimal
it would be good to harmonize: 0400 makes more sense to me, so not sure why we are using 256 elsewhere
should be in a follow-up PR though

antoninbas · 2021-09-23T19:53:45Z

ci/kind/test-e2e-kind.sh

+               "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.7" \
+               "projects.registry.vmware.com/antrea/wireguard-go:0.0.20210424")
+FLOW_VISIBILITY_IMAGES_LIST=("projects.registry.vmware.com/antrea/ipfix-collector:latest" \


ipfix-collector listed twice with different tags?
I think we should always specify a released version. Even if it means more frequent updates, the tests are less likely to break.

antoninbas · 2021-09-23T19:59:11Z

cmd/flow-aggregator/flow-aggregator.go

+			caCertPath = path.Join(certDir, CACertFile)
+			tlsCertPath = path.Join(certDir, TLSCertFile)
+			tlsKeyPath = path.Join(certDir, TLSKeyFile)
+			// The secret may be created after the Pod is created, for example, when cert-manager is used the secret
+			// is created asynchronously. It waits for a while before it's considered to be failed.
+			if err = wait.PollImmediate(2*time.Second, certReadyTimeout, func() (bool, error) {
+				for _, path := range []string{caCertPath, tlsCertPath, tlsKeyPath} {
+					f, err := os.Open(path)
+					if err != nil {
+						klog.Warningf("Couldn't read %s when applying the kafka TLS certificate, retrying", path)
+						return false, nil
+					}
+					f.Close()
+				}
+				return true, nil
+			}); err != nil {
+				return fmt.Errorf("error reading Kafka TLS CA cert (%s), cert (%s), and key (%s) files present at \"%s\"", CACertFile, TLSCertFile, TLSKeyFile, certDir)
+			}
+		}


there is no reload mechanism for the certificates? what we do for the antrea-controller's apiserver I think is that the certificates are reloaded from file periodically and changes are detected

antoninbas · 2021-09-23T20:01:42Z

pkg/flowaggregator/kafka/convertor.go

+const (
+	AntreaFlowMsg string = "AntreaFlowMsg"
+	AntreaTopic   string = "AntreaTopic"
+)


this doesn't make it clear what the constant represents when you use it

if possible we should have somthing like this

type ProtobufSchema string const ( ProtobufSchemaAntreaFlowMsg ProtobufSchema = "AntreaFlowMsg" )

antoninbas · 2021-09-23T20:03:20Z

pkg/flowaggregator/flowaggregator.go

+				if fa.kafkaProducer.GetSaramaProducer() == nil {
+					err := fa.kafkaProducer.InitSaramaProducer()
+					if err != nil {
+						klog.ErrorS(err, "Error when initializing kafka producer", "retry timeout", fa.activeFlowRecordTimeout)


uniformize
s/kafka/Kafka

antoninbas · 2021-09-23T20:04:17Z

pkg/flowaggregator/kafka/convertor.go

+				klog.Warningf("Do not expect source IP: %v to be filled already", flowMsg.SrcIP)
+			}
+			flowMsg.SrcIP = ie.Value.(net.IP).String()
+		case "destinationIPv4Address", "destinationIPv6Address":
+			if flowMsg.DstIP != "" {
+				klog.Warningf("Do not expect destination IP: %v to be filled already", flowMsg.DstIP)


as part of the transition to structured logging, I don't think we use klog.Warningf, but just klog.InfoS. @tnqn is that correct?

antoninbas · 2021-09-23T20:05:43Z

pkg/flowaggregator/kafka/protobuf/antrea.proto

@@ -0,0 +1,66 @@
+// Copyright 2021 VMware, Inc.


should the proto file be versioned if you make breaking changes in the future?

zyiou · 2021-10-04T20:11:10Z

cmd/flow-aggregator/config.go

+	// given protobuf format. Kafka flow export is only enabled
+	// if the broker address is provided.
+	// Defaults to ""
+	BrokerAddress string `yaml:"kafkaBrokerAddress,omitempty"`


Field names in this file are not consistent with those defined in conf file. I think that's why the e2e tests do not pass.

zyiou · 2021-10-04T22:27:02Z

pkg/flowaggregator/kafka/protobuf/antrea.proto

+  uint64 TimeFlowStartInMilliSecs = 27;
+  uint32 TimeFlowEndInMilliSecs = 28;


Noticed that the data formats are different. We can remove these as they are not used in our record.

tnqn · 2021-10-28T17:01:15Z

@srikartati Do we still target v1.4 for this PR?

srikartati · 2021-10-28T17:35:12Z

@srikartati Do we still target v1.4 for this PR?

We have been working on a different pipeline, so halted this PR. Removed the release milestone.

This commit supports Kafka flow export in Flow Aggregator by adding Kafka producer. Given a Kafka broker, Kafka producer converts the aggregated flow record to a Kafka flow message and send it to the Kafka broker, which will inturn be sent to the Kafka consumer, where it could get decoded. Our Kafka producer and consumer are dependent on sarama Kafka library. Our Kafka producer in Flow Aggregator has the capability to support multiple proto schema. However, we are currently supporting one proto schema, which is named as "AntreaFlowMsg". We add e2e tests by utilizing the Kafka flow collector provided in go-ipfix repo. Kafka flow collector is a combination of a lean Kafka broker and Kafka consumer. Signed-off-by: Srikar Tati <[email protected]>

srikartati · 2021-11-12T02:17:20Z

Thanks, @zyiou for agreeing to take over this PR. I provided you the write access to my fork.
If that doesn't work, please feel free to open a separate PR.

srikartati force-pushed the kafka_export branch from 7439b0a to 308835b Compare July 24, 2021 05:06

srikartati marked this pull request as draft July 24, 2021 05:06

srikartati force-pushed the kafka_export branch 2 times, most recently from 13475b3 to 6dcc491 Compare July 26, 2021 21:19

srikartati force-pushed the kafka_export branch from 6dcc491 to 33988d7 Compare July 27, 2021 04:11

srikartati force-pushed the kafka_export branch from 33988d7 to 93a6667 Compare July 28, 2021 04:49

srikartati force-pushed the kafka_export branch from 93a6667 to 57b07dc Compare July 28, 2021 05:55

srikartati force-pushed the kafka_export branch from 57b07dc to 26e399f Compare August 18, 2021 05:19

srikartati force-pushed the kafka_export branch from 26e399f to 0d86a27 Compare August 18, 2021 19:33

srikartati marked this pull request as ready for review August 18, 2021 19:36

srikartati requested review from antoninbas and zyiou August 18, 2021 19:36

antrea-io deleted a comment from lgtm-com bot Aug 18, 2021

srikartati requested a review from dreamtalen August 18, 2021 20:30

zyiou reviewed Aug 18, 2021

View reviewed changes

srikartati force-pushed the kafka_export branch from 0d86a27 to 6266f94 Compare August 23, 2021 20:56

srikartati added this to the Antrea v1.3 release milestone Aug 23, 2021

antoninbas reviewed Aug 24, 2021

View reviewed changes

srikartati force-pushed the kafka_export branch from 6266f94 to cf9aefa Compare August 25, 2021 06:25

antrea-io deleted a comment from lgtm-com bot Aug 25, 2021

antoninbas modified the milestones: Antrea v1.3 release, Antrea v1.4 release Aug 25, 2021

srikartati force-pushed the kafka_export branch 2 times, most recently from 413db08 to 3ead296 Compare August 31, 2021 00:27

srikartati force-pushed the kafka_export branch 2 times, most recently from 8d926fd to b855a0e Compare September 22, 2021 23:06

srikartati requested a review from antoninbas September 22, 2021 23:07

antoninbas reviewed Sep 23, 2021

View reviewed changes

srikartati force-pushed the kafka_export branch from b855a0e to dd94b3d Compare September 28, 2021 04:11

zyiou reviewed Oct 4, 2021

View reviewed changes

zyiou reviewed Oct 7, 2021

View reviewed changes

srikartati removed this from the Antrea v1.4 release milestone Oct 28, 2021

srikartati force-pushed the kafka_export branch from dd94b3d to 17199df Compare October 28, 2021 17:37

srikartati added 2 commits November 11, 2021 16:17

Address comments

f44030a

srikartati force-pushed the kafka_export branch from 17199df to f44030a Compare November 12, 2021 00:26

srikartati closed this Jan 20, 2022

		uint64 TimeFlowStartInMilliSecs = 27;
		uint32 TimeFlowEndInMilliSecs = 28;

Support Kafka flow export in Flow Aggregator #2466

Support Kafka flow export in Flow Aggregator #2466

Conversation

srikartati commented Jul 24, 2021 • edited Loading

lgtm-com bot commented Jul 24, 2021

lgtm-com bot commented Jul 26, 2021

lgtm-com bot commented Jul 27, 2021

lgtm-com bot commented Jul 28, 2021

lgtm-com bot commented Aug 18, 2021

srikartati commented Aug 18, 2021

lgtm-com bot commented Aug 18, 2021

codecov-commenter commented Aug 18, 2021 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srikartati commented Aug 23, 2021

lgtm-com bot commented Aug 23, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

antoninbas commented Aug 25, 2021

srikartati commented Sep 22, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tnqn commented Oct 28, 2021

srikartati commented Oct 28, 2021

srikartati commented Nov 12, 2021

srikartati commented Jul 24, 2021 •

edited

Loading

codecov-commenter commented Aug 18, 2021 •

edited

Loading