Move controller concurrency parsing to main

Author: cbandy

cmd/postgres-operator/main.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -40,6 +41,7 @@ import (
 	"github.com/crunchydata/postgres-operator/internal/registration"
 	"github.com/crunchydata/postgres-operator/internal/upgradecheck"
 	"github.com/crunchydata/postgres-operator/internal/util"
+	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
 
 var versionString string
@@ -66,6 +68,8 @@ func initLogging() {
 //+kubebuilder:rbac:groups="coordination.k8s.io",resources="leases",verbs={get,create,update}
 
 func initManager() (runtime.Options, error) {
+	log := logging.FromContext(context.Background())
+
 	options := runtime.Options{}
 	options.Cache.SyncPeriod = initialize.Pointer(time.Hour)
 
@@ -86,6 +90,18 @@ func initManager() (runtime.Options, error) {
 		options.Cache.DefaultNamespaces = map[string]runtime.CacheConfig{namespace: {}}
 	}
 
+	options.Controller.GroupKindConcurrency = map[string]int{
+		"PostgresCluster." + v1beta1.GroupVersion.Group: 2,
+	}
+
+	if s := os.Getenv("PGO_WORKERS"); s != "" {
+		if i, err := strconv.Atoi(s); err == nil && i > 0 {
+			options.Controller.GroupKindConcurrency["PostgresCluster."+v1beta1.GroupVersion.Group] = i
+		} else {
+			log.Error(err, "PGO_WORKERS must be a positive number")
+		}
+	}
+
 	return options, nil
 }
 

cmd/postgres-operator/main_test.go

@@ -16,6 +16,7 @@ limitations under the License.
 package main
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
@@ -32,8 +33,21 @@ func TestInitManager(t *testing.T) {
 			assert.Equal(t, *options.Cache.SyncPeriod, time.Hour)
 		}
 
+		assert.DeepEqual(t, options.Controller.GroupKindConcurrency,
+			map[string]int{
+				"PostgresCluster.postgres-operator.crunchydata.com": 2,
+			})
+
 		assert.Assert(t, options.Cache.DefaultNamespaces == nil)
 		assert.Assert(t, options.LeaderElection == false)
+
+		{
+			options.Cache.SyncPeriod = nil
+			options.Controller.GroupKindConcurrency = nil
+
+			assert.Assert(t, reflect.ValueOf(options).IsZero(),
+				"expected remaining fields to be unset:\n%+v", options)
+		}
 	})
 
 	t.Run("PGO_CONTROLLER_LEASE_NAME", func(t *testing.T) {
@@ -73,4 +87,30 @@ func TestInitManager(t *testing.T) {
 			assert.Equal(t, k, "some-such")
 		}
 	})
+
+	t.Run("PGO_WORKERS", func(t *testing.T) {
+		t.Run("Invalid", func(t *testing.T) {
+			for _, v := range []string{"-3", "0", "3.14"} {
+				t.Setenv("PGO_WORKERS", v)
+
+				options, err := initManager()
+				assert.NilError(t, err)
+				assert.DeepEqual(t, options.Controller.GroupKindConcurrency,
+					map[string]int{
+						"PostgresCluster.postgres-operator.crunchydata.com": 2,
+					})
+			}
+		})
+
+		t.Run("Valid", func(t *testing.T) {
+			t.Setenv("PGO_WORKERS", "19")
+
+			options, err := initManager()
+			assert.NilError(t, err)
+			assert.DeepEqual(t, options.Controller.GroupKindConcurrency,
+				map[string]int{
+					"PostgresCluster.postgres-operator.crunchydata.com": 19,
+				})
+		})
+	})
 }

internal/controller/postgrescluster/controller.go

@@ -19,8 +19,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"os"
-	"strconv"
 
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/trace"
@@ -36,7 +34,6 @@ import (
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -457,24 +454,8 @@ func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
 		}
 	}
 
-	var opts controller.Options
-
-	// TODO(cbandy): Move this to main with controller-runtime v0.9+
-	// - https://github.com/kubernetes-sigs/controller-runtime/commit/82fc2564cf
-	if s := os.Getenv("PGO_WORKERS"); s != "" {
-		if i, err := strconv.Atoi(s); err == nil && i > 0 {
-			opts.MaxConcurrentReconciles = i
-		} else {
-			mgr.GetLogger().Error(err, "PGO_WORKERS must be a positive number")
-		}
-	}
-	if opts.MaxConcurrentReconciles == 0 {
-		opts.MaxConcurrentReconciles = 2
-	}
-
 	return builder.ControllerManagedBy(mgr).
 		For(&v1beta1.PostgresCluster{}).
-		WithOptions(opts).
 		Owns(&corev1.ConfigMap{}).
 		Owns(&corev1.Endpoints{}).
 		Owns(&corev1.PersistentVolumeClaim{}).