Fix e2e tests with Kafka

- Update strimzi to 0.43
- Refactor tests: avoid using external address+node port+kafka client to
  inject and read data. Instead, leverage FLP entirely with 2 FLPs:
  - FLP1: ingest from static file, stamp, output to kafka
  - FLP2: ingest from flp1/kafka, stamp, output to logs

show pods in logs

Sticky kind version

Author: jotak

Makefile

@@ -42,6 +42,7 @@ ifneq ($(CLEAN_BUILD),)
 endif
 
 GOLANGCI_LINT_VERSION = v1.61.0
+KIND_VERSION = v0.22.0
 
 FLP_BIN_FILE=flowlogs-pipeline
 CG_BIN_FILE=confgenerator
@@ -105,7 +106,7 @@ prereqs: ## Check if prerequisites are met, and install missing dependencies
 .PHONY: prereqs-kind
 prereqs-kind: ## Check if prerequisites are met for running kind, and install missing dependencies
 	@echo "### Checking if KIND prerequisites are met, and installing missing dependencies"
-	test -f $(shell go env GOPATH)/bin/kind || GOFLAGS="" go install sigs.k8s.io/kind@latest
+	GOFLAGS="" go install sigs.k8s.io/kind@${KIND_VERSION}
 
 .PHONY: vendors
 vendors: ## Check go vendors

README.md

@@ -955,6 +955,7 @@ Images
   image-push            Push MULTIARCH_TARGETS images  
   manifest-build        Build MULTIARCH_TARGETS manifest  
   manifest-push         Push MULTIARCH_TARGETS manifest  
+  extract-binaries      Extract all MULTIARCH_TARGETS binaries  
   goyacc                Regenerate filters query langage  
   
 kubernetes  

docs/api.md

@@ -229,6 +229,7 @@ Following is the supported API format for filter transformations:
                                      input: entry input field
                                      value: specified value of input field:
                                      castInt: set true to cast the value field as an int (numeric values are float64 otherwise)
+         samplingField: sampling field name to be set when sampling is used; if the field already exists in flows, its value is multiplied with the new sampling
 </pre>
 ## Transform Network API
 Following is the supported API format for network transformations:

hack/update-docs.sh

@@ -28,7 +28,7 @@ update-animated-gif() {
 
 main() {
   update-readme
-  update-animated-gif
+#  update-animated-gif
 }
 
 main

pkg/test/e2e/kafka/flp-1.yaml

@@ -0,0 +1,25 @@
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: flp-1
+  labels:
+    app: flp-1
+spec:
+  template:
+    spec:
+      containers:
+        - name: flowlogs-pipeline
+          image: quay.io/netobserv/flowlogs-pipeline:e2e-amd64
+          args:
+            - "--config=/etc/flowlogs-pipeline/flowlogs-pipeline.conf.yaml"
+          imagePullPolicy: Never
+          volumeMounts:
+            - name: configuration
+              mountPath: "/etc/flowlogs-pipeline/"
+      volumes:
+        - name: configuration
+          configMap:
+            name: flp-config-1
+      restartPolicy: Never
+---

pkg/test/e2e/kafka/flp-2.yaml

@@ -0,0 +1,22 @@
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: flp-2
+  labels:
+    app: flp-2
+spec:
+  containers:
+    - name: flowlogs-pipeline
+      image: quay.io/netobserv/flowlogs-pipeline:e2e-amd64
+      args:
+        - "--config=/etc/flowlogs-pipeline/flowlogs-pipeline.conf.yaml"
+      imagePullPolicy: Never
+      volumeMounts:
+        - name: configuration
+          mountPath: "/etc/flowlogs-pipeline/"
+  volumes:
+    - name: configuration
+      configMap:
+        name: flp-config-2
+---

pkg/test/e2e/kafka/flp-config.yaml

@@ -0,0 +1,81 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: flp-config-1
+data:
+  flowlogs-pipeline.conf.yaml: |
+    pipeline:
+      - name: file_ingest
+      - name: transform
+        follows: file_ingest
+      - name: kafka_encode
+        follows: transform
+      - name: logs
+        follows: transform
+    parameters:
+      - name: file_ingest
+        ingest:
+          type: file
+          file:
+            filename: /etc/flowlogs-pipeline/ingest.jsonlogs
+            decoder:
+              type: json
+      - name: transform
+        transform:
+          type: filter
+          filter:
+            rules: 
+              - type: add_field
+                addField: 
+                  input: ProcessedByFLP1
+                  value: true
+      - name: kafka_encode
+        encode:
+          type: kafka
+          kafka:
+            address: my-cluster-kafka-bootstrap.default.svc:9092
+            topic: test
+      - name: logs
+        write:
+          type: stdout
+  ingest.jsonlogs: |
+    {"Bytes":20800,"DstAddr":"10.130.2.2","DstPort":36936}
+    {"Bytes":116000,"DstAddr":"10.0.156.12","DstPort":9537}
+    {"Bytes":20800,"DstAddr":"10.130.2.13","DstPort":3100}
+    {"Bytes":20800,"DstAddr":"10.128.4.12","DstPort":3100}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: flp-config-2
+data:
+  flowlogs-pipeline.conf.yaml: |
+    pipeline:
+      - name: kafka_ingest
+      - name: transform
+        follows: kafka_ingest
+      - name: logs
+        follows: transform
+    parameters:
+      - name: kafka_ingest
+        ingest:
+          type: kafka
+          kafka:
+            brokers: [my-cluster-kafka-bootstrap.default.svc:9092]
+            topic: test
+            groupid: group_test_in
+            decoder:
+              type: json
+      - name: transform
+        transform:
+          type: filter
+          filter:
+            rules: 
+              - type: add_field
+                addField: 
+                  input: ProcessedByFLP2
+                  value: true
+      - name: logs
+        write:
+          type: stdout
+---

pkg/test/e2e/kafka/flp-configs.yaml

@@ -1,12 +1,12 @@
 # Based on
-# https://raw.githubusercontent.com/strimzi/strimzi-kafka-operator/0.31.0/examples/kafka/kafka-ephemeral.yaml
+# https://raw.githubusercontent.com/strimzi/strimzi-kafka-operator/0.43.0/examples/kafka/kafka-ephemeral.yaml
 apiVersion: kafka.strimzi.io/v1beta2
 kind: Kafka
 metadata:
   name: my-cluster
 spec:
   kafka:
-    version: 3.2.1
+    version: 3.8.0
     replicas: 1
     listeners:
       - name: plain
@@ -27,7 +27,7 @@ spec:
       transaction.state.log.min.isr: 1
       default.replication.factor: 1
       min.insync.replicas: 1
-      inter.broker.protocol.version: "3.2"
+      inter.broker.protocol.version: "3.8"
     storage:
       type: ephemeral
   zookeeper: