client-go/tools/leaderelection/leasecandidate.go

/*
Copyright 2024 The Kubernetes Authors.

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 leaderelection

import (
	"context"
	"reflect"
	"sync"
	"time"

	v1 "k8s.io/api/coordination/v1"
	v1beta1 "k8s.io/api/coordination/v1beta1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	coordinationv1beta1client "k8s.io/client-go/kubernetes/typed/coordination/v1beta1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog/v2"
	"k8s.io/utils/clock"

	coordinationv1beta1listers "k8s.io/client-go/listers/coordination/v1beta1"
)

const requeueInterval = 5 * time.Minute

type CacheSyncWaiter interface {
	WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
}

type LeaseCandidate struct {
	leaseClient            coordinationv1beta1client.LeaseCandidateInterface
	leaseCandidateInformer cache.SharedIndexInformer
	leaseCandidateLister   coordinationv1beta1listers.LeaseCandidateLister
	informerFactory        informers.SharedInformerFactory
	hasSynced              cache.InformerSynced

	// At most there will be one item in this Queue (since we only watch one item)
	queue workqueue.TypedRateLimitingInterface[int]

	name      string
	namespace string

	// controller lease
	leaseName string

	clock clock.Clock

	binaryVersion, emulationVersion string
	strategy                        v1.CoordinatedLeaseStrategy
}

// NewCandidate creates new LeaseCandidate controller that creates a
// LeaseCandidate object if it does not exist and watches changes
// to the corresponding object and renews if PingTime is set.
// WARNING: This is an ALPHA feature. Ensure that the CoordinatedLeaderElection
// feature gate is on.
func NewCandidate(clientset kubernetes.Interface,
	candidateNamespace string,
	candidateName string,
	targetLease string,
	binaryVersion, emulationVersion string,
	strategy v1.CoordinatedLeaseStrategy,
) (*LeaseCandidate, CacheSyncWaiter, error) {
	fieldSelector := fields.OneTermEqualSelector("metadata.name", candidateName).String()
	// A separate informer factory is required because this must start before informerFactories
	// are started for leader elected components
	informerFactory := informers.NewSharedInformerFactoryWithOptions(
		clientset, 5*time.Minute,
		informers.WithNamespace(candidateNamespace),
		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
			options.FieldSelector = fieldSelector
		}),
	)
	leaseCandidateInformer := informerFactory.Coordination().V1beta1().LeaseCandidates().Informer()
	leaseCandidateLister := informerFactory.Coordination().V1beta1().LeaseCandidates().Lister()

	lc := &LeaseCandidate{
		leaseClient:            clientset.CoordinationV1beta1().LeaseCandidates(candidateNamespace),
		leaseCandidateInformer: leaseCandidateInformer,
		leaseCandidateLister:   leaseCandidateLister,
		informerFactory:        informerFactory,
		name:                   candidateName,
		namespace:              candidateNamespace,
		leaseName:              targetLease,
		clock:                  clock.RealClock{},
		binaryVersion:          binaryVersion,
		emulationVersion:       emulationVersion,
		strategy:               strategy,
	}
	lc.queue = workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[int](), workqueue.TypedRateLimitingQueueConfig[int]{Name: "leasecandidate"})

	h, err := leaseCandidateInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		UpdateFunc: func(oldObj, newObj interface{}) {
			if leasecandidate, ok := newObj.(*v1beta1.LeaseCandidate); ok {
				if leasecandidate.Spec.PingTime != nil && leasecandidate.Spec.PingTime.After(leasecandidate.Spec.RenewTime.Time) {
					lc.enqueueLease()
				}
			}
		},
	})
	if err != nil {
		return nil, nil, err
	}
	lc.hasSynced = h.HasSynced

	return lc, informerFactory, nil
}

func (c *LeaseCandidate) Run(ctx context.Context) {
	logger := klog.FromContext(ctx)
	logger = klog.LoggerWithName(logger, "leasecandidate")
	ctx = klog.NewContext(ctx, logger)

	var wg sync.WaitGroup
	defer func() {
		c.queue.ShutDown()
		wg.Wait()
	}()

	c.informerFactory.Start(ctx.Done())
	if !cache.WaitForNamedCacheSyncWithContext(ctx, c.hasSynced) {
		return
	}

	c.enqueueLease()
	wg.Go(func() {
		c.runWorker(ctx)
	})
	<-ctx.Done()
}

func (c *LeaseCandidate) runWorker(ctx context.Context) {
	for c.processNextWorkItem(ctx) {
	}
}

func (c *LeaseCandidate) processNextWorkItem(ctx context.Context) bool {
	key, shutdown := c.queue.Get()
	if shutdown {
		return false
	}
	defer c.queue.Done(key)

	err := c.ensureLease(ctx)
	if err == nil {
		c.queue.AddAfter(key, requeueInterval)
		return true
	}

	utilruntime.HandleErrorWithContext(ctx, err, "Ensuring lease failed")
	c.queue.AddRateLimited(key)

	return true
}

func (c *LeaseCandidate) enqueueLease() {
	c.queue.Add(0)
}

// ensureLease creates the lease if it does not exist and renew it if it exists. Returns the lease and
// a bool (true if this call created the lease), or any error that occurs.
func (c *LeaseCandidate) ensureLease(ctx context.Context) error {
	logger := klog.FromContext(ctx)
	lease, err := c.leaseCandidateLister.LeaseCandidates(c.namespace).Get(c.name)
	if apierrors.IsNotFound(err) {
		logger.V(2).Info("Creating lease candidate")
		// lease does not exist, create it.
		leaseToCreate := c.newLeaseCandidate()
		if _, err := c.leaseClient.Create(ctx, leaseToCreate, metav1.CreateOptions{}); err != nil {
			return err
		}
		logger.V(2).Info("Created lease candidate")
		return nil
	} else if err != nil {
		return err
	}
	logger.V(2).Info("Lease candidate exists. Renewing.")
	clone := lease.DeepCopy()
	clone.Spec.RenewTime = &metav1.MicroTime{Time: c.clock.Now()}
	_, err = c.leaseClient.Update(ctx, clone, metav1.UpdateOptions{})
	if err != nil {
		return err
	}
	return nil
}

func (c *LeaseCandidate) newLeaseCandidate() *v1beta1.LeaseCandidate {
	lc := &v1beta1.LeaseCandidate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      c.name,
			Namespace: c.namespace,
		},
		Spec: v1beta1.LeaseCandidateSpec{
			LeaseName:        c.leaseName,
			BinaryVersion:    c.binaryVersion,
			EmulationVersion: c.emulationVersion,
			Strategy:         c.strategy,
		},
	}
	lc.Spec.RenewTime = &metav1.MicroTime{Time: c.clock.Now()}
	return lc
}