Feat/leader election mechanism using leases

cmd/kubedownscaler/main.go

@@ -6,22 +6,27 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"os/signal"
 	"regexp"
 	"sync"
+	"syscall"
 	"time"
 	_ "time/tzdata"
 
 	"github.com/caas-team/gokubedownscaler/internal/api/kubernetes"
 	"github.com/caas-team/gokubedownscaler/internal/pkg/scalable"
 	"github.com/caas-team/gokubedownscaler/internal/pkg/util"
 	"github.com/caas-team/gokubedownscaler/internal/pkg/values"
+	"k8s.io/client-go/tools/leaderelection"
 )
 
 const (
 	// value defaults.
 	defaultGracePeriod       = 15 * time.Minute
 	defaultDownscaleReplicas = 0
 
+	leaseName = "downscaler-lease"
+
 	// runtime config defaults.
 	defaultInterval = 30 * time.Second
 )
@@ -42,25 +47,30 @@ func main() {
 		Kubeconfig:        "",
 	}
 
+	config.ParseConfigFlags()
+
+	err := config.ParseConfigEnvVars()
+	if err != nil {
+		slog.Error("failed to parse env vars for config", "error", err)
+		os.Exit(1)
+	}
+
 	layerCli := values.NewLayer()
 	layerEnv := values.NewLayer()
 
+	err = layerEnv.GetLayerFromEnv()
+	if err != nil {
+		slog.Error("failed to get layer from env", "error", err)
+		os.Exit(1)
+	}
+
 	// set defaults for layers
 	layerCli.GracePeriod = defaultGracePeriod
 	layerCli.DownscaleReplicas = defaultDownscaleReplicas
-
-	config.ParseConfigFlags()
-
 	layerCli.ParseLayerFlags()
 
 	flag.Parse()
 
-	err := layerEnv.GetLayerFromEnv()
-	if err != nil {
-		slog.Error("failed to get layer from env", "error", err)
-		os.Exit(1)
-	}
-
 	if config.Debug || config.DryRun {
 		slog.SetLogLoggerLevel(slog.LevelDebug)
 	}
@@ -70,8 +80,6 @@ func main() {
 		os.Exit(1)
 	}
 
-	ctx := context.Background()
-
 	slog.Debug("getting client for kubernetes")
 
 	client, err := kubernetes.NewClient(config.Kubeconfig, config.DryRun)
@@ -80,20 +88,103 @@ func main() {
 		os.Exit(1)
 	}
 
-	slog.Info("started downscaler")
+	ctx, cancel := context.WithCancel(context.Background())
+
+	if !config.LeaderElection {
+		runWithoutLeaderElection(client, ctx, &layerCli, &layerEnv, config)
+		return
+	}
+
+	downscalerNamespace, err := kubernetes.GetCurrentNamespace()
+	if err != nil {
+		slog.Warn("couldn't get namespace or running outside of cluster; skipping leader election", "error", err)
+		os.Exit(1)
+	}
+
+	runWithLeaderElection(client, downscalerNamespace, cancel, ctx, &layerCli, &layerEnv, config)
+}
 
-	err = scanWorkloads(client, ctx, &layerCli, &layerEnv, config)
+func runWithLeaderElection(
+	client kubernetes.Client,
+	downscalerNamespace string,
+	cancel context.CancelFunc,
+	ctx context.Context,
+	layerCli, layerEnv *values.Layer,
+	config *util.RuntimeConfiguration,
+) {
+	lease, err := client.CreateLease(leaseName, downscalerNamespace)
+	if err != nil {
+		slog.Warn("failed to create lease", "error", err)
+		os.Exit(1)
+	}
+
+	defer cancel()
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		<-sigs
+		cancel()
+	}()
+
+	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
+		Lock:            lease,
+		ReleaseOnCancel: true,
+		LeaseDuration:   30 * time.Second,
+		RenewDeadline:   20 * time.Second,
+		RetryPeriod:     5 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				slog.Info("started leading")
+				err = startScanning(client, ctx, layerCli, layerEnv, config)
+				if err != nil {
+					slog.Error("an error occurred while scanning workloads", "error", err)
+					cancel()
+				}
+			},
+			OnStoppedLeading: func() {
+				slog.Info("stopped leading")
+				cancel()
+			},
+			OnNewLeader: func(identity string) {
+				slog.Info("new leader elected", "identity", identity)
+			},
+		},
+	})
+}
+
+func runWithoutLeaderElection(
+	client kubernetes.Client,
+	ctx context.Context,
+	layerCli, layerEnv *values.Layer,
+	config *util.RuntimeConfiguration,
+) {
+	slog.Warn("proceeding without leader election, this may cause multiple downscaler instances to conflict when modifying the same resources")
+
+	err := startScanning(client, ctx, layerCli, layerEnv, config)
 	if err != nil {
-		slog.Error("failed to scan over workloads",
-			"error", err,
-			"config", config,
-			"CliLayer", layerCli,
-			"EnvLayer", layerEnv,
-		)
+		slog.Error("an error occurred while scanning workloads, exiting", "error", err)
 		os.Exit(1)
 	}
 }
 
+func startScanning(
+	client kubernetes.Client,
+	ctx context.Context,
+	layerCli, layerEnv *values.Layer,
+	config *util.RuntimeConfiguration,
+) error {
+	slog.Info("started downscaler")
+
+	err := scanWorkloads(client, ctx, layerCli, layerEnv, config)
+	if err != nil {
+		return fmt.Errorf("failed to scan over workloads: %w", err)
+	}
+
+	return nil
+}
+
 // scanWorkloads scans over all workloads every scan.
 func scanWorkloads(
 	client kubernetes.Client,

deployments/chart/templates/_helpers.tpl

@@ -23,6 +23,13 @@ Create chart name and version as used by the chart label.
 {{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
 {{- end }}
 
+{{/*
+If replicaCount is greater than 1 leader election is enabled by default.
+*/}}
+{{- define "go-kube-downscaler.leaderElection" -}}
+{{- if (.Values.replicaCount | int | gt 1) }}
+{{- end }}
+
 {{/*
 Common labels
 */}}

deployments/chart/templates/deployment.yaml

@@ -23,6 +23,9 @@ spec:
           {{- with .Values.arguments }}
           {{- toYaml . | nindent 10 }}
           {{- end }}
+          {{- if include "go-kube-downscaler.leaderElection" . }}
+          - --leader-election
+          {{- end }}
           {{- if .Values.constrainedDownscaler }}
           - --namespace={{ join "," .Values.constrainedNamespaces }}
           {{- end }}

deployments/chart/templates/leaserole.yaml

@@ -0,0 +1,32 @@
+{{- if include "go-kube-downscaler.leaderElection" . }}
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: {{ include "go-kube-downscaler.fullname" . }}-lease-role
+  namespace:  {{ .Release.Namespace }}
+rules:
+- apiGroups:
+    - coordination.k8s.io
+  resources:
+    - leases
+  verbs:
+    - get
+    - create
+    - watch
+    - list
+    - update
+    - delete
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: {{ include "go-kube-downscaler.fullname" . }}-lease-rolebinding
+  namespace:  {{ .Release.Namespace }}
+subjects:
+  - kind: ServiceAccount
+    name: {{ include "go-kube-downscaler.serviceAccountName" . }}
+roleRef:
+  kind: Role
+  name: {{ include "go-kube-downscaler.fullname" . }}-lease-role
+  apiGroup: rbac.authorization.k8s.io
+{{- end }}

deployments/chart/values.yaml

@@ -1,3 +1,4 @@
+# If replicaCount is greater than 1, the leader election is enabled by default
 replicaCount: 1
 
 image:

internal/api/kubernetes/client.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"log/slog"
+	"os"
 	"strings"
 	"time"
 
@@ -16,6 +17,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 )
 
 const (
@@ -33,6 +35,8 @@ type Client interface {
 	DownscaleWorkload(replicas int32, workload scalable.Workload, ctx context.Context) error
 	// UpscaleWorkload upscales the workload to the original replicas
 	UpscaleWorkload(workload scalable.Workload, ctx context.Context) error
+	// CreateLease creates a new lease for the downscaler
+	CreateLease(leaseName, leaseNamespace string) (*resourcelock.LeaseLock, error)
 	// addWorkloadEvent creates a new event on the workload
 	addWorkloadEvent(eventType string, reason string, id string, message string, workload scalable.Workload, ctx context.Context) error
 }
@@ -240,3 +244,24 @@ func (c client) addWorkloadEvent(eventType, reason, identifier, message string,
 
 	return nil
 }
+
+func (c client) CreateLease(leaseName, leaseNamespace string) (*resourcelock.LeaseLock, error) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		slog.Error("failed to get hostname", "error", err)
+		return nil, fmt.Errorf("failed to get hostname: %w", err)
+	}
+
+	lease := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseName,
+			Namespace: leaseNamespace,
+		},
+		Client: c.clientsets.Kubernetes.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: hostname,
+		},
+	}
+
+	return lease, nil
+}

internal/api/kubernetes/util.go

@@ -1,6 +1,9 @@
 package kubernetes
 
 import (
+	"fmt"
+	"os"
+
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
@@ -13,3 +16,15 @@ func getConfig(kubeconfig string) (*rest.Config, error) {
 
 	return clientcmd.BuildConfigFromFlags("", kubeconfig) //nolint: wrapcheck // error gets wrapped in the calling function, so its fine
 }
+
+// GetCurrentNamespace retrieves downscaler namespace from its service account file.
+func GetCurrentNamespace() (string, error) {
+	const namespaceFile = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+
+	namespace, err := os.ReadFile(namespaceFile)
+	if err != nil {
+		return "", fmt.Errorf("failed to read namespace file: %w", err)
+	}
+
+	return string(namespace), nil
+}

internal/pkg/util/config.go

@@ -14,6 +14,8 @@ type RuntimeConfiguration struct {
 	Debug bool
 	// Once sets if the scan should only run once.
 	Once bool
+	// LeaderElection sets if leader election should be performed.
+	LeaderElection bool
 	// Interval sets how long to wait between scans.
 	Interval time.Duration
 	// IncludeNamespaces sets the list of namespaces to restrict the downscaler to.
@@ -52,6 +54,12 @@ func (c *RuntimeConfiguration) ParseConfigFlags() {
 		false,
 		"run scan only once (default: false)",
 	)
+	flag.BoolVar(
+		&c.Once,
+		"leader-election",
+		false,
+		"enables leader election (default: false)",
+	)
 	flag.Var(
 		(*DurationValue)(&c.Interval),
 		"interval",