Merge pull request #440 from alenkacz/av/set-ownership

operator/controller: Set ownerReference for created objects

Author: emaxerrno

src/go/k8s/controllers/redpanda/cluster_controller.go

@@ -7,7 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0
 
-// Package redpanda is responsible for Reconcile the redpanda.vectorized.io Custom Resource Definition
+// Package redpanda contains reconciliation logic for redpanda.vectorized.io CRD
 package redpanda
 
 import (
@@ -32,6 +32,7 @@ import (
 	"k8s.io/utils/pointer"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
 const (
@@ -91,7 +92,7 @@ func (r *ClusterReconciler) Reconcile(
 	if errors.IsNotFound(err) {
 		log.V(debugLevel).Info("Creating headless service")
 
-		if err = r.createHeadlessService(ctx, &redpandaCluster); err != nil {
+		if err = r.createHeadlessService(ctx, &redpandaCluster, r.Scheme); err != nil {
 			log.Error(err, "Failed to create new service",
 				"Service.Namespace", redpandaCluster.Namespace,
 				"Service.Name", redpandaCluster.Name)
@@ -112,7 +113,7 @@ func (r *ClusterReconciler) Reconcile(
 	if errors.IsNotFound(err) {
 		log.V(debugLevel).Info("Creating seed redpanda ConfigMap")
 
-		if err = r.createBootstrapConfigMap(ctx, &redpandaCluster); err != nil {
+		if err = r.createBootstrapConfigMap(ctx, &redpandaCluster, r.Scheme); err != nil {
 			log.Error(err, "Failed to create new seed redpanda ConfigMap",
 				"Configmap.Namespace", redpandaCluster.Namespace,
 				"Configmap.Name", redpandaCluster.Name+seedSuffix)
@@ -133,7 +134,7 @@ func (r *ClusterReconciler) Reconcile(
 	if errors.IsNotFound(err) {
 		log.V(debugLevel).Info("Creating bootstrap StatefulSet")
 
-		if err = r.createBootstrapStatefulSet(ctx, &redpandaCluster); err != nil {
+		if err = r.createBootstrapStatefulSet(ctx, &redpandaCluster, r.Scheme); err != nil {
 			log.Error(err, "Failed to create new bootstrap StatefulSet",
 				"Configmap.Namespace", redpandaCluster.Namespace, "StatefulSet.Name", redpandaCluster.Name+seedSuffix)
 
@@ -181,9 +182,11 @@ func (r *ClusterReconciler) Reconcile(
 }
 
 func (r *ClusterReconciler) createHeadlessService(
-	ctx context.Context, clusterSpec *redpandav1alpha1.Cluster,
+	ctx context.Context,
+	clusterSpec *redpandav1alpha1.Cluster,
+	scheme *runtime.Scheme,
 ) error {
-	return r.Create(ctx, &corev1.Service{
+	svc := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:	clusterSpec.Namespace,
 			Name:		clusterSpec.Name,
@@ -201,11 +204,18 @@ func (r *ClusterReconciler) createHeadlessService(
 			},
 			Selector:	clusterSpec.Labels,
 		},
-	})
+	}
+	err := controllerutil.SetControllerReference(clusterSpec, svc, scheme)
+	if err != nil {
+		return err
+	}
+	return r.Create(ctx, svc)
 }
 
 func (r *ClusterReconciler) createBootstrapConfigMap(
-	ctx context.Context, cluster *redpandav1alpha1.Cluster,
+	ctx context.Context,
+	cluster *redpandav1alpha1.Cluster,
+	scheme *runtime.Scheme,
 ) error {
 	cfg := config.Default()
 	cfg.Redpanda = copyConfig(&cluster.Spec.Configuration, &cfg.Redpanda)
@@ -225,7 +235,7 @@ func (r *ClusterReconciler) createBootstrapConfigMap(
 		return err
 	}
 
-	return r.Create(ctx, &corev1.ConfigMap{
+	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:	cluster.Namespace,
 			Name:		cluster.Name + seedSuffix,
@@ -234,7 +244,13 @@ func (r *ClusterReconciler) createBootstrapConfigMap(
 		Data: map[string]string{
 			"redpanda.yaml": string(cfgBytes),
 		},
-	})
+	}
+	err = controllerutil.SetControllerReference(cluster, cm, scheme)
+	if err != nil {
+		return err
+	}
+
+	return r.Create(ctx, cm)
 }
 
 func copyConfig(
@@ -276,14 +292,16 @@ func copyConfig(
 
 // nolint:funlen // The definition needs further refinement
 func (r *ClusterReconciler) createBootstrapStatefulSet(
-	ctx context.Context, cluster *redpandav1alpha1.Cluster,
+	ctx context.Context,
+	cluster *redpandav1alpha1.Cluster,
+	scheme *runtime.Scheme,
 ) error {
 	memory, exist := cluster.Spec.Resources.Limits["memory"]
 	if !exist {
 		memory = resource.MustParse("2Gi")
 	}
 
-	return r.Create(ctx, &appsv1.StatefulSet{
+	ss := &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:	cluster.Namespace,
 			Name:		cluster.Name + seedSuffix,
@@ -405,7 +423,14 @@ func (r *ClusterReconciler) createBootstrapStatefulSet(
 				},
 			},
 		},
-	})
+	}
+
+	err := controllerutil.SetControllerReference(cluster, ss, scheme)
+	if err != nil {
+		return err
+	}
+
+	return r.Create(ctx, ss)
 }
 
 // SetupWithManager sets up the controller with the Manager.

src/go/k8s/controllers/redpanda/cluster_controller_test.go

@@ -85,7 +85,8 @@ var _ = Describe("RedPandaCluster controller", func() {
 				err := k8sClient.Get(context.Background(), key, &svc)
 				return err == nil &&
 					svc.Spec.ClusterIP == corev1.ClusterIPNone &&
-					svc.Spec.Ports[0].Port == kafkaPort
+					svc.Spec.Ports[0].Port == kafkaPort &&
+					validOwner(redpandaCluster, svc.OwnerReferences)
 			}, timeout, interval).Should(BeTrue())
 
 			By("Creating Configmap with the redpanda configuration")
@@ -96,7 +97,8 @@ var _ = Describe("RedPandaCluster controller", func() {
 					return false
 				}
 				_, exist := cm.Data[redpandaConfigurationFile]
-				return exist
+				return exist &&
+					validOwner(redpandaCluster, cm.OwnerReferences)
 			}, timeout, interval).Should(BeTrue())
 
 			By("Creating StatefulSet")
@@ -105,11 +107,22 @@ var _ = Describe("RedPandaCluster controller", func() {
 				err := k8sClient.Get(context.Background(), seedKey, &sts)
 				return err == nil &&
 					*sts.Spec.Replicas == replicas &&
-					sts.Spec.Template.Spec.Containers[0].Image == "vectorized/redpanda:"+redpandaContainerTag
+					sts.Spec.Template.Spec.Containers[0].Image == "vectorized/redpanda:"+redpandaContainerTag &&
+					validOwner(redpandaCluster, sts.OwnerReferences)
 			}, timeout, interval).Should(BeTrue())
 
 			Expect(sts.Spec.Template.Spec.Containers[0].Resources.Requests).Should(Equal(resources))
 			Expect(sts.Spec.Template.Spec.Containers[0].Resources.Limits).Should(Equal(resources))
 		})
 	})
 })
+
+func validOwner(
+	cluster *v1alpha1.Cluster, owners []metav1.OwnerReference,
+) bool {
+	if len(owners) != 1 {
+		return false
+	}
+	owner := owners[0]
+	return owner.Name == cluster.Name && owner.Controller != nil && *owner.Controller
+}