Merge pull request ManageIQ#1101 from nasark/remove_health_check_topic

Remove creation of manageiq.liveness-check topic
miq-bot · May 24, 2024 · 3ee8237 · 3ee8237
2 parents 1dcbe4b + 73b87c5
commit 3ee8237
Show file tree

Hide file tree

Showing 4 changed files with 48 additions and 4 deletions.
diff --git a/...rator/api/v1alpha1/helpers/cr_migration/20240508124600_remove_kafka_health_check_topic.go b/...rator/api/v1alpha1/helpers/cr_migration/20240508124600_remove_kafka_health_check_topic.go
@@ -0,0 +1,27 @@
+package cr_migration
+
+import (
+	"context"
+
+	miqv1alpha1 "github.com/ManageIQ/manageiq-pods/manageiq-operator/api/v1alpha1"
+	miqutilsv1alpha1 "github.com/ManageIQ/manageiq-pods/manageiq-operator/api/v1alpha1/miqutils"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func migrate20240508124600(cr *miqv1alpha1.ManageIQ, client client.Client, scheme *runtime.Scheme) *miqv1alpha1.ManageIQ {
+	migrationId := "20240508124600"
+	for _, migration := range cr.Spec.MigrationsRan {
+		if migration == migrationId {
+			return cr
+		}
+	}
+
+	if topic := miqutilsv1alpha1.FindKafkaTopic(client, scheme, cr.Namespace, "manageiq.liveness-check", "kafka.strimzi.io"); topic != nil {
+		client.Delete(context.TODO(), topic)
+	}
+
+	cr.Spec.MigrationsRan = append(cr.Spec.MigrationsRan, migrationId)
+
+	return cr
+}
diff --git a/manageiq-operator/api/v1alpha1/helpers/cr_migration/cr_migration.go b/manageiq-operator/api/v1alpha1/helpers/cr_migration/cr_migration.go
@@ -2,13 +2,16 @@ package cr_migration
 
 import (
 	miqv1alpha1 "github.com/ManageIQ/manageiq-pods/manageiq-operator/api/v1alpha1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
-func Migrate(cr *miqv1alpha1.ManageIQ) (*miqv1alpha1.ManageIQ, controllerutil.MutateFn) {
+func Migrate(cr *miqv1alpha1.ManageIQ, client client.Client, scheme *runtime.Scheme) (*miqv1alpha1.ManageIQ, controllerutil.MutateFn) {
 	f := func() error {
 		cr = migrate20210503163000(cr)
 		cr = migrate20210504113000(cr)
+		cr = migrate20240508124600(cr, client, scheme)
 
 		return nil
 	}

diff --git a/manageiq-operator/api/v1alpha1/miqutils/find.go b/manageiq-operator/api/v1alpha1/miqutils/find.go
@@ -2,6 +2,8 @@ package miqutils
 
 import (
 	"context"
+	"strings"
+
 	olmv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -11,7 +13,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"strings"
 )
 
 func FindPodByName(client client.Client, namespace string, name string) *corev1.Pod {
@@ -59,6 +60,19 @@ func FindKafka(client client.Client, scheme *runtime.Scheme, namespace string, n
 	return kafka
 }
 
+func FindKafkaTopic(client client.Client, scheme *runtime.Scheme, namespace string, name string, group string) *unstructured.Unstructured {
+	kafkaTopicKey := types.NamespacedName{Namespace: namespace, Name: name}
+	kafkaTopic := &unstructured.Unstructured{}
+	kafkaTopic.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   group,
+		Kind:    "KafkaTopic",
+		Version: "v1beta2",
+	})
+	client.Get(context.TODO(), kafkaTopicKey, kafkaTopic)
+
+	return kafkaTopic
+}
+
 func FindCatalogSourceByName(client client.Client, namespace string, name string) *olmv1alpha1.CatalogSource {
 	catalogSourceKey := types.NamespacedName{Namespace: namespace, Name: name}
 	catalogSource := &olmv1alpha1.CatalogSource{}

diff --git a/manageiq-operator/internal/controller/manageiq_controller.go b/manageiq-operator/internal/controller/manageiq_controller.go
@@ -586,7 +586,7 @@ func (r *ManageIQReconciler) generateKafkaResources(cr *miqv1alpha1.ManageIQ) er
 		logger.Info("Kafka User has been reconciled", "result", result)
 	}
 
-	topics := []string{"manageiq.liveness-check", "manageiq.ems", "manageiq.ems-events", "manageiq.ems-inventory", "manageiq.metrics"}
+	topics := []string{"manageiq.ems", "manageiq.ems-events", "manageiq.ems-inventory", "manageiq.metrics"}
 	for i := 0; i < len(topics); i++ {
 		kafkaTopicCR, mutateFunc := miqkafka.KafkaTopic(cr, r.Scheme, topics[i])
 		if result, err := controllerutil.CreateOrUpdate(context.TODO(), r.Client, kafkaTopicCR, mutateFunc); err != nil {
@@ -762,7 +762,7 @@ func (r *ManageIQReconciler) generateSecrets(cr *miqv1alpha1.ManageIQ) error {
 }
 
 func (r *ManageIQReconciler) migrateCR(cr *miqv1alpha1.ManageIQ) error {
-	manageiq, mutateFunc := cr_migration.Migrate(cr)
+	manageiq, mutateFunc := cr_migration.Migrate(cr, r.Client, r.Scheme)
 	if result, err := controllerutil.CreateOrUpdate(context.TODO(), r.Client, manageiq, mutateFunc); err != nil {
 		return err
 	} else if result != controllerutil.OperationResultNone {