From 68d060266da089078fe9e86c4ebf30721a0df9f6 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 bd823d5c..b7155711 100644
--- a/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go
+++ b/manageiq-operator/api/v1alpha1/helpers/miq-components/kafka/kafka.go
@@ -302,6 +302,8 @@ func KafkaCluster(cr *miqv1alpha1.ManageIQ, client client.Client, scheme *runtim
 	zookeeperResourceLimits["memory"] = "512Mi"
 	zookeeperResourceLimits["cpu"] = "250m"
 
+	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 4b955a8a..630c1f59 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
+}