Added kafka config in the CR

api/v1alpha1/flowcollector_types.go

@@ -68,6 +68,19 @@ type FlowCollectorIPFIX struct {
 	Sampling int32 `json:"sampling,omitempty" mapstructure:"sampling,omitempty"`
 }
 
+// FlowCollectorKafka defines the desired Kafka config of FlowCollector
+type FlowCollectorKafka struct {
+	// Important: Run "make generate" to regenerate code after modifying this file
+
+	//+kubebuilder:default:=""
+	// Address of the kafka server
+	Address string `json:"address"`
+
+	//+kubebuilder:default:=""
+	// Address of the kafka topic to use
+	Topic string `json:"topic"`
+}
+
 // FlowCollectorFLP defines the desired flowlogs-pipeline state of FlowCollector
 type FlowCollectorFLP struct {
 	// Important: Run "make generate" to regenerate code after modifying this file
@@ -128,6 +141,10 @@ type FlowCollectorFLP struct {
 	//+kubebuilder:default:=false
 	// PrintOutput is a debug flag to print flows exported in flowlogs-pipeline stdout
 	PrintOutput bool `json:"printOutput,omitempty"`
+
+	// Kafka configurations, if empty the operator will deploy a all-in-one FLP
+	// +optional
+	Kafka *FlowCollectorKafka `json:"kafka,omitempty"`
 }
 
 type FlowCollectorHPA struct {

config/crd/bases/flows.netobserv.io_flowcollectors.yaml

@@ -1180,6 +1180,22 @@ spec:
                     - Always
                     - Never
                     type: string
+                  kafka:
+                    description: Kafka configurations, if empty the operator will
+                      deploy a all-in-one FLP
+                    properties:
+                      address:
+                        default: ""
+                        description: Address of the kafka server
+                        type: string
+                      topic:
+                        default: ""
+                        description: Address of the kafka topic to use
+                        type: string
+                    required:
+                    - address
+                    - topic
+                    type: object
                   kind:
                     default: DaemonSet
                     description: Kind is the workload kind, either DaemonSet or Deployment

controllers/flowcollector_controller.go

@@ -136,7 +136,7 @@ func (r *FlowCollectorReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	}
 
 	// OVS config map for CNO
-	if err := ovsConfigController.Reconcile(ctx, desired); err != nil {
+	if err := ovsConfigController.Reconcile(ctx, desired, gfReconciler.GetServiceName(&desired.Spec.FlowlogsPipeline)); err != nil {
 		log.Error(err, "Failed to reconcile ovs-flows-config ConfigMap")
 	}
 

controllers/flowcollector_controller_test.go

@@ -44,6 +44,14 @@ var _ = Describe("FlowCollector Controller", func() {
 		Name:      constants.FLPName,
 		Namespace: otherNamespace,
 	}
+	gfKeyKafkaIngestor := types.NamespacedName{
+		Name:      constants.FLPName + flowlogspipeline.FlpConfSuffix[flowlogspipeline.ConfKafkaIngestor],
+		Namespace: operatorNamespace,
+	}
+	gfKeyKafkaTransformer := types.NamespacedName{
+		Name:      constants.FLPName + flowlogspipeline.FlpConfSuffix[flowlogspipeline.ConfKafkaTransformer],
+		Namespace: operatorNamespace,
+	}
 	cpKey1 := types.NamespacedName{
 		Name:      "network-observability-plugin",
 		Namespace: operatorNamespace,
@@ -437,6 +445,79 @@ var _ = Describe("FlowCollector Controller", func() {
 		})
 	})
 
+	Context("Changing kafka config", func() {
+		It("Should update kafka config successfully", func() {
+			Eventually(func() error {
+				fc := flowsv1alpha1.FlowCollector{}
+				if err := k8sClient.Get(ctx, crKey, &fc); err != nil {
+					return err
+				}
+				fc.Spec.FlowlogsPipeline.Kafka = &flowsv1alpha1.FlowCollectorKafka{Address: "loaclhost:9092", Topic: "FLP"}
+				return k8sClient.Update(ctx, &fc)
+			}).Should(Succeed())
+		})
+
+		It("Should deploy kafka ingestor and transformer", func() {
+			By("Expecting ingestor daemonset to be created")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKeyKafkaIngestor, &appsv1.DaemonSet{})
+			}, timeout, interval).Should(Succeed())
+
+			By("Expecting transformer deployment to be created")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKeyKafkaTransformer, &appsv1.Deployment{})
+			}, timeout, interval).Should(Succeed())
+
+			By("Not Expecting transformer service to be created")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKeyKafkaTransformer, &v1.Service{})
+			}, timeout, interval).Should(MatchError(`services "flowlogs-pipeline-ktransform" not found`))
+		})
+
+		It("Should delete previous flp deployment", func() {
+			By("Expecting deployment to be deleted")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKey1, &appsv1.DaemonSet{})
+			}, timeout, interval).Should(MatchError(`daemonsets.apps "flowlogs-pipeline" not found`))
+
+			By("Expecting service to be deleted")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKey1, &v1.Service{})
+			}, timeout, interval).Should(MatchError(`services "flowlogs-pipeline" not found`))
+		})
+
+		It("Should remove kafka config successfully", func() {
+			Eventually(func() error {
+				fc := flowsv1alpha1.FlowCollector{}
+				if err := k8sClient.Get(ctx, crKey, &fc); err != nil {
+					return err
+				}
+				fc.Spec.FlowlogsPipeline.Kafka = nil
+				return k8sClient.Update(ctx, &fc)
+			}).Should(Succeed())
+		})
+
+		It("Should deploy single flp again", func() {
+			By("Expecting daemonset to be created")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKey1, &appsv1.DaemonSet{})
+			}, timeout, interval).Should(Succeed())
+		})
+
+		It("Should delete kafka ingestor and transformer", func() {
+			By("Expecting ingestor daemonset to be deleted")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKeyKafkaIngestor, &appsv1.DaemonSet{})
+			}, timeout, interval).Should(MatchError(`daemonsets.apps "flowlogs-pipeline-kingestor" not found`))
+
+			By("Expecting transformer deployment to be deleted")
+			Eventually(func() interface{} {
+				return k8sClient.Get(ctx, gfKeyKafkaTransformer, &appsv1.Deployment{})
+			}, timeout, interval).Should(MatchError(`deployments.apps "flowlogs-pipeline-ktransform" not found`))
+		})
+
+	})
+
 	Context("Changing namespace", func() {
 		It("Should update namespace successfully", func() {
 			Eventually(func() error {

controllers/flowlogspipeline/flp_objects.go

@@ -37,10 +37,10 @@ const (
 	ConfKafkaTransformer = "kafkaTransformer"
 )
 
-var flpConfSuffix = map[string]string{
+var FlpConfSuffix = map[string]string{
 	ConfSingle:           "",
 	ConfKafkaIngestor:    "-kingestor",
-	ConfKafkaTransformer: "-ktransformer",
+	ConfKafkaTransformer: "-ktransform",
 }
 
 // PodConfigurationDigest is an annotation name to facilitate pod restart after
@@ -61,15 +61,15 @@ func newBuilder(ns string, desired *flowsv1alpha1.FlowCollectorFLP, desiredLoki
 	return builder{
 		namespace: ns,
 		labels: map[string]string{
-			"app":     constants.FLPName + flpConfSuffix[confKind],
+			"app":     constants.FLPName + FlpConfSuffix[confKind],
 			"version": version,
 		},
 		selector: map[string]string{
-			"app": constants.FLPName + flpConfSuffix[confKind],
+			"app": constants.FLPName + FlpConfSuffix[confKind],
 		},
 		desired:        desired,
 		desiredLoki:    desiredLoki,
-		confKindSuffix: flpConfSuffix[confKind],
+		confKindSuffix: FlpConfSuffix[confKind],
 	}
 }
 
@@ -349,7 +349,7 @@ func (b *builder) autoScaler() *ascv2.HorizontalPodAutoscaler {
 
 func buildAppLabel(confKind string) map[string]string {
 	return map[string]string{
-		"app": constants.FLPName + flpConfSuffix[confKind],
+		"app": constants.FLPName + FlpConfSuffix[confKind],
 	}
 }