func (r *ReconcileHelper) ReconcileAstarteResources(instance *apiv1alpha1.Astarte) error {
	if err := recon.EnsureHousekeepingKey(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if err := recon.EnsureGenericErlangConfiguration(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if err := recon.EnsureRabbitMQ(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if err := recon.EnsureCassandra(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if err := recon.EnsureCFSSL(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if version.CheckConstraintAgainstAstarteVersion("< 1.0.0", instance.Spec.Version) == nil {
		if err := recon.EnsureCFSSLCASecret(instance, r.Client, r.Scheme); err != nil {
			return err
		}
	}
	if err := r.EnsureAstarteMicroservices(instance); err != nil {
		return err
	}
	if err := recon.EnsureVerneMQ(instance, r.Client, r.Scheme); err != nil {
		return err
	}
	if err := recon.EnsureAstarteDashboard(instance, instance.Spec.Components.Dashboard, r.Client, r.Scheme); err != nil {
		return err
	}
	return nil
}
ReconcileAstarteResources reconciles all third-party dependencies, when needed
package controllerutils
import (
	"context"
	"fmt"
	"strings"
	"time"
	semver "github.com/Masterminds/semver/v3"
	"github.com/go-logr/logr"
	"github.com/openlyinc/pointy"
	appsv1 "k8s.io/api/apps/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	commontypes "github.com/astarte-platform/astarte-kubernetes-operator/apis/api/commontypes"
	apiv1alpha1 "github.com/astarte-platform/astarte-kubernetes-operator/apis/api/v1alpha1"
	"github.com/astarte-platform/astarte-kubernetes-operator/lib/migrate"
	"github.com/astarte-platform/astarte-kubernetes-operator/lib/misc"
	recon "github.com/astarte-platform/astarte-kubernetes-operator/lib/reconcile"
	"github.com/astarte-platform/astarte-kubernetes-operator/lib/upgrade"
	"github.com/astarte-platform/astarte-kubernetes-operator/version"
)
type ReconcileHelper struct {
	Client   client.Client
	Scheme   *runtime.Scheme
	Recorder record.EventRecorder
}
func (r *ReconcileHelper) CheckAndPerformUpgrade(reqLogger logr.Logger, instance *apiv1alpha1.Astarte, newAstarteSemVersion *semver.Version) (ctrl.Result, error) {
	computedClusterHealth := r.ComputeClusterHealth(reqLogger, instance)
	if computedClusterHealth != commontypes.AstarteClusterHealthGreen && instance.Status.Health != commontypes.AstarteClusterHealthGreen {
		reqLogger.Error(fmt.Errorf("Astarte Upgrade requested, but the cluster isn't reporting stable Health. Refusing to upgrade"),
			"Cluster health is unstable, refusing to upgrade. Please revert to the previous version and wait for the cluster to settle.",
			"Reported Health", instance.Status.Health, "Computed Health", computedClusterHealth)
		r.Recorder.Event(instance, "Warning", commontypes.AstarteResourceEventCriticalError.String(),
			fmt.Sprintf("Cluster health is %s, refusing to upgrade. Please revert to the previous version and wait for the cluster to settle", computedClusterHealth))
		return ctrl.Result{Requeue: false}, fmt.Errorf("Astarte Upgrade requested, but the cluster isn't reporting stable Health. Refusing to upgrade")
	}
	versionString := instance.Status.AstarteVersion
	if strings.Contains(instance.Status.AstarteVersion, "-snapshot") {
		versionString = strings.ReplaceAll(versionString, "-snapshot", ".0")
		reqLogger.Info("You are running a Release snapshot. This is generally not a good idea in production. Assuming a Release version", "Version", versionString)
		r.Recorder.Eventf(instance, "Normal", commontypes.AstarteResourceEventUpgrade.String(),
			"Requested an upgrade from a Release snapshot. Assuming the base Release version is %v", versionString)
	}
	oldAstarteSemVersion, err := version.GetAstarteSemanticVersionFrom(versionString)
	if err != nil {
		r.Recorder.Eventf(instance, "Warning", commontypes.AstarteResourceEventCriticalError.String(),
			err.Error(), versionString)
		return ctrl.Result{RequeueAfter: time.Minute}, err
	}
	if e := upgrade.EnsureAstarteUpgrade(oldAstarteSemVersion, newAstarteSemVersion, instance, r.Client, r.Scheme, r.Recorder); e != nil {
		return ctrl.Result{}, e
	}
	return ctrl.Result{}, nil
}
func (r *ReconcileHelper) ComputeClusterHealth(reqLogger logr.Logger, instance *apiv1alpha1.Astarte) commontypes.AstarteClusterHealth {
	astarteDeployments := &appsv1.DeploymentList{}
	nonReadyDeployments := 0
	if err := r.Client.List(context.TODO(), astarteDeployments, client.InNamespace(instance.Namespace),
		client.MatchingLabels{"component": "astarte"}); err == nil {
		for _, deployment := range astarteDeployments.Items {
			if deployment.Status.ReadyReplicas == 0 && pointy.Int32Value(deployment.Spec.Replicas, 0) > 0 {
				nonReadyDeployments++
			}
		}
	} else {
		reqLogger.Info("Could not list Astarte deployments to compute health.")
		nonReadyDeployments = 5
	}
	astarteStatefulSet := &appsv1.StatefulSet{}
	if pointy.BoolValue(instance.Spec.VerneMQ.Deploy, true) {
		if err := r.Client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.Name + "-vernemq"},
			astarteStatefulSet); err == nil {
			if astarteStatefulSet.Status.ReadyReplicas == 0 {
				nonReadyDeployments++
			}
		} else {
			reqLogger.V(1).Info("Could not Get Astarte VerneMQ StatefulSet to compute health.")
			nonReadyDeployments++
		}
	}
	if !r.computeCFSSLHealth(reqLogger, instance) {
		nonReadyDeployments++
	}
	if !r.computeRabbitMQHealth(reqLogger, instance) {
		nonReadyDeployments++
	}
	if nonReadyDeployments == 0 {
		return commontypes.AstarteClusterHealthGreen
	} else if nonReadyDeployments == 1 {
		return commontypes.AstarteClusterHealthYellow
	}
	return commontypes.AstarteClusterHealthRed
}
func (r *ReconcileHelper) computeCFSSLHealth(reqLogger logr.Logger, instance *apiv1alpha1.Astarte) bool {
	if !pointy.BoolValue(instance.Spec.CFSSL.Deploy, true) {
		return true
	}
	switch version.CheckConstraintAgainstAstarteVersion("< 1.0.0", instance.Spec.Version) {
	case nil:
		cfsslStatefulSet := &appsv1.StatefulSet{}
		if err := r.Client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.Name + "-cfssl"},
			cfsslStatefulSet); err == nil {
			if cfsslStatefulSet.Status.ReadyReplicas == 0 {
				return false
			}
		} else {
			reqLogger.V(1).Info("Could not Get Astarte CFSSL StatefulSet to compute health.")
			return false
		}
	case version.ErrConstraintNotSatisfied:
		cfsslDeployment := &appsv1.Deployment{}
		if err := r.Client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.Name + "-cfssl"},
			cfsslDeployment); err == nil {
			if cfsslDeployment.Status.ReadyReplicas == 0 {
				return false
			}
		} else {
			reqLogger.V(1).Info("Could not Get Astarte CFSSL Deployment to compute health.")
			return false
		}
	default:
		return false
	}
	return true
}
func (r *ReconcileHelper) computeRabbitMQHealth(reqLogger logr.Logger, instance *apiv1alpha1.Astarte) bool {
	if !pointy.BoolValue(instance.Spec.RabbitMQ.Deploy, true) {
		return true
	}
	rabbitMQStatefulSet := &appsv1.StatefulSet{}
	if err := r.Client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.Name + "-rabbitmq"}, rabbitMQStatefulSet); err == nil {
		if rabbitMQStatefulSet.Status.ReadyReplicas == 0 {
			return false
		}
	} else {
		reqLogger.V(1).Info("Could not Get Astarte RabbitMQ StatefulSet to compute health.")
		return false
	}
	return true
}
func (r *ReconcileHelper) EnsureStatusCoherency(reqLogger logr.Logger, instance *apiv1alpha1.Astarte, request ctrl.Request) (ctrl.Result, error) {
	if instance.Status.AstarteVersion != "" {
		return ctrl.Result{}, nil
	}
	hkDeployment := &appsv1.Deployment{}
	if err := r.Client.Get(context.TODO(),
		types.NamespacedName{Name: instance.Name + "-housekeeping", Namespace: instance.Namespace}, hkDeployment); err == nil {
		r.Recorder.Event(instance, "Warning", commontypes.AstarteResourceEventMigration.String(),
			"Found an invalid status. The resource will be migrated to latest format")
		reqLogger.Info("Found an invalid status. Attempting to migrate the resource.")
		if e := migrate.ToNewCR(instance, r.Client, r.Scheme); e != nil {
			return ctrl.Result{}, e
		}
		instance.Status.ReconciliationPhase = commontypes.ReconciliationPhaseReconciling
		instance.Status.OperatorVersion = version.Version
		instance.Status.Health = commontypes.AstarteClusterHealthRed
		instance.Status.BaseAPIURL = instance.Spec.API.Host
		instance.Status.BrokerURL = instance.Spec.VerneMQ.Host
		reqLogger.Info("Reconciling Astarte Version from Housekeeping's image tag")
		hkImage := hkDeployment.Spec.Template.Spec.Containers[0].Image
		hkImageTokens := strings.Split(hkImage, ":")
		if len(hkImageTokens) != 2 {
			r.Recorder.Eventf(instance, "Warning", commontypes.AstarteResourceEventCriticalError.String(),
				"Could not parse Astarte version from Housekeeping Image tag %s. Please fix your resource definition", hkImage)
			return ctrl.Result{RequeueAfter: time.Minute},
				fmt.Errorf("Could not parse Astarte version from Housekeeping Image tag %s. Refusing to proceed", hkImage)
		}
		instance.Status.AstarteVersion = hkImageTokens[1]
		if e := r.Client.Status().Update(context.TODO(), instance); e != nil {
			r.Recorder.Event(instance, "Warning", commontypes.AstarteResourceEventReconciliationFailed.String(),
				"Failed to update Astarte status - will retry reconciliation")
			reqLogger.Error(e, "Failed to update Astarte status.")
			return ctrl.Result{}, e
		}
		r.Recorder.Eventf(instance, "Normal", commontypes.AstarteResourceEventMigration.String(),
			"Resource version reconciled to %v from Housekeeping", hkImageTokens[1])
		instance = &apiv1alpha1.Astarte{}
		if e := r.Client.Get(context.TODO(), request.NamespacedName, instance); e != nil {
			return ctrl.Result{}, e
		}
	} else if !errors.IsNotFound(err) {
		return ctrl.Result{}, err
	}
	r.Recorder.Event(instance, "Normal", commontypes.AstarteResourceEventStatus.String(),
		"Running first resource reconciliation")
	reqLogger.V(1).Info("Apparently running first reconciliation.")
	return ctrl.Result{}, nil
}
func (r *ReconcileHelper) ComputeAstarteStatusResource(reqLogger logr.Logger, instance *apiv1alpha1.Astarte) commontypes.AstarteStatus {
	oldAstarteHealth := instance.Status.Health
	newAstarteStatus := instance.Status
	newAstarteStatus.Health = r.ComputeClusterHealth(reqLogger, instance)
	if oldAstarteHealth != newAstarteStatus.Health && oldAstarteHealth != "" {
		eventtype := "Normal"
		if oldAstarteHealth == commontypes.AstarteClusterHealthGreen {
			eventtype = "Warning"
		}
		r.Recorder.Eventf(instance, eventtype, commontypes.AstarteResourceEventStatus.String(),
			"Astarte Cluster status changed from %v to %v", oldAstarteHealth, newAstarteStatus.Health)
	}
	newAstarteStatus.AstarteVersion = instance.Spec.Version
	newAstarteStatus.OperatorVersion = version.Version
	newAstarteStatus.ReconciliationPhase = commontypes.ReconciliationPhaseReconciled
	newAstarteStatus.BaseAPIURL = "https://" + instance.Spec.API.Host
	newAstarteStatus.BrokerURL = misc.GetVerneMQBrokerURL(instance)
	return newAstarteStatus
}