add kafka node affinity

Author: nasark

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

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
+}