Add owner reference to the resources made by rotator to delete them when read replica resources are deleted

Makefile.d/k8s.mk

@@ -153,6 +153,45 @@ k8s/vald-helm-operator/delete:
 	kubectl delete -f $(TEMP_DIR)/vald-helm-operator/crds
 	rm -rf $(TEMP_DIR)
 
+.PHONY: k8s/vald-readreplica/deploy
+## deploy vald-readreplica to k8s
+k8s/vald-readreplica/deploy:
+	helm template \
+	    --values $(HELM_VALUES) \
+	    --set defaults.image.tag=$(VERSION) \
+	    --set agent.image.repository=$(CRORG)/$(AGENT_IMAGE) \
+	    --set agent.sidecar.image.repository=$(CRORG)/$(AGENT_SIDECAR_IMAGE) \
+	    --set discoverer.image.repository=$(CRORG)/$(DISCOVERER_IMAGE) \
+	    --set gateway.filter.image.repository=$(CRORG)/$(FILTER_GATEWAY_IMAGE) \
+	    --set gateway.lb.image.repository=$(CRORG)/$(LB_GATEWAY_IMAGE) \
+	    --set manager.index.image.repository=$(CRORG)/$(MANAGER_INDEX_IMAGE) \
+	    --set manager.index.creator.image.repository=$(CRORG)/$(INDEX_CREATION_IMAGE) \
+	    --set manager.index.saver.image.repository=$(CRORG)/$(INDEX_SAVE_IMAGE) \
+	    $(HELM_EXTRA_OPTIONS) \
+	    --output-dir $(TEMP_DIR) \
+	    charts/vald-readreplica
+	kubectl apply -f $(TEMP_DIR)/vald-readreplica/templates
+	sleep 2
+	kubectl wait --for=condition=ready pod -l app.kubernetes.io/name=vald-readreplica --timeout=600s
+
+.PHONY: k8s/vald-readreplica/delete
+## delete vald-helm-operator from k8s
+k8s/vald-readreplica/delete:
+	helm template \
+	    --values $(HELM_VALUES) \
+	    --set defaults.image.tag=$(VERSION) \
+	    --set agent.image.repository=$(CRORG)/$(AGENT_IMAGE) \
+	    --set agent.sidecar.image.repository=$(CRORG)/$(AGENT_SIDECAR_IMAGE) \
+	    --set discoverer.image.repository=$(CRORG)/$(DISCOVERER_IMAGE) \
+	    --set gateway.filter.image.repository=$(CRORG)/$(FILTER_GATEWAY_IMAGE) \
+	    --set gateway.lb.image.repository=$(CRORG)/$(LB_GATEWAY_IMAGE) \
+	    --set manager.index.image.repository=$(CRORG)/$(MANAGER_INDEX_IMAGE) \
+	    --output-dir $(TEMP_DIR) \
+	    charts/vald-readreplica
+	kubectl delete -f $(TEMP_DIR)/vald-readreplica/templates
+	kubectl wait --for=delete pod -l app.kubernetes.io/name=vald-readreplica --timeout=600s
+	rm -rf $(TEMP_DIR)
+
 .PHONY: k8s/vr/deploy
 ## deploy ValdRelease resource to k8s
 k8s/vr/deploy: \

pkg/index/job/readreplica/rotate/service/rotator.go

@@ -29,6 +29,7 @@
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 )
 
 const (
@@ -109,12 +110,20 @@
 }
 
 func (r *rotator) rotate(ctx context.Context) error {
-	newSnap, oldSnap, err := r.createSnapshot(ctx)
+	// get deployment here to pass to create methods of snapshot and pvc
+	// and put it as owner reference of them so that they will be deleted when the deployment is deleted
+	deployment, err := r.getDeployment(ctx)
 	if err != nil {
+		log.Errorf("failed to get Deployment.")
 		return err
 	}
 
-	newPvc, oldPvc, err := r.createPVC(ctx, newSnap.GetName())
+	newSnap, oldSnap, err := r.createSnapshot(ctx, deployment)
+	if err != nil {
+		return err
+	}
+
+	newPvc, oldPvc, err := r.createPVC(ctx, newSnap.GetName(), deployment)
 	if err != nil {
 		log.Errorf("failed to create PVC. removing the new snapshot(%s)...", newSnap.GetName())
 		if dserr := r.deleteSnapshot(ctx, newSnap); dserr != nil {
@@ -123,7 +132,7 @@
 		return err
 	}
 
-	err = r.updateDeployment(ctx, newPvc.GetName())
+	err = r.updateDeployment(ctx, newPvc.GetName(), deployment)
 	if err != nil {
 		log.Errorf("failed to update Deployment. removing the new snapshot(%s) and pvc(%s)...", newSnap.GetName(), newPvc.GetName())
 		if dperr := r.deletePVC(ctx, newPvc); dperr != nil {
@@ -148,7 +157,7 @@
 	return nil
 }
 
-func (r *rotator) createSnapshot(ctx context.Context) (newSnap, oldSnap *client.VolumeSnapshot, err error) {
+func (r *rotator) createSnapshot(ctx context.Context, deployment *appsv1.Deployment) (newSnap, oldSnap *client.VolumeSnapshot, err error) {
 	list := snapshotv1.VolumeSnapshotList{}
 	if err := r.client.List(ctx, &list, &r.listOpts); err != nil {
 		return nil, nil, fmt.Errorf("failed to get snapshot: %w", err)
@@ -168,6 +177,15 @@
 			Name:      fmt.Sprintf("%s%d", newNameBase, time.Now().Unix()),
 			Namespace: cur.GetNamespace(),
 			Labels:    cur.GetObjectMeta().GetLabels(),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Deployment",
+					Name:       deployment.GetName(),
+					UID:        deployment.GetUID(),
+					Controller: pointer.Bool(true),
+				},
+			},
 		},
 		Spec: cur.Spec,
 	}
@@ -183,7 +201,7 @@
 	return newSnap, oldSnap, nil
 }
 
-func (r *rotator) createPVC(ctx context.Context, newSnapShot string) (newPvc, oldPvc *v1.PersistentVolumeClaim, err error) {
+func (r *rotator) createPVC(ctx context.Context, newSnapShot string, deployment *appsv1.Deployment) (newPvc, oldPvc *v1.PersistentVolumeClaim, err error) {
 	list := v1.PersistentVolumeClaimList{}
 	if err := r.client.List(ctx, &list, &r.listOpts); err != nil {
 		return nil, nil, fmt.Errorf("failed to get PVC: %w", err)
@@ -205,6 +223,15 @@
 			Name:      fmt.Sprintf("%s%d", newNameBase, time.Now().Unix()),
 			Namespace: cur.GetNamespace(),
 			Labels:    cur.GetObjectMeta().GetLabels(),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Deployment",
+					Name:       deployment.GetName(),
+					UID:        deployment.GetUID(),
+					Controller: pointer.Bool(true),
+				},
+			},
 		},
 		Spec: v1.PersistentVolumeClaimSpec{
 			AccessModes: cur.Spec.AccessModes,
@@ -227,16 +254,19 @@
 	return newPvc, oldPvc, nil
 }
 
-func (r *rotator) updateDeployment(ctx context.Context, newPVC string) error {
+func (r *rotator) getDeployment(ctx context.Context) (*appsv1.Deployment, error) {
 	list := appsv1.DeploymentList{}
 	if err := r.client.List(ctx, &list, &r.listOpts); err != nil {
-		return fmt.Errorf("failed to get deployment through client: %w", err)
+		return nil, fmt.Errorf("failed to get deployment through client: %w", err)
 	}
 	if len(list.Items) == 0 {
-		return fmt.Errorf("no deployment found")
+		return nil, fmt.Errorf("no deployment found")
 	}
 
-	deployment := list.Items[0]
+	return &list.Items[0], nil
+}
+
+func (r *rotator) updateDeployment(ctx context.Context, newPVC string, deployment *appsv1.Deployment) error {
 	if deployment.Spec.Template.ObjectMeta.Annotations == nil {
 		deployment.Spec.Template.ObjectMeta.Annotations = map[string]string{}
 	}
@@ -251,7 +281,7 @@
 	log.Infof("updating deployment(%s)...", deployment.GetName())
 	log.Debugf("deployment detail: %#v", deployment)
 
-	if err := r.client.Update(ctx, &deployment); err != nil {
+	if err := r.client.Update(ctx, deployment); err != nil {
 		return fmt.Errorf("failed to update deployment: %w", err)
 	}