Merge pull request #1005 from nasark/strimzi_kafka

Replace Bitnami with Strimzi Kafka

manageiq-operator/api/v1alpha1/helpers/miq-components/network_policies.go

@@ -227,7 +227,7 @@ func NetworkPolicyAllowKafka(cr *miqv1alpha1.ManageIQ, scheme *runtime.Scheme, c
 		addAppLabel(cr.Spec.AppName, &networkPolicy.ObjectMeta)
 		setIngressPolicyType(networkPolicy)
 
-		networkPolicy.Spec.PodSelector.MatchLabels = map[string]string{"name": "kafka"}
+		networkPolicy.Spec.PodSelector.MatchLabels = map[string]string{"strimzi.io/pod-name": "manageiq-kafka-0"}
 
 		pod := orchestratorPod(*c)
 		if pod == nil {

manageiq-operator/api/v1alpha1/helpers/miq-components/orchestrator.go

@@ -111,46 +111,40 @@ func orchestratorObjectName(cr *miqv1alpha1.ManageIQ) string {
 	return cr.Spec.AppName + "-orchestrator"
 }
 
-func addMessagingEnv(cr *miqv1alpha1.ManageIQ, c *corev1.Container) {
+func addMessagingEnv(cr *miqv1alpha1.ManageIQ, c *corev1.Container, client client.Client) {
 	if !*cr.Spec.DeployMessagingService {
 		return
 	}
 
 	messagingEnv := []corev1.EnvVar{
 		corev1.EnvVar{
-			Name: "MESSAGING_HOSTNAME",
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecretName(cr)},
-					Key:                  "hostname",
-				},
-			},
+			Name:  "MESSAGING_HOSTNAME",
+			Value: cr.Spec.AppName + "-kafka-bootstrap",
 		},
 		corev1.EnvVar{
 			Name: "MESSAGING_PASSWORD",
 			ValueFrom: &corev1.EnvVarSource{
 				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecretName(cr)},
+					LocalObjectReference: corev1.LocalObjectReference{Name: cr.Spec.AppName + "-user"},
 					Key:                  "password",
 				},
 			},
 		},
 		corev1.EnvVar{
 			Name:  "MESSAGING_PORT",
-			Value: "9092",
+			Value: "9093",
 		},
 		corev1.EnvVar{
 			Name:  "MESSAGING_TYPE",
 			Value: "kafka",
 		},
 		corev1.EnvVar{
-			Name: "MESSAGING_USERNAME",
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecretName(cr)},
-					Key:                  "username",
-				},
-			},
+			Name:  "MESSAGING_USERNAME",
+			Value: cr.Spec.AppName + "-user",
+		},
+		corev1.EnvVar{
+			Name:  "MESSAGING_SASL_MECHANISM",
+			Value: "SCRAM-SHA-512",
 		},
 	}
 
@@ -245,7 +239,7 @@ func OrchestratorDeployment(cr *miqv1alpha1.ManageIQ, scheme *runtime.Scheme, cl
 		},
 	}
 
-	addMessagingEnv(cr, &container)
+	addMessagingEnv(cr, &container, client)
 	err = addResourceReqs(cr.Spec.OrchestratorMemoryLimit, cr.Spec.OrchestratorMemoryRequest, cr.Spec.OrchestratorCpuLimit, cr.Spec.OrchestratorCpuRequest, &container)
 	if err != nil {
 		return nil, nil, err
@@ -305,6 +299,15 @@ func OrchestratorDeployment(cr *miqv1alpha1.ManageIQ, scheme *runtime.Scheme, cl
 			deployment.Spec.Template.Spec.Containers[0].Env = addOrUpdateEnvVar(deployment.Spec.Template.Spec.Containers[0].Env, corev1.EnvVar{Name: "UI_SSL_SECRET_NAME", Value: cr.Spec.InternalCertificatesSecret})
 		}
 
+		messagingCAPath := ""
+		if certSecret := InternalCertificatesSecret(cr, client); certSecret.Data["root_crt"] != nil && certSecret.Data["root_key"] != nil {
+			messagingCAPath = "/etc/pki/ca-trust/source/anchors/root.crt"
+		} else {
+			messagingCAPath = "/etc/pki/ca-trust/source/anchors/ca.crt"
+		}
+
+		deployment.Spec.Template.Spec.Containers[0].Env = addOrUpdateEnvVar(deployment.Spec.Template.Spec.Containers[0].Env, corev1.EnvVar{Name: "MESSAGING_SSL_CA", Value: messagingCAPath})
+
 		volumeMount := corev1.VolumeMount{Name: "encryption-key", MountPath: "/run/secrets/manageiq/application", ReadOnly: true}
 		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = addOrUpdateVolumeMount(deployment.Spec.Template.Spec.Containers[0].VolumeMounts, volumeMount)
 
@@ -359,5 +362,10 @@ func addInternalRootCertificate(cr *miqv1alpha1.ManageIQ, d *appsv1.Deployment,
 			d.Spec.Template.Spec.Containers[0].Env = addOrUpdateEnvVar(d.Spec.Template.Spec.Containers[0].Env, corev1.EnvVar{Name: "MEMCACHED_ENABLE_SSL", Value: "true"})
 			d.Spec.Template.Spec.Containers[0].Env = addOrUpdateEnvVar(d.Spec.Template.Spec.Containers[0].Env, corev1.EnvVar{Name: "MEMCACHED_SSL_CA", Value: "/etc/pki/ca-trust/source/anchors/root.crt"})
 		}
+	} else {
+		volumeMount := corev1.VolumeMount{Name: "messaging-certificate", MountPath: "/etc/pki/ca-trust/source/anchors", ReadOnly: true}
+		d.Spec.Template.Spec.Containers[0].VolumeMounts = addOrUpdateVolumeMount(d.Spec.Template.Spec.Containers[0].VolumeMounts, volumeMount)
+		secretVolumeSource := corev1.SecretVolumeSource{SecretName: "manageiq-cluster-ca-cert", Items: []corev1.KeyToPath{corev1.KeyToPath{Key: "ca.crt", Path: "ca.crt"}}}
+		d.Spec.Template.Spec.Volumes = addOrUpdateVolume(d.Spec.Template.Spec.Volumes, corev1.Volume{Name: "messaging-certificate", VolumeSource: corev1.VolumeSource{Secret: &secretVolumeSource}})
 	}
 }

manageiq-operator/api/v1alpha1/miqutils/find.go

@@ -2,8 +2,12 @@ package miqutils
 
 import (
 	"context"
+	olmv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -31,3 +35,32 @@ func FindDeploymentByName(client client.Client, namespace string, name string) *
 
 	return deployment
 }
+
+func FindSecretByName(client client.Client, namespace string, name string) *corev1.Secret {
+	secretKey := types.NamespacedName{Namespace: namespace, Name: name}
+	secret := &corev1.Secret{}
+	client.Get(context.TODO(), secretKey, secret)
+
+	return secret
+}
+
+func FindKafka(client client.Client, scheme *runtime.Scheme, namespace string, name string) *unstructured.Unstructured {
+	kafkaKey := types.NamespacedName{Namespace: namespace, Name: name}
+	kafka := &unstructured.Unstructured{}
+	kafka.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   "kafka.strimzi.io",
+		Kind:    "Kafka",
+		Version: "v1beta2",
+	})
+	client.Get(context.TODO(), kafkaKey, kafka)
+
+	return kafka
+}
+
+func FindCatalogSourceByName(client client.Client, namespace string, name string) *olmv1alpha1.CatalogSource {
+	catalogSourceKey := types.NamespacedName{Namespace: namespace, Name: name}
+	catalogSource := &olmv1alpha1.CatalogSource{}
+	client.Get(context.TODO(), catalogSourceKey, catalogSource)
+
+	return catalogSource
+}

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

manageiq-operator/cmd/main.go

@@ -38,6 +38,8 @@ import (
 
 	manageiqv1alpha1 "github.com/ManageIQ/manageiq-pods/manageiq-operator/api/v1alpha1"
 	"github.com/ManageIQ/manageiq-pods/manageiq-operator/internal/controller"
+	olmv1 "github.com/operator-framework/api/pkg/operators/v1"
+	olmv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -53,6 +55,9 @@ func init() {
 	utilruntime.Must(manageiqv1alpha1.AddToScheme(scheme))
 	//+kubebuilder:scaffold:scheme
 
+	utilruntime.Must(olmv1alpha1.SchemeBuilder.AddToScheme(scheme))
+	utilruntime.Must(olmv1.SchemeBuilder.AddToScheme(scheme))
+
 	utilruntime.Must(routev1.AddToScheme(scheme))
 }
 

manageiq-operator/config/rbac/role.yaml

@@ -76,6 +76,20 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - kafka.strimzi.io
+  resources:
+  - kafkas
+  - kafkatopics
+  - kafkausers
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - manageiq.org
   resources:
@@ -122,6 +136,19 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - operators.coreos.com
+  resources:
+  - operatorgroups
+  - subscriptions
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - rbac.authorization.k8s.io
   resources:

manageiq-operator/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.30.0
 	github.com/openshift/api v0.0.0-20231123212421-7955d3da79e8
+	github.com/operator-framework/api v0.20.0
 	k8s.io/api v0.28.4
 	k8s.io/apimachinery v0.28.4
 	k8s.io/client-go v0.28.4
@@ -14,6 +15,7 @@ require (
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
@@ -45,6 +47,7 @@ require (
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/sirupsen/logrus v1.9.2 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect

manageiq-operator/go.sum

@@ -1,5 +1,7 @@
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
+github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
@@ -106,6 +108,8 @@ github.com/onsi/gomega v1.30.0 h1:hvMK7xYz4D3HapigLTeGdId/NcfQx1VHMJc60ew99+8=
 github.com/onsi/gomega v1.30.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ=
 github.com/openshift/api v0.0.0-20231123212421-7955d3da79e8 h1:JfXWa9HQc3GCMQeRxL3WLOW0eTYLoNtOFRajLjua/S0=
 github.com/openshift/api v0.0.0-20231123212421-7955d3da79e8/go.mod h1:qNtV0315F+f8ld52TLtPvrfivZpdimOzTi3kn9IVbtU=
+github.com/operator-framework/api v0.20.0 h1:A2YCRhr+6s0k3pRJacnwjh1Ue8BqjIGuQ2jvPg9XCB4=
+github.com/operator-framework/api v0.20.0/go.mod h1:rXPOhrQ6mMeXqCmpDgt1ALoar9ZlHL+Iy5qut9R99a4=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -120,13 +124,16 @@ github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/sirupsen/logrus v1.9.2 h1:oxx1eChJGI6Uks2ZC4W1zpLlVgqB8ner4EuQwV4Ik1Y=
+github.com/sirupsen/logrus v1.9.2/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
@@ -179,6 +186,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=