/*
Copyright 2023.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package vmoperation

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	vm "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	"github.com/go-logr/logr"
	promapi "github.com/prometheus/client_golang/api"
	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"github.com/prometheus/common/model"
	o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/structs"
	appsv1 "k8s.io/api/apps/v1"
	batchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/utils/ptr"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

var (
	requeue               = reconcile.Result{Requeue: true}
	storageClassVMStorage = "vmcluster-storage"
	interval              = 5 * time.Second
	timeout               = 30 * time.Second
)

const (
	defaultBaseDelay = 5 * time.Second
	defaultMaxDelay  = 2 * time.Minute

	dot                            = "."
	PendingReason                  = "Wait for the scheduled time"
	normalEvent                    = "Normal"
	warningEvent                   = "Warning"
	vmstoragePVCPrefix             = "vmstorage-db-"
	copyPVCSuffix                  = "-copy"
	vmstorageSTSPrefix             = "vmstorage-"
	vminsertPrefix                 = "vminsert-"
	vminsertParamStorageNode       = "-storageNode="
	Gi                       int64 = 1024 * 1024 * 1024
	jobLabelKey                    = "job-name"
	VolumeUsagePromQLTmpl          = "ceil(max_over_time(kubelet_volume_stats_used_bytes{persistentvolumeclaim=\"%s\"}[1d])/1024/1024/1024/0.85)"
)

// VMOperationReconciler reconciles a VMOperation object
type VMOperationReconciler struct {
	client.Client
	promCli       promv1.API
	ServiceInfo   structs.ServiceInfo
	eventRecorder record.EventRecorder
}

func NewVMOperationReconciler(client client.Client, serviceInfo structs.ServiceInfo, eventRecorder record.EventRecorder) (*VMOperationReconciler, error) {
	cli, err := newPromClient(serviceInfo.SelfMonitorEndpoint)
	if err != nil {
		return nil, err
	}
	return &VMOperationReconciler{
		Client:        client,
		promCli:       cli,
		ServiceInfo:   serviceInfo,
		eventRecorder: eventRecorder,
	}, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *VMOperationReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&o11yv1.VMOperation{}).
		WithOptions(
			controller.Options{
				RateLimiter: workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](defaultBaseDelay, defaultMaxDelay),
			}).
		Complete(r)
}

func (r *VMOperationReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, _ error) {
	logger := log.FromContext(ctx)
	logger.Info("Reconciling VMOperation")
	defer func() { logger.Info("Finish reconcile", "Requeue", res.Requeue, "RequeueAfter", res.RequeueAfter) }()

	op := &o11yv1.VMOperation{}
	if err := r.Get(ctx, req.NamespacedName, op); client.IgnoreNotFound(err) != nil {
		LogAndEmitEvent(logger, r.eventRecorder, op, warningEvent, "Failed to get the VMOperation CR", err)
		return ctrl.Result{}, nil
	}

	// Handle the deletion reconciliation loop
	if op.GetDeletionTimestamp() != nil {
		if op.Status.Phase == o11yv1.VMOperationPhaseRunning {
			// We need to wait for the VMOperation to be completed or failed before deleting it
			logger.Info("wait for VMOperation to complete/fail before deleting it")
		} else {
			controllerutil.RemoveFinalizer(op, o11yv1.VMOperationFinalizer)
			if err := r.Update(ctx, op); err != nil {
				logger.Error(err, "failed to remove the finalizer from VMOperation")
				return ctrl.Result{}, err
			}
			logger.Info("remove the finalizer for VMOperation")
			return ctrl.Result{}, nil
		}
	}
	// Ensure finalizer
	if !controllerutil.ContainsFinalizer(op, o11yv1.VMOperationFinalizer) {
		controllerutil.AddFinalizer(op, o11yv1.VMOperationFinalizer)
		if err := r.Update(ctx, op); err != nil {
			logger.Error(err, "Add Finalizer failed")
			return requeue, err
		}
	}

	if op.Spec.Paused {
		LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, "VMOperation is paused, skip processing", nil)
		return ctrl.Result{}, nil
	}

	switch op.Status.Phase {
	case "":
		op.Status.Phase = o11yv1.VMOperationPhasePending
		op.Status.SucceededSteps = make(o11yv1.Steps, 0)
		op.Status.NextSteps = make(o11yv1.Steps, 0, op.Spec.Steps.Size())
		return ctrl.Result{}, r.Status().Update(ctx, op)
	case o11yv1.VMOperationPhasePending:
		return r.reconcilePending(ctx, op)
	case o11yv1.VMOperationPhaseRunning:
		return r.reconcileRunning(ctx, op)
	case o11yv1.VMOperationPhaseFailed, o11yv1.VMOperationPhaseCompleted:
		return r.reconcileFinished(ctx, op)
	default:
		logger.Info(fmt.Sprintf("skip due to unknown phase %s", op.Status.Phase))
		return ctrl.Result{}, nil
	}
}

func (r *VMOperationReconciler) reconcilePending(ctx context.Context, op *o11yv1.VMOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	now, schedule := time.Now(), op.Spec.Schedule.Time
	if now.After(schedule) {
		op.Status.NextSteps = op.Spec.Steps
		op.Status.Phase = o11yv1.VMOperationPhaseRunning
		op.Status.Reason = "Begin to upgrade"
		LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, "The schedule is reached", nil)
	} else {
		if op.Status.Reason == PendingReason {
			interval := schedule.Sub(now)
			LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, fmt.Sprintf("Wait for the schedule after inteval %s", interval), nil)
			return ctrl.Result{RequeueAfter: interval}, nil
		} else {
			op.Status.Reason = PendingReason
		}
	}
	return ctrl.Result{}, r.Status().Update(ctx, op)
}

func (r *VMOperationReconciler) reconcileRunning(ctx context.Context, op *o11yv1.VMOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	if op.Status.CurrentStep == nil {
		op.UpdateCurrentStep(time.Now())
		if op.Status.Phase == o11yv1.VMOperationPhaseCompleted {
			// unpause vmcluster
			if err := unpauseVMCluster(ctx, r.Client, op); err != nil {
				return ctrl.Result{}, err
			}
			LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, "Unpaused the network and cluster", nil)
		}
		return ctrl.Result{}, r.Status().Update(ctx, op)
	}

	// pause vmcluster before every processor
	if err := pauseVMCluster(ctx, r.Client, op, true); err != nil {
		return ctrl.Result{}, err
	}
	switch p := op.Status.CurrentStep.Processor; p {
	case o11yv1.VMOperationProcessorCopyVolume:
		return r.processCopyVolume(ctx, op)
	case o11yv1.VMOperationProcessorRestartPod:
		return r.processRestartPod(ctx, op)
	default:
		logger.Info("unknown processor, forget it")
		return ctrl.Result{}, nil
	}
}

// 1. If the new pvc not exists, query the actual usage of the current volume, and create a new ebs disk based on the usage and target utilization rate
// 2. Create or update pv, pvc based on new ebs disk
// 3. Create a util job to mount the pvc and specify the schedule to the node where the vmstorage is located. The pod executes rsync to achieve data replication, based on rsync -c (checksum) to ensure data consistency
// 4. Check the pod status. If it ends normally, it means the data replication is successful and consistent. If it exits abnormally, it means the data replication is abnormal.
func (r *VMOperationReconciler) processCopyVolume(ctx context.Context, op *o11yv1.VMOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	logger.Info("maintain vminsert's vmstorage Node")
	if updated, err := r.maintainVMStorageForVMInsert(ctx, op); err != nil {
		return ctrl.Result{}, err
	} else if updated {
		return requeue, nil
	}

	curPVCName, copyPVCName := vmstoragePVCPrefix+op.Spec.TargetRef.PodName, vmstoragePVCPrefix+op.Spec.TargetRef.PodName+copyPVCSuffix
	curPVC := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      curPVCName,
		},
	}
	copyPVC := curPVC.DeepCopy()
	copyPVC.Name = copyPVCName

	if err := r.Get(ctx, client.ObjectKeyFromObject(curPVC), curPVC); client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get pvc %s: %v", curPVCName, err)
	}
	curPVName := curPVC.Spec.VolumeName

	err := r.Get(ctx, client.ObjectKeyFromObject(copyPVC), copyPVC)
	if client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, err
	}
	if apierrors.IsNotFound(err) {
		logger.Info("new pvc is not found, create a new pvc with ebs disk")
		// query prometheus for current disk usage
		q := fmt.Sprintf(VolumeUsagePromQLTmpl, curPVCName)
		if val, _, err := r.promCli.Query(ctx, q, time.Now()); err != nil {
			return ctrl.Result{}, err
		} else {
			if vec, ok := val.(model.Vector); ok && len(vec) > 0 {
				usage := resource.NewQuantity(int64(vec[0].Value)*Gi, resource.BinarySI)
				if minUsage, ok := op.Status.CurrentStep.Options[o11yv1.CopyVolumeOptsMinVolumeSize]; ok {
					if m, err := resource.ParseQuantity(minUsage); err == nil {
						if usage.Cmp(m) < 0 {
							usage = &m
						}
					} else {
						logger.Info("parse minVolumeSize failed", "minVolumeSize", minUsage)
					}
				}
				// create new disk
				copyPVC.Spec.Resources.Requests = make(corev1.ResourceList)
				copyPVC.Spec.Resources.Requests[corev1.ResourceStorage] = *usage
				copyPVC.Spec.StorageClassName = &storageClassVMStorage
				copyPVC.Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}
			} else {
				return ctrl.Result{}, fmt.Errorf("prometheus query value invalid: %v", err)
			}
			if err := r.Create(ctx, copyPVC); err != nil {
				return ctrl.Result{}, err
			}
		}
	}

	// mount pvc to new pod, this will dynamically create a new pv and ebs too.
	nodeName, err := getTargetNode(ctx, r.Client, op)
	if err != nil {
		return ctrl.Result{}, err
	}
	job, err := r.createRsyncJob(ctx, op, copyPVCName, curPVName, nodeName)
	if err != nil {
		return ctrl.Result{}, err
	}

	// check job status
	if failed, reason := jobFailed(job); failed {
		op.FailOperation(fmt.Sprintf("copy-volume job failed due to %s", reason))
		LogAndEmitEvent(logger, r.eventRecorder, op, warningEvent, reason, nil)
		return requeue, r.Status().Update(ctx, op)
	}

	if job.Status.Succeeded == 0 {
		logger.Info("requeue operation, since job is not completed")
		return ctrl.Result{Requeue: true, RequeueAfter: 1 * time.Minute}, nil
	}
	LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, "Rsync completed successfully", nil)

	// update new pvName and status
	if err := r.Get(ctx, client.ObjectKeyFromObject(copyPVC), copyPVC); err != nil {
		return ctrl.Result{}, err
	}
	op.Status.CopyPVCSpec = copyPVC.Spec.DeepCopy()
	op.FinishCurrentStep()
	return requeue, r.Status().Update(ctx, op)
}

func (r *VMOperationReconciler) maintainVMStorageForVMInsert(ctx context.Context, op *o11yv1.VMOperation) (bool, error) {
	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      vminsertPrefix + op.Spec.TargetRef.ClusterName,
			Namespace: op.Spec.TargetRef.Namespace,
		},
	}
	if err := r.Get(ctx, client.ObjectKeyFromObject(deploy), deploy); err != nil {
		return false, err
	}
	vminsert := deploy.Spec.Template.Spec.Containers[0]
	updated := false
	for i, arg := range vminsert.Args {
		if strings.Contains(arg, vminsertParamStorageNode) {
			sns := strings.Split(strings.TrimPrefix(arg, vminsertParamStorageNode), ",")
			newSns := []string{}
			for _, sn := range sns {
				if !strings.Contains(sn, op.Spec.TargetRef.PodName+dot) {
					newSns = append(newSns, sn)
				} else {
					updated = true
				}
			}
			vminsert.Args[i] = vminsertParamStorageNode + strings.Join(newSns, ",")
		}
	}
	if updated {
		return updated, r.Update(ctx, deploy)
	}

	return false, wait.PollImmediate(interval, timeout, func() (bool, error) {
		// make sure rolling update is done
		if err := r.Get(ctx, client.ObjectKeyFromObject(deploy), deploy); err != nil {
			return false, err
		}
		if deploy.Status.AvailableReplicas != deploy.Status.ReadyReplicas || deploy.Status.ReadyReplicas != *deploy.Spec.Replicas {
			return false, fmt.Errorf("deployment not ready")
		}

		// make sure all the old pods are terminated.
		pods := &corev1.PodList{}
		if err := r.List(ctx, pods, client.InNamespace(op.Spec.TargetRef.Namespace), client.MatchingLabels(deploy.Spec.Template.Labels)); err != nil {
			return false, err
		}
		for _, pod := range pods.Items {
			if pod.Status.Phase != corev1.PodRunning {
				return false, fmt.Errorf("pod %s is not running", pod.Name)
			}
			for _, arg := range pod.Spec.Containers[0].Args {
				if strings.Contains(arg, vminsertParamStorageNode) && strings.Contains(arg, op.Spec.TargetRef.PodName+dot) {
					return false, fmt.Errorf("pod %s is still writing data to storage %s", pod.Name, op.Spec.TargetRef.PodName)
				}
			}
		}
		return true, nil
	})
}

// 1. If the pvc points to the old pv
//   - Delete statefulset --cascade = orphan, stop the reconcile of sts
//   - Delete vmStorage pod
//   - Rebuild pvc based on new pv, retain old pv
//   - rebuild the statefulset, a new targetPod will be created by the statefulset
//
// 2. Check if the pod status of the corresponding id is running.
func (r *VMOperationReconciler) processRestartPod(ctx context.Context, op *o11yv1.VMOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	pvc := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      vmstoragePVCPrefix + op.Spec.TargetRef.PodName,
		},
	}
	err := r.Get(ctx, client.ObjectKeyFromObject(pvc), pvc)
	if client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, err
	}
	pvcMetaCopy := pvc.ObjectMeta.DeepCopy()
	pvcFound := err == nil

	if pvcFound && pvc.Spec.VolumeName == op.Status.CopyPVCSpec.VolumeName {
		// check if pod is running
		pod := &corev1.Pod{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: op.Spec.TargetRef.Namespace,
				Name:      op.Spec.TargetRef.PodName,
			},
		}
		err := r.Get(ctx, client.ObjectKeyFromObject(pod), pod)
		if client.IgnoreNotFound(err) != nil {
			return ctrl.Result{}, err
		}
		if apierrors.IsNotFound(err) {
			// since volumeName is changed, we assume it is the pvc with new pv,
			// but the pod is not found, which means the sts is not created yet.
			return ctrl.Result{Requeue: true, RequeueAfter: 5 * time.Second}, r.rebuildSts(ctx, op)
		}
		if pod.Status.Phase != corev1.PodRunning {
			LogAndEmitEvent(logger, r.eventRecorder, op, warningEvent, "Pod is not running", nil)
			return ctrl.Result{}, fmt.Errorf("pod is not running")
		}
		LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, "New Pod is running", nil)
		op.FinishCurrentStep()
		return requeue, r.Status().Update(ctx, op)
	}

	// pv is not replaced by the copied one, execute rebuild steps:
	// 1. ensure the sts is deleted, so as to stop the reconciling of vmstorage pods.
	// if statefulset is not found, walk through the pod, pvc, sts rebuild steps.
	logger.Info("pv is not replaced yet, ensure old statefulset is delted")
	sts := &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      vmstorageSTSPrefix + op.Spec.TargetRef.ClusterName,
		},
	}
	if err := r.Get(ctx, client.ObjectKeyFromObject(sts), sts); client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, err
	} else if err == nil {
		// vmstorage sts is found.
		if op.Status.VMStorageSts == nil {
			if stsValue, err := json.Marshal(sts); err != nil {
				return ctrl.Result{}, fmt.Errorf("marshal sts: %w", err)
			} else {
				op.Status.VMStorageSts = ptr.To[string](string(stsValue))
				return requeue, r.Status().Update(ctx, op)
			}
		}
		if controllerutil.RemoveFinalizer(sts, vm.FinalizerName) {
			if err := r.Update(ctx, sts); err != nil {
				return ctrl.Result{}, fmt.Errorf("failed remove sts's finalizer: %w", err)
			}
		}

		if err := r.Delete(ctx, sts, &client.DeleteOptions{
			PropagationPolicy: ptr.To[metav1.DeletionPropagation](metav1.DeletePropagationOrphan),
		}); client.IgnoreNotFound(err) != nil {
			return ctrl.Result{}, err
		}
	}

	// 2. ensure the old vmstorage pod is deleted, make it possible to delete the old pvc.
	logger.Info("ensure old vmstorage pod is deleted")
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      op.Spec.TargetRef.PodName,
		},
	}
	err = r.Get(ctx, client.ObjectKeyFromObject(pod), pod)
	if client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, err
	}
	if !apierrors.IsNotFound(err) {
		if err := r.Delete(ctx, pod); err != nil {
			return ctrl.Result{}, err
		}
	}

	// 3. start rebuild pvc
	// delete rsync job, since the completed pod will block the deletion of pvc.
	logger.Info("ensure rsync job & pod deleted")
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      JobPrefix + op.Spec.TargetRef.PodName,
		},
	}
	if err := r.Delete(ctx, job); client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, fmt.Errorf("failed to delete job: %w", err)
	}
	// delete pod of job
	if err := r.DeleteAllOf(ctx, &corev1.Pod{}, client.InNamespace(job.Namespace), client.MatchingLabels{
		jobLabelKey: job.Name,
	}); err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to delete pod of job: %w", err)
	}

	// delete copied pvc and retain related pv
	logger.Info("delete copied pvc and retain related pv")
	if err := retainPV(ctx, r.Client, op.Status.CopyPVCSpec.VolumeName); err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to retain copied pv: %w", err)
	}
	copyPVC := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      vmstoragePVCPrefix + op.Spec.TargetRef.PodName + copyPVCSuffix,
		},
	}
	if err := r.Delete(ctx, copyPVC); client.IgnoreNotFound(err) != nil {
		return ctrl.Result{}, fmt.Errorf("failed to delete copied pvc: %w", err)
	}

	if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
		return apierrors.IsNotFound(r.Get(ctx, client.ObjectKeyFromObject(copyPVC), copyPVC)), nil
	}); err != nil {
		return ctrl.Result{}, fmt.Errorf("wait pvc deletion timeout: %v", err)
	}

	logger.Info("delete old pvc and retain related pv")
	if pvcFound {
		if err := retainPV(ctx, r.Client, pvc.Spec.VolumeName); err != nil {
			return ctrl.Result{}, fmt.Errorf("failed to retain old pv: %w", err)
		}
		if err := r.Delete(ctx, pvc); client.IgnoreNotFound(err) != nil {
			return ctrl.Result{}, fmt.Errorf("failed to delete current pvc: %w", err)
		}
		logger.Info("wait old pvc to be deleted")
		if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
			return apierrors.IsNotFound(r.Get(ctx, client.ObjectKeyFromObject(pvc), pvc)), nil
		}); err != nil {
			return ctrl.Result{}, fmt.Errorf("wait pvc deletion timeout: %v", err)
		}
	}

	logger.Info("remove claimRef in copied pv")
	if err := removePVCRef(ctx, r.Client, op.Status.CopyPVCSpec.VolumeName); err != nil {
		return ctrl.Result{}, err
	}
	logger.Info("recreate pvc with copied pv")

	newPVC := &corev1.PersistentVolumeClaim{
		ObjectMeta: *pvcMetaCopy,
		Spec:       *op.Status.CopyPVCSpec.DeepCopy(),
	}
	ensureVMOperatorPVCLabels(newPVC, op.Spec.TargetRef.ClusterName)
	if err := r.Create(ctx, newPVC); err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to create new pvc: %w", err)
	}

	logger.Info("rebuild sts")
	return requeue, r.rebuildSts(ctx, op)
}

func (r *VMOperationReconciler) rebuildSts(ctx context.Context, op *o11yv1.VMOperation) error {
	sts := &appsv1.StatefulSet{}
	if op.Status.VMStorageSts != nil {
		if err := json.Unmarshal([]byte(*op.Status.VMStorageSts), sts); err != nil {
			return fmt.Errorf("unmarshal sts: %w", err)
		}
		sts.ResourceVersion = ""
		_, err := controllerutil.CreateOrUpdate(ctx, r.Client, sts, func() error { return nil })
		return err
	}
	return fmt.Errorf("vmstorage sts is not set in status")
}

func (r *VMOperationReconciler) reconcileFinished(ctx context.Context, op *o11yv1.VMOperation) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	now := time.Now()
	var timeToDelete time.Time
	ttl := 3600 * 24 * 7 * time.Second
	if op.Spec.TTLSeconds != nil {
		ttl = op.Spec.TTLSeconds.Duration
	}
	if op.Status.EndTime != nil {
		timeToDelete = op.Status.EndTime.Add(ttl)
	} else if op.Status.BeginTime != nil {
		timeToDelete = op.Status.BeginTime.Add(ttl)
	} else {
		timeToDelete = op.GetCreationTimestamp().Add(ttl)
	}

	logger.WithValues("TTL", ttl, "now", now, "timeToDelete", timeToDelete)

	if now.After(timeToDelete) {
		logger.Info("delete VMOperation")
		return ctrl.Result{}, r.Delete(ctx, op)
	}

	interval := timeToDelete.Sub(now)
	LogAndEmitEvent(logger, r.eventRecorder, op, normalEvent, fmt.Sprintf("Wait for the VMOperation to be deleted after interval %s", interval), nil)
	return ctrl.Result{RequeueAfter: interval}, nil
}

// LogAndEmitEvent logs the event and emits it using the provided logger and eventRecorder.
// The event type determines how the event is logged.
func LogAndEmitEvent(logger logr.Logger, eventRecorder record.EventRecorder, obj runtime.Object, eventType, msg string, err error) {
	switch eventType {
	case normalEvent:
		logger.Info(msg)

	case warningEvent:
		logger.Error(err, msg)
	}
	eventRecorder.Event(obj, eventType, "", msg)
}

func newPromClient(url string) (promv1.API, error) {
	apiClient, err := promapi.NewClient(promapi.Config{
		Address: url,
	})

	if err != nil {
		return nil, err
	}
	return promv1.NewAPI(apiClient), nil
}

func unpauseVMCluster(ctx context.Context, cli client.Client, op *o11yv1.VMOperation) error {
	return pauseVMCluster(ctx, cli, op, false)
}

func pauseVMCluster(ctx context.Context, cli client.Client, op *o11yv1.VMOperation, pause bool) error {
	vmcluster := &vm.VMCluster{}
	if err := cli.Get(ctx, types.NamespacedName{
		Namespace: op.Spec.TargetRef.Namespace,
		Name:      op.Spec.TargetRef.ClusterName,
	}, vmcluster); err != nil {
		return err
	}
	if pause != vmcluster.Paused() {
		vmcluster.Spec.Paused = pause
		return cli.Update(ctx, vmcluster)
	}
	return nil
}

func retainPV(ctx context.Context, cli client.Client, pvName string) error {
	pv := &corev1.PersistentVolume{}
	if err := cli.Get(ctx, types.NamespacedName{Name: pvName}, pv); err != nil {
		return err
	}
	if pv.Spec.PersistentVolumeReclaimPolicy == corev1.PersistentVolumeReclaimRetain {
		return nil
	}
	pv.Spec.PersistentVolumeReclaimPolicy = corev1.PersistentVolumeReclaimRetain
	return cli.Update(ctx, pv)
}

func removePVCRef(ctx context.Context, cli client.Client, pvName string) error {
	pv := &corev1.PersistentVolume{}
	if err := cli.Get(ctx, types.NamespacedName{Name: pvName}, pv); err != nil {
		return err
	}
	if pv.Spec.ClaimRef == nil {
		return nil
	}
	pv.Spec.ClaimRef = nil
	return cli.Update(ctx, pv)
}

func getTargetNode(ctx context.Context, cli client.Client, op *o11yv1.VMOperation) (string, error) {
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: op.Spec.TargetRef.Namespace,
			Name:      op.Spec.TargetRef.PodName,
		},
	}
	if err := cli.Get(ctx, client.ObjectKeyFromObject(pod), pod); err != nil {
		return "", err
	}
	return pod.Spec.NodeName, nil
}

func jobFailed(job *batchv1.Job) (bool, string) {
	for _, condition := range job.Status.Conditions {
		if condition.Type == v1.JobFailed && condition.Status == "True" {
			return true, condition.Reason
		}
	}
	return false, ""
}

func ensureVMOperatorPVCLabels(pvc *corev1.PersistentVolumeClaim, clusterName string) {
	if pvc.Labels == nil {
		pvc.Labels = map[string]string{}
	}
	pvc.Labels["app.kubernetes.io/name"] = "vmstorage"
	pvc.Labels["app.kubernetes.io/instance"] = clusterName
	pvc.Labels["app.kubernetes.io/component"] = "monitoring"
	pvc.Labels["managed-by"] = "vm-operator"
}