Add leader election to PGO, including necessary RBAC for leases and tests. Return error if PGO_CONTROLLER_LEASE_NAME is invalid.

Author: dsessler7

cmd/postgres-operator/main.go

@@ -91,7 +91,7 @@ func main() {
 	// deprecation warnings when using an older version of a resource for backwards compatibility).
 	rest.SetDefaultWarningHandler(rest.NoWarnings{})
 
-	mgr, err := runtime.CreateRuntimeManager(os.Getenv("PGO_TARGET_NAMESPACE"), cfg, false)
+	mgr, err := runtime.CreateRuntimeManager(ctx, os.Getenv("PGO_TARGET_NAMESPACE"), cfg, false)
 	assertNoError(err)
 
 	openshift := isOpenshift(cfg)

config/rbac/cluster/role.yaml

@@ -88,6 +88,15 @@ rules:
   - list
   - patch
   - watch
+- apiGroups:
+  - coordination.k8s.io
+  resources:
+  - leases
+  verbs:
+  - create
+  - get
+  - update
+  - watch
 - apiGroups:
   - policy
   resources:

config/rbac/namespace/role.yaml

@@ -88,6 +88,15 @@ rules:
   - list
   - patch
   - watch
+- apiGroups:
+  - coordination.k8s.io
+  resources:
+  - leases
+  verbs:
+  - create
+  - get
+  - update
+  - watch
 - apiGroups:
   - policy
   resources:

internal/controller/postgrescluster/helpers_test.go

@@ -158,15 +158,15 @@ func testCluster() *v1beta1.PostgresCluster {
 // setupManager creates the runtime manager used during controller testing
 func setupManager(t *testing.T, cfg *rest.Config,
 	controllerSetup func(mgr manager.Manager)) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(context.Background())
 
-	mgr, err := runtime.CreateRuntimeManager("", cfg, true)
+	mgr, err := runtime.CreateRuntimeManager(ctx, "", cfg, true)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	controllerSetup(mgr)
 
-	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
 		if err := mgr.Start(ctx); err != nil {
 			t.Error(err)

internal/controller/runtime/runtime.go

@@ -16,13 +16,19 @@ limitations under the License.
 package runtime
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
@@ -48,8 +54,12 @@ var refreshInterval = 60 * time.Minute
 // controllers that will be responsible for managing PostgreSQL clusters using the
 // 'postgrescluster' custom resource.  Additionally, the manager will only watch for resources in
 // the namespace specified, with an empty string resulting in the manager watching all namespaces.
-func CreateRuntimeManager(namespace string, config *rest.Config,
+
+// +kubebuilder:rbac:groups="coordination.k8s.io",resources="leases",verbs={get,create,update}
+
+func CreateRuntimeManager(ctx context.Context, namespace string, config *rest.Config,
 	disableMetrics bool) (manager.Manager, error) {
+	log := log.FromContext(ctx)
 
 	// Watch all namespaces by default
 	options := manager.Options{
@@ -70,6 +80,14 @@ func CreateRuntimeManager(namespace string, config *rest.Config,
 		options.Metrics.BindAddress = "0"
 	}
 
+	// Add leader election options
+	options, err := addLeaderElectionOptions(options)
+	if err != nil {
+		return nil, err
+	} else {
+		log.Info("Leader election enabled.")
+	}
+
 	// create controller runtime manager
 	mgr, err := manager.New(config, options)
 	if err != nil {
@@ -81,3 +99,33 @@ func CreateRuntimeManager(namespace string, config *rest.Config,
 
 // GetConfig creates a *rest.Config for talking to a Kubernetes API server.
 func GetConfig() (*rest.Config, error) { return config.GetConfig() }
+
+// addLeaderElectionOptions takes the manager.Options as an argument and will
+// add leader election options if PGO_CONTROLLER_LEASE_NAME is set and valid.
+// If PGO_CONTROLLER_LEASE_NAME is not valid, the function will return the
+// original options and an error. If PGO_CONTROLLER_LEASE_NAME is not set at all,
+// the function will return the original options.
+func addLeaderElectionOptions(opts manager.Options) (manager.Options, error) {
+	errs := []error{}
+
+	leaderLeaseName := os.Getenv("PGO_CONTROLLER_LEASE_NAME")
+	if len(leaderLeaseName) > 0 {
+		// If no errors are returned by IsDNS1123Subdomain(), turn on leader election,
+		// otherwise, return the errors
+		dnsSubdomainErrors := validation.IsDNS1123Subdomain(leaderLeaseName)
+		if len(dnsSubdomainErrors) == 0 {
+			opts.LeaderElection = true
+			opts.LeaderElectionNamespace = os.Getenv("PGO_NAMESPACE")
+			opts.LeaderElectionID = leaderLeaseName
+		} else {
+			for _, errString := range dnsSubdomainErrors {
+				err := errors.New(errString)
+				errs = append(errs, err)
+			}
+
+			return opts, fmt.Errorf("value for PGO_CONTROLLER_LEASE_NAME is invalid: %v", errs)
+		}
+	}
+
+	return opts, nil
+}

internal/controller/runtime/runtime_test.go

@@ -0,0 +1,65 @@
+/*
+ Copyright 2021 - 2024 Crunchy Data Solutions, Inc.
+ 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 runtime
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+func TestAddLeaderElectionOptions(t *testing.T) {
+	t.Setenv("PGO_NAMESPACE", "test-namespace")
+
+	t.Run("PGO_CONTROLLER_LEASE_NAME is not set", func(t *testing.T) {
+		opts := manager.Options{HealthProbeBindAddress: "0"}
+
+		opts, err := addLeaderElectionOptions(opts)
+
+		assert.NilError(t, err)
+		assert.Assert(t, opts.HealthProbeBindAddress == "0")
+		assert.Assert(t, !opts.LeaderElection)
+		assert.Assert(t, opts.LeaderElectionNamespace == "")
+		assert.Assert(t, opts.LeaderElectionID == "")
+	})
+
+	t.Run("PGO_CONTROLLER_LEASE_NAME is invalid", func(t *testing.T) {
+		t.Setenv("PGO_CONTROLLER_LEASE_NAME", "INVALID_NAME")
+		opts := manager.Options{HealthProbeBindAddress: "0"}
+
+		opts, err := addLeaderElectionOptions(opts)
+
+		assert.ErrorContains(t, err, "value for PGO_CONTROLLER_LEASE_NAME is invalid:")
+		assert.Assert(t, opts.HealthProbeBindAddress == "0")
+		assert.Assert(t, !opts.LeaderElection)
+		assert.Assert(t, opts.LeaderElectionNamespace == "")
+		assert.Assert(t, opts.LeaderElectionID == "")
+	})
+
+	t.Run("PGO_CONTROLLER_LEASE_NAME is valid", func(t *testing.T) {
+		t.Setenv("PGO_CONTROLLER_LEASE_NAME", "valid-name")
+		opts := manager.Options{HealthProbeBindAddress: "0"}
+
+		opts, err := addLeaderElectionOptions(opts)
+
+		assert.NilError(t, err)
+		assert.Assert(t, opts.HealthProbeBindAddress == "0")
+		assert.Assert(t, opts.LeaderElection)
+		assert.Assert(t, opts.LeaderElectionNamespace == "test-namespace")
+		assert.Assert(t, opts.LeaderElectionID == "valid-name")
+	})
+}