fix: nutanix credentials Secrets owner refs (#711)

**What problem does this PR solve?**:
Add `ownerRef` to Nutanix CSI Secret with the corresponding Cluster.

CAPX already adds an ownerRef to the PC credentials Secret that CAPX
uses so did not add it there.

**Which issue(s) this PR fixes**:
Fixes https://jira.nutanix.com/browse/NCN-100575

**How Has This Been Tested?**:
<!--
Please describe the tests that you ran to verify your changes.
Provide output from the tests and any manual steps needed to replicate
the tests.
-->

**Special notes for your reviewer**:
<!--
Use this to provide any additional information to the reviewers.
This may include:
- Best way to review the PR.
- Where the author wants the most review attention on.
- etc.
-->

Author: dkoshkin

pkg/handlers/generic/lifecycle/ccm/nutanix/handler.go

@@ -98,6 +98,18 @@ func (p *provider) Apply(
 	// However, that would leave the credentials visible in the HelmChartProxy.
 	// Instead, we'll create the Secret on the remote cluster and reference it in the Helm values.
 	if clusterConfig.Addons.CCM.Credentials != nil {
+		err = lifecycleutils.EnsureOwnerRefForSecret(
+			ctx,
+			p.client,
+			clusterConfig.Addons.CCM.Credentials.SecretRef.Name,
+			cluster,
+		)
+		if err != nil {
+			return fmt.Errorf(
+				"error updating owner references on Nutanix CCM source Secret: %w",
+				err,
+			)
+		}
 		key := ctrlclient.ObjectKey{
 			Name:      defaultCredentialsSecretName,
 			Namespace: defaultHelmReleaseNamespace,

pkg/handlers/generic/lifecycle/csi/nutanix-csi/handler.go

@@ -97,11 +97,23 @@ func (n *NutanixCSI) Apply(
 	}
 
 	if provider.Credentials != nil {
+		err := lifecycleutils.EnsureOwnerRefForSecret(
+			ctx,
+			n.client,
+			provider.Credentials.SecretRef.Name,
+			&req.Cluster,
+		)
+		if err != nil {
+			return fmt.Errorf(
+				"error updating owner references on Nutanix CSI driver source Secret: %w",
+				err,
+			)
+		}
 		key := ctrlclient.ObjectKey{
 			Name:      defaultCredentialsSecretName,
 			Namespace: defaultStorageHelmReleaseNamespace,
 		}
-		err := lifecycleutils.CopySecretToRemoteCluster(
+		err = lifecycleutils.CopySecretToRemoteCluster(
 			ctx,
 			n.client,
 			provider.Credentials.SecretRef.Name,

pkg/handlers/generic/lifecycle/utils/secrets.go

@@ -11,6 +11,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/controllers/remote"
+	"sigs.k8s.io/cluster-api/util"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/k8s/client"
@@ -25,17 +26,7 @@ func CopySecretToRemoteCluster(
 	dstSecretKey ctrlclient.ObjectKey,
 	cluster *clusterv1.Cluster,
 ) error {
-	sourceSecret := &corev1.Secret{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: corev1.SchemeGroupVersion.String(),
-			Kind:       "Secret",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      srcSecretName,
-			Namespace: cluster.Namespace,
-		},
-	}
-	err := cl.Get(ctx, ctrlclient.ObjectKeyFromObject(sourceSecret), sourceSecret)
+	sourceSecret, err := getSecretForCluster(ctx, cl, srcSecretName, cluster)
 	if err != nil {
 		return err
 	}
@@ -71,3 +62,51 @@ func CopySecretToRemoteCluster(
 
 	return nil
 }
+
+// EnsureOwnerRefForSecret will ensure that the secretName Secret has an OwnerReference of the cluster.
+func EnsureOwnerRefForSecret(
+	ctx context.Context,
+	cl ctrlclient.Client,
+	secretName string,
+	cluster *clusterv1.Cluster,
+) error {
+	secret, err := getSecretForCluster(ctx, cl, secretName, cluster)
+	if err != nil {
+		return err
+	}
+
+	secret.OwnerReferences = util.EnsureOwnerRef(
+		secret.OwnerReferences,
+		metav1.OwnerReference{
+			APIVersion: clusterv1.GroupVersion.String(),
+			Kind:       cluster.Kind,
+			UID:        cluster.UID,
+			Name:       cluster.Name,
+		},
+	)
+
+	err = cl.Update(ctx, secret)
+	if err != nil {
+		return fmt.Errorf("failed to update Secret with owner references: %w", err)
+	}
+	return nil
+}
+
+func getSecretForCluster(
+	ctx context.Context,
+	c ctrlclient.Client,
+	secretName string,
+	cluster *clusterv1.Cluster,
+) (*corev1.Secret, error) {
+	secret := &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: corev1.SchemeGroupVersion.String(),
+			Kind:       "Secret",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: cluster.Namespace,
+		},
+	}
+	return secret, c.Get(ctx, ctrlclient.ObjectKeyFromObject(secret), secret)
+}

pkg/handlers/generic/lifecycle/utils/secrets_test.go

@@ -0,0 +1,131 @@
+// Copyright 2024 Nutanix. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package utils
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+var (
+	testSecret = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-secret",
+		},
+	}
+	testSecretWithOwnerRef = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-secret",
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: clusterv1.GroupVersion.String(),
+					Kind:       "Cluster",
+					Name:       testCluster.Name,
+					UID:        "12345",
+				},
+			},
+		},
+	}
+
+	testCluster = &clusterv1.Cluster{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: clusterv1.GroupVersion.String(),
+			Kind:       "Cluster",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-cluster",
+		},
+	}
+	anotherTestCluster = &clusterv1.Cluster{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: clusterv1.GroupVersion.String(),
+			Kind:       "Cluster",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "another-test-cluster",
+		},
+	}
+)
+
+func Test_EnsureOwnerRefForSecret(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name          string
+		client        client.Client
+		secretName    string
+		cluster       *clusterv1.Cluster
+		wantOwnerRefs int
+		wantErr       error
+	}{
+		{
+			name:          "add owner reference to Secret",
+			client:        buildFakeClient(t, testSecret, testCluster),
+			secretName:    testSecret.Name,
+			cluster:       testCluster,
+			wantOwnerRefs: 1,
+		},
+		{
+			name:          "update existing owner reference in Secret",
+			client:        buildFakeClient(t, testSecretWithOwnerRef, testCluster),
+			secretName:    testSecretWithOwnerRef.Name,
+			cluster:       testCluster,
+			wantOwnerRefs: 1,
+		},
+		{
+			name:          "add owner reference to Secret with an existing owner reference",
+			client:        buildFakeClient(t, testSecretWithOwnerRef, anotherTestCluster),
+			secretName:    testSecretWithOwnerRef.Name,
+			cluster:       anotherTestCluster,
+			wantOwnerRefs: 2,
+		},
+		{
+			name:       "should error on a missing Secret",
+			client:     buildFakeClient(t, testSecret, testCluster),
+			secretName: "missing-secret",
+			cluster:    testCluster,
+			wantErr:    errors.NewNotFound(corev1.Resource("secrets"), "missing-secret"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			err := EnsureOwnerRefForSecret(context.Background(), tt.client, tt.secretName, tt.cluster)
+			require.Equal(t, tt.wantErr, err)
+			if tt.wantErr != nil {
+				return
+			}
+			// verify that the owner reference was added
+			secret := &corev1.Secret{}
+			err = tt.client.Get(
+				context.Background(),
+				client.ObjectKey{Namespace: tt.cluster.Namespace, Name: tt.secretName},
+				secret,
+			)
+			assert.NoError(t, err, "failed to get updated secret")
+			assert.Len(t, secret.OwnerReferences, tt.wantOwnerRefs)
+		})
+	}
+}
+
+func buildFakeClient(t *testing.T, objs ...client.Object) client.Client {
+	t.Helper()
+	clientScheme := runtime.NewScheme()
+	utilruntime.Must(clientgoscheme.AddToScheme(clientScheme))
+	utilruntime.Must(clusterv1.AddToScheme(clientScheme))
+	return fake.NewClientBuilder().WithScheme(clientScheme).WithObjects(objs...).Build()
+}