open-cluster-management/pkg/registration/spoke/addon/lease_controller.go

package addon

import (
	"context"
	"fmt"
	"time"

	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	coordv1client "k8s.io/client-go/kubernetes/typed/coordination/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/utils/clock"

	addonv1beta1 "open-cluster-management.io/api/addon/v1beta1"
	addonclient "open-cluster-management.io/api/client/addon/clientset/versioned"
	addoninformerv1beta1 "open-cluster-management.io/api/client/addon/informers/externalversions/addon/v1beta1"
	addonlisterv1beta1 "open-cluster-management.io/api/client/addon/listers/addon/v1beta1"
	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
	"open-cluster-management.io/sdk-go/pkg/patcher"
)

const leaseDurationTimes = 5

// AddOnLeaseControllerLeaseDurationSeconds is exposed so that integration tests can crank up the lease update speed.
// TODO we may add this to ManagedClusterAddOn API to allow addon to adjust its own lease duration seconds
var AddOnLeaseControllerLeaseDurationSeconds = 60

// managedClusterAddOnLeaseController updates the managed cluster addons status on the hub cluster through checking the add-on
// lease on the managed/management cluster.
type managedClusterAddOnLeaseController struct {
	clusterName string
	clock       clock.Clock
	patcher     patcher.Patcher[
		*addonv1beta1.ManagedClusterAddOn, addonv1beta1.ManagedClusterAddOnSpec, addonv1beta1.ManagedClusterAddOnStatus]
	addOnLister           addonlisterv1beta1.ManagedClusterAddOnLister
	managementLeaseClient coordv1client.CoordinationV1Interface
	spokeLeaseClient      coordv1client.CoordinationV1Interface
}

// NewManagedClusterAddOnLeaseController returns an instance of managedClusterAddOnLeaseController
func NewManagedClusterAddOnLeaseController(clusterName string,
	addOnClient addonclient.Interface,
	addOnInformer addoninformerv1beta1.ManagedClusterAddOnInformer,
	managementLeaseClient coordv1client.CoordinationV1Interface,
	spokeLeaseClient coordv1client.CoordinationV1Interface,
	resyncInterval time.Duration) factory.Controller {
	c := &managedClusterAddOnLeaseController{
		clusterName: clusterName,
		clock:       clock.RealClock{},
		patcher: patcher.NewPatcher[
			*addonv1beta1.ManagedClusterAddOn, addonv1beta1.ManagedClusterAddOnSpec, addonv1beta1.ManagedClusterAddOnStatus](
			addOnClient.AddonV1beta1().ManagedClusterAddOns(clusterName)),
		addOnLister:           addOnInformer.Lister(),
		managementLeaseClient: managementLeaseClient,
		spokeLeaseClient:      spokeLeaseClient,
	}

	// TODO We do not add leaser informer to support kubernetes version lower than 1.17. Lease v1 api
	// is introduced in v1.17, hence adding lease informer in this controller will cause the hang of
	// informer cache sync and result in fatal exit of this controller. The code will be factored
	// when we no longer support kubernetes version lower than 1.17.
	return factory.New().
		WithSync(c.sync).
		ResyncEvery(resyncInterval).
		ToController("ManagedClusterAddOnLeaseController")
}

func (c *managedClusterAddOnLeaseController) sync(ctx context.Context, syncCtx factory.SyncContext, queueKey string) error {
	if queueKey == factory.DefaultQueueKey {
		addOns, err := c.addOnLister.ManagedClusterAddOns(c.clusterName).List(labels.Everything())
		if err != nil {
			return err
		}
		for _, addOn := range addOns {
			// enqueue the addon to reconcile
			syncCtx.Queue().Add(fmt.Sprintf("%s/%s", getAddOnInstallationNamespace(addOn), addOn.Name))
		}
		return nil
	}

	addOnNamespace, addOnName, err := cache.SplitMetaNamespaceKey(queueKey)
	if err != nil {
		// queue key is bad format, ignore it.
		return nil
	}

	addOn, err := c.addOnLister.ManagedClusterAddOns(c.clusterName).Get(addOnName)
	if errors.IsNotFound(err) {
		// addon is not found, could be deleted, ignore it.
		return nil
	}
	if err != nil {
		return err
	}

	// "Customized" mode health check is supposed to delegate the health checking
	// to the addon manager.
	if addOn.Status.HealthCheck.Mode == addonv1beta1.HealthCheckModeCustomized {
		return nil
	}

	return c.syncSingle(ctx, syncCtx, addOnNamespace, addOn)
}

func (c *managedClusterAddOnLeaseController) syncSingle(ctx context.Context,
	syncCtx factory.SyncContext,
	leaseNamespace string,
	addOn *addonv1beta1.ManagedClusterAddOn) error {
	now := c.clock.Now()
	gracePeriod := time.Duration(leaseDurationTimes*AddOnLeaseControllerLeaseDurationSeconds) * time.Second

	// if the add-on agent is running on the managed cluster, try to fetch the add-on lease on the managed cluster,
	// otherwise (running outside of the managed cluster), fetch the add-on lease on the management cluster instead.
	leaseClient := c.spokeLeaseClient
	if isAddonRunningOutsideManagedCluster(addOn) {
		leaseClient = c.managementLeaseClient
	}

	// addon lease name should be same with the addon name.
	observedLease, err := leaseClient.Leases(leaseNamespace).Get(ctx, addOn.Name, metav1.GetOptions{})

	var condition metav1.Condition
	switch {
	case errors.IsNotFound(err):
		message := fmt.Sprintf("The status of %s add-on is unknown.", addOn.Name)
		if isAddonRunningOutsideManagedCluster(addOn) {
			// No lease on the declared hosting cluster usually means it doesn't match the klusterlet's actual hosting cluster.
			hostingCluster := addOn.Annotations[addonv1beta1.HostingClusterNameAnnotationKey]
			message = fmt.Sprintf("%s No lease found on hosting cluster %q; verify it matches the cluster "+
				"where this managed cluster's klusterlet actually runs.", message, hostingCluster)
		}
		condition = metav1.Condition{
			Type:    addonv1beta1.ManagedClusterAddOnConditionAvailable,
			Status:  metav1.ConditionUnknown,
			Reason:  "ManagedClusterAddOnLeaseNotFound",
			Message: message,
		}
	case err != nil:
		return err
	case err == nil:
		if now.Before(observedLease.Spec.RenewTime.Add(gracePeriod)) {
			// the lease is constantly updated, update its addon status to available
			condition = metav1.Condition{
				Type:    addonv1beta1.ManagedClusterAddOnConditionAvailable,
				Status:  metav1.ConditionTrue,
				Reason:  "ManagedClusterAddOnLeaseUpdated",
				Message: fmt.Sprintf("%s add-on is available.", addOn.Name),
			}
			break
		}

		// the lease is not constantly updated, update its addon status to unavailable
		condition = metav1.Condition{
			Type:    addonv1beta1.ManagedClusterAddOnConditionAvailable,
			Status:  metav1.ConditionFalse,
			Reason:  "ManagedClusterAddOnLeaseUpdateStopped",
			Message: fmt.Sprintf("%s add-on is not available.", addOn.Name),
		}
	}

	newAddon := addOn.DeepCopy()
	meta.SetStatusCondition(&newAddon.Status.Conditions, condition)
	updated, err := c.patcher.PatchStatus(ctx, newAddon, newAddon.Status, addOn.Status)
	if err != nil {
		return err
	}
	if updated {
		syncCtx.Recorder().Eventf(ctx, "ManagedClusterAddOnStatusUpdated",
			"update managed cluster addon %q available condition to %q with its lease %q/%q status",
			addOn.Name, condition.Status, leaseNamespace, addOn.Name)
	}

	return nil
}

func (c *managedClusterAddOnLeaseController) queueKeyFunc(lease runtime.Object) string {
	accessor, _ := meta.Accessor(lease)

	name := accessor.GetName()
	// addon lease name should be same with the addon name.
	addOn, err := c.addOnLister.ManagedClusterAddOns(c.clusterName).Get(name)
	if err != nil {
		// failed to get addon from hub, ignore this reconciliation.
		return ""
	}

	namespace := accessor.GetNamespace()
	if namespace != getAddOnInstallationNamespace(addOn) {
		// the lease namesapce is not same with its addon installation namespace, ignore it.
		return ""
	}

	return namespace + "/" + name
}