NETOBSERV-589: send kafka flows as protobuf lines (#55)

* NETOBSERV-589: send kafka flows as protobuf batches

* Moved to 1 flow == 1 kafka message

* fixed e2e tests manifests

* fixed config doc

* remove linger comment

Author: Mario Macias

docs/config.md

@@ -23,27 +23,15 @@ The following environment variables are available to configure the NetObserv eBF
   that flows are kept in the accounting cache before being flushed to the collector.
 * `LOG_LEVEL` (default: `info`). From more to less verbose: `trace`, `debug`, `info`, `warn`,
   `error`, `fatal`, `panic`.
-* `BUFFERS_LENGTH` (default: `50`). Length of the internal communication channels between the different
-  processing stages. Most probably you won't need to change this value.
-* `LISTEN_INTERFACES` (default: `watch`). Mechanism used by the agent to listen for added or removed
-  network interfaces. Accepted values are:
-  - `watch`: interfaces are traced immediately after they are created. This is
-    the recommended setting for most configurations.
-  - `poll`: recommended mostly as a fallback mechanism if `watch` misbehaves. It periodically 
-    queries the current network interfaces. The poll frequency is specified by the
-    `LISTEN_POLL_PERIOD` variable.
-* `LISTEN_POLL_PERIOD` (default: `10s`). When `LISTEN_INTERFACES` value is `poll`, this duration
-  string specifies the frequency in which the current network interfaces are polled.
 * `KAFKA_BROKERS` (required if `EXPORT` is `kafka`). Comma-separated list of tha addresses of the
   brokers of the Kafka cluster that this agent is configured to send messages to.
 * `KAFKA_TOPIC`(default: `network-flows`). Name of the topic where the flows' processor will receive
   the flows from.
-* `KAFKA_BATCH_SIZE` (default: `100`). Limit on how many messages will be buffered before being sent
+* `KAFKA_BATCH_MESSAGES` (default: `1000`). Limit on how many messages will be buffered before being sent
   to a Kafka partition.
-* `KAFKA_BATCH_BYTES` (default: `1048576`). Limit of the maximum size of a request in bytes before
+  you actually need to set the `CACHE_MAX_FLOWS` and/or `MESSAGE_MAX_FLOW_ENTRIES`
+* `KAFKA_BATCH_SIZE` (default: `1048576`). Limit of the maximum size of a request in bytes before
   being sent to a Kafka partition.
-* `KAFKA_ASYNC` (default: `true`). If `true`, the message writing process will never block. It also
-  means that errors are ignored since the caller will not receive the returned value.
 * `KAFKA_COMPRESSION` (default: `none`). Compression codec to be used to compress messages. Accepted
   values: `none`, `gzip`, `snappy`, `lz4`, `zstd`.
 * `KAFKA_ENABLE_TLS` (default: false). If `true`, enable TLS encryption for Kafka messages. The following settings are used only when TLS is enabled:
@@ -53,3 +41,23 @@ The following environment variables are available to configure the NetObserv eBF
   * `KAFKA_TLS_USER_KEY_PATH` (default: unset). Path to the user (client) private key for mutual TLS connections.
 * `PROFILE_PORT` (default: unset). Sets the listening port for [Go's Pprof tool](https://pkg.go.dev/net/http/pprof).
   If it is not set, profile is disabled.
+
+## Development-only variables
+
+The following configuration variables are mostly used for development and fine-grained debugging,
+so no user should need to change them.
+
+* `BUFFERS_LENGTH` (default: `50`). Length of the internal communication channels between the different
+  processing stages.
+* `KAFKA_ASYNC` (default: `true`). If `true`, the message writing process will never block. It also
+  means that errors are ignored since the caller will not receive the returned value.
+* `LISTEN_INTERFACES` (default: `watch`). Mechanism used by the agent to listen for added or removed
+  network interfaces. Accepted values are:
+  - `watch`: interfaces are traced immediately after they are created. This is
+    the recommended setting for most configurations.
+  - `poll`: recommended mostly as a fallback mechanism if `watch` misbehaves. It periodically
+    queries the current network interfaces. The poll frequency is specified by the
+    `LISTEN_POLL_PERIOD` variable.
+* `LISTEN_POLL_PERIOD` (default: `10s`). When `LISTEN_INTERFACES` value is `poll`, this duration
+  string specifies the frequency in which the current network interfaces are polled.
+

e2e/cluster/base/03-flp.yml

@@ -16,7 +16,7 @@ spec:
       serviceAccountName: ebpf-agent-test
       containers:
         - name: flp
-          image: quay.io/netobserv/flowlogs-pipeline:v0.1.3-rc3
+          image: quay.io/netobserv/flowlogs-pipeline:latest
           ports:
             - containerPort: 9999
               hostPort: 9999
@@ -64,7 +64,6 @@ data:
         write:
           type: loki
           loki:
-            type: loki
             staticLabels:
               app: netobserv-flowcollector
             labels:

e2e/kafka/manifests/20-flp-transformer.yml

@@ -16,7 +16,7 @@ spec:
       serviceAccountName: ebpf-agent-test
       containers:
         - name: flp
-          image: quay.io/netobserv/flowlogs-pipeline:v0.1.3-rc3
+          image: quay.io/netobserv/flowlogs-pipeline:latest
           args:
             - --config=/etc/flp/config.yaml
           volumeMounts:
@@ -49,7 +49,7 @@ data:
             topic: network-flows
             groupId: kafka-transformer
             decoder:
-              type: json
+              type: protobuf
       - name: enrich
         transform:
           type: network
@@ -65,7 +65,6 @@ data:
         write:
           type: loki
           loki:
-            type: loki
             staticLabels:
               app: netobserv-flowcollector
             labels:

pkg/agent/agent.go

@@ -95,21 +95,27 @@ func FlowsAgent(cfg *Config) (*Flows, error) {
 			}
 			transport.TLS = tlsConfig
 		}
-		exportFunc = (&exporter.KafkaJSON{
+		exportFunc = (&exporter.KafkaProto{
 			Writer: &kafkago.Writer{
 				Addr:      kafkago.TCP(cfg.KafkaBrokers...),
 				Topic:     cfg.KafkaTopic,
-				BatchSize: cfg.KafkaBatchSize,
+				BatchSize: cfg.KafkaBatchMessages,
+				// Assigning KafkaBatchSize to BatchBytes instead of BatchSize might be confusing here.
+				// The reason is that the "standard" Kafka name for this variable is "batch.size",
+				// which specifies the size of messages in terms of bytes, and not in terms of entries.
+				// We have decided to hide this library implementation detail and expose to the
+				// customer the common, standard name and meaning for batch.size
+				BatchBytes: int64(cfg.KafkaBatchSize),
 				// Segmentio's Kafka-go does not behave as standard Kafka library, and would
 				// throttle any Write invocation until reaching the timeout.
 				// Since we invoke write once each CacheActiveTimeout, we can safely disable this
 				// timeout throttling
 				// https://github.com/netobserv/flowlogs-pipeline/pull/233#discussion_r897830057
 				BatchTimeout: time.Nanosecond,
-				BatchBytes:   int64(cfg.KafkaBatchBytes),
 				Async:        cfg.KafkaAsync,
 				Compression:  compression,
 				Transport:    &transport,
+				Balancer:     &kafkago.RoundRobin{},
 			},
 		}).ExportFlows
 	default:

pkg/agent/config.go

@@ -55,12 +55,12 @@ type Config struct {
 	KafkaBrokers []string `env:"KAFKA_BROKERS" envSeparator:","`
 	// KafkaTopic is the name of the topic where the flows' processor will receive the flows from.
 	KafkaTopic string `env:"KAFKA_TOPIC" envDefault:"network-flows"`
-	// KafkaBatchSize sets the limit on how many messages will be buffered before being sent to a
+	// KafkaBatchMessages sets the limit on how many messages will be buffered before being sent to a
 	// partition.
-	KafkaBatchSize int `env:"KAFKA_BATCH_SIZE" envDefault:"100"`
-	// KafkaBatchBytes sets the limit of the maximum size of a request in bytes before being sent
+	KafkaBatchMessages int `env:"KAFKA_BATCH_MESSAGES" envDefault:"1000"`
+	// KafkaBatchSize sets the limit, in bytes, of the maximum size of a request before being sent
 	// to a partition.
-	KafkaBatchBytes int `env:"KAFKA_BATCH_BYTES" envDefault:"1048576"`
+	KafkaBatchSize int `env:"KAFKA_BATCH_SIZE" envDefault:"1048576"`
 	// KafkaAsync. If it's true, the message writing process will never block. It also means that
 	// errors are ignored since the caller will not receive the returned value.
 	KafkaAsync bool `env:"KAFKA_ASYNC" envDefault:"true"`

pkg/exporter/grpc_proto.go

@@ -3,12 +3,9 @@ package exporter
 import (
 	"context"
 
-	"github.com/sirupsen/logrus"
-	"google.golang.org/protobuf/types/known/timestamppb"
-
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/flow"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/grpc"
-	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
+	"github.com/sirupsen/logrus"
 )
 
 var glog = logrus.WithField("component", "exporter/GRPCProto")
@@ -37,99 +34,13 @@ func StartGRPCProto(hostPort string) (*GRPCProto, error) {
 func (g *GRPCProto) ExportFlows(input <-chan []*flow.Record) {
 	log := glog.WithField("collector", g.hostPort)
 	for inputRecords := range input {
-		entries := make([]*pbflow.Record, 0, len(inputRecords))
-		for _, record := range inputRecords {
-			entries = append(entries, flowToPB(record))
-		}
-		log.Debugf("sending %d records", len(entries))
-		if _, err := g.clientConn.Client().Send(context.TODO(), &pbflow.Records{
-			Entries: entries,
-		}); err != nil {
+		pbRecords := flowsToPB(inputRecords)
+		log.Debugf("sending %d records", len(pbRecords.Entries))
+		if _, err := g.clientConn.Client().Send(context.TODO(), pbRecords); err != nil {
 			log.WithError(err).Error("couldn't send flow records to collector")
 		}
-
 	}
 	if err := g.clientConn.Close(); err != nil {
 		log.WithError(err).Warn("couldn't close flow export client")
 	}
 }
-
-func v4FlowToPB(fr *flow.Record) *pbflow.Record {
-	return &pbflow.Record{
-		EthProtocol: uint32(fr.EthProtocol),
-		Direction:   pbflow.Direction(fr.Direction),
-		DataLink: &pbflow.DataLink{
-			SrcMac: macToUint64(&fr.DataLink.SrcMac),
-			DstMac: macToUint64(&fr.DataLink.DstMac),
-		},
-		Network: &pbflow.Network{
-			SrcAddr: &pbflow.IP{IpFamily: &pbflow.IP_Ipv4{Ipv4: fr.Network.SrcAddr.IntEncodeV4()}},
-			DstAddr: &pbflow.IP{IpFamily: &pbflow.IP_Ipv4{Ipv4: fr.Network.DstAddr.IntEncodeV4()}},
-		},
-		Transport: &pbflow.Transport{
-			Protocol: uint32(fr.Transport.Protocol),
-			SrcPort:  uint32(fr.Transport.SrcPort),
-			DstPort:  uint32(fr.Transport.DstPort),
-		},
-		Bytes: fr.Bytes,
-		TimeFlowStart: &timestamppb.Timestamp{
-			Seconds: fr.TimeFlowStart.Unix(),
-			Nanos:   int32(fr.TimeFlowStart.Nanosecond()),
-		},
-		TimeFlowEnd: &timestamppb.Timestamp{
-			Seconds: fr.TimeFlowEnd.Unix(),
-			Nanos:   int32(fr.TimeFlowEnd.Nanosecond()),
-		},
-		Packets:   uint64(fr.Packets),
-		Interface: fr.Interface,
-	}
-}
-
-func v6FlowToPB(fr *flow.Record) *pbflow.Record {
-	return &pbflow.Record{
-		EthProtocol: uint32(fr.EthProtocol),
-		Direction:   pbflow.Direction(fr.Direction),
-		DataLink: &pbflow.DataLink{
-			SrcMac: macToUint64(&fr.DataLink.SrcMac),
-			DstMac: macToUint64(&fr.DataLink.DstMac),
-		},
-		Network: &pbflow.Network{
-			SrcAddr: &pbflow.IP{IpFamily: &pbflow.IP_Ipv6{Ipv6: fr.Network.SrcAddr[:]}},
-			DstAddr: &pbflow.IP{IpFamily: &pbflow.IP_Ipv6{Ipv6: fr.Network.DstAddr[:]}},
-		},
-		Transport: &pbflow.Transport{
-			Protocol: uint32(fr.Transport.Protocol),
-			SrcPort:  uint32(fr.Transport.SrcPort),
-			DstPort:  uint32(fr.Transport.DstPort),
-		},
-		Bytes: fr.Bytes,
-		TimeFlowStart: &timestamppb.Timestamp{
-			Seconds: fr.TimeFlowStart.Unix(),
-			Nanos:   int32(fr.TimeFlowStart.Nanosecond()),
-		},
-		TimeFlowEnd: &timestamppb.Timestamp{
-			Seconds: fr.TimeFlowEnd.Unix(),
-			Nanos:   int32(fr.TimeFlowEnd.Nanosecond()),
-		},
-		Packets:   uint64(fr.Packets),
-		Interface: fr.Interface,
-	}
-}
-
-func flowToPB(fr *flow.Record) *pbflow.Record {
-	if fr.EthProtocol == flow.IPv6Type {
-		return v6FlowToPB(fr)
-	}
-	return v4FlowToPB(fr)
-}
-
-// Mac bytes are encoded in the same order as in the array. This is, a Mac
-// like 11:22:33:44:55:66 will be encoded as 0x112233445566
-func macToUint64(m *flow.MacAddr) uint64 {
-	return uint64(m[5]) |
-		(uint64(m[4]) << 8) |
-		(uint64(m[3]) << 16) |
-		(uint64(m[2]) << 24) |
-		(uint64(m[1]) << 32) |
-		(uint64(m[0]) << 40)
-}

pkg/exporter/kafka_json.go

@@ -0,0 +1,54 @@
+package exporter
+
+import (
+	"context"
+
+	"github.com/netobserv/netobserv-ebpf-agent/pkg/flow"
+	kafkago "github.com/segmentio/kafka-go"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/proto"
+)
+
+var klog = logrus.WithField("component", "exporter/KafkaProto")
+
+type kafkaWriter interface {
+	WriteMessages(ctx context.Context, msgs ...kafkago.Message) error
+}
+
+// KafkaProto exports flows over Kafka, encoded as a protobuf that is understandable by the
+// Flowlogs-Pipeline collector
+type KafkaProto struct {
+	Writer kafkaWriter
+}
+
+func (kp *KafkaProto) ExportFlows(input <-chan []*flow.Record) {
+	klog.Info("starting Kafka exporter")
+	for records := range input {
+		kp.batchAndSubmit(records)
+	}
+}
+
+func (kp *KafkaProto) batchAndSubmit(records []*flow.Record) {
+	klog.Debugf("sending %d records", len(records))
+	msgs := make([]kafkago.Message, 0, len(records))
+	for _, record := range records {
+		pbBytes, err := proto.Marshal(flowToPB(record))
+		if err != nil {
+			klog.WithError(err).Debug("can't encode protobuf message. Ignoring")
+			continue
+		}
+		msgs = append(msgs, kafkago.Message{Value: pbBytes})
+	}
+
+	if err := kp.Writer.WriteMessages(context.TODO(), msgs...); err != nil {
+		klog.WithError(err).Error("can't write messages into Kafka")
+	}
+}
+
+type JSONRecord struct {
+	*flow.Record
+	TimeFlowStart   int64
+	TimeFlowEnd     int64
+	TimeFlowStartMs int64
+	TimeFlowEndMs   int64
+}