From 0f53d3be6b2ed19d60c15f42c4ea200efe59e5d3 Mon Sep 17 00:00:00 2001
From: Nasar Khan <nasarak98@gmail.com>
Date: Thu, 2 Nov 2023 15:27:33 -0400
Subject: [PATCH] add kafka node affinity

---
 .../helpers/miq-components/kafka/kafka.go     |  2 ++
 .../api/v1alpha1/miqutils/node_affinity.go    | 29 +++++++++++++++++++
 2 files changed, 31 insertions(+)

diff --git a/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go b/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go
index 0af0af8d5..348e51a65 100644
--- a/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go
+++ b/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go
@@ -368,6 +368,8 @@ func KafkaCluster(cr *miqv1alpha1.ManageIQ, client client.Client, scheme *runtim
 
 	kafkaCRSpec = miqutilsv1alpha1.SetKafkaNodeAffinity(kafkaCRSpec, []string{"amd64"})
 
+	kafkaCRSpec = miqutilsv1alpha1.SetKafkaNodeAffinity(kafkaCRSpec, []string{"amd64"})
+
 	mutateFunc := func() error {
 		if err := controllerutil.SetControllerReference(cr, kafkaClusterCR, scheme); err != nil {
 			return err
diff --git a/manageiq-operator/api/v1alpha1/miqutils/node_affinity.go b/manageiq-operator/api/v1alpha1/miqutils/node_affinity.go
index 4b955a8ab..630c1f59a 100644
--- a/manageiq-operator/api/v1alpha1/miqutils/node_affinity.go
+++ b/manageiq-operator/api/v1alpha1/miqutils/node_affinity.go
@@ -59,3 +59,32 @@ func SetDeploymentNodeAffinity(deployment *appsv1.Deployment, client client.Clie
 		},
 	}
 }
+
+func SetKafkaNodeAffinity(kafkaCRSpec map[string]interface{}, archs []string) map[string]interface{} {
+	nodeAffinity := map[string]interface{}{
+		"nodeAffinity": map[string]interface{}{
+			"requiredDuringSchedulingIgnoredDuringExecution": map[string]interface{}{
+				"nodeSelectorTerms": []map[string]interface{}{
+					map[string]interface{}{
+						"matchExpressions": []map[string]interface{}{
+							map[string]interface{}{
+								"key":      "kubernetes.io/arch",
+								"operator": "In",
+								"values":   archs,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	kafkaPod := kafkaCRSpec["kafka"].(map[string]interface{})["template"].(map[string]interface{})["pod"].(map[string]interface{})
+	kafkaPod["affinity"] = nodeAffinity
+	zookeeperPod := kafkaCRSpec["zookeeper"].(map[string]interface{})["template"].(map[string]interface{})["pod"].(map[string]interface{})
+	zookeeperPod["affinity"] = nodeAffinity
+	operatorEntityPod := kafkaCRSpec["entityOperator"].(map[string]interface{})["template"].(map[string]interface{})["pod"].(map[string]interface{})
+	operatorEntityPod["affinity"] = nodeAffinity
+
+	return kafkaCRSpec
+}