Start and stop upgrade-check using controller-runtime

Blocking functions can be added to a controller-runtime Manager so that
they start after caches have started and synced. They also stop before
caches have stopped.

Author: cbandy

Makefile

@@ -101,7 +101,7 @@ deploy-dev: build-postgres-operator createnamespaces
 	hack/create-kubeconfig.sh postgres-operator pgo
 	env \
 		CRUNCHY_DEBUG=true \
-		CHECK_FOR_UPGRADES=false \
+		CHECK_FOR_UPGRADES='$(if $(CHECK_FOR_UPGRADES),$(CHECK_FOR_UPGRADES),false)' \
 		KUBECONFIG=hack/.kube/postgres-operator/pgo \
 		$(shell $(PGO_KUBE_CLIENT) kustomize ./config/dev | \
 			sed -ne '/^kind: Deployment/,/^---/ { \

cmd/postgres-operator/main.go

@@ -82,39 +82,38 @@ func main() {
 	mgr, err := runtime.CreateRuntimeManager(os.Getenv("PGO_TARGET_NAMESPACE"), cfg, false)
 	assertNoError(err)
 
+	openshift := isOpenshift(ctx, cfg)
+
 	// add all PostgreSQL Operator controllers to the runtime manager
-	err = addControllersToManager(ctx, mgr)
+	err = addControllersToManager(mgr, openshift)
 	assertNoError(err)
 
-	log.Info("starting controller runtime manager and will wait for signal to exit")
-
 	// Enable upgrade checking
 	upgradeCheckingDisabled := strings.EqualFold(os.Getenv("CHECK_FOR_UPGRADES"), "false")
 	if !upgradeCheckingDisabled {
 		log.Info("upgrade checking enabled")
 		// get the URL for the check for upgrades endpoint if set in the env
-		upgradeCheckURL := os.Getenv("CHECK_FOR_UPGRADES_URL")
-		go upgradecheck.CheckForUpgradesScheduler(ctx, versionString, upgradeCheckURL,
-			mgr.GetClient(), mgr.GetConfig(), isOpenshift(ctx, mgr.GetConfig()),
-			mgr.GetCache(),
-		)
+		assertNoError(upgradecheck.ManagedScheduler(mgr,
+			openshift, os.Getenv("CHECK_FOR_UPGRADES_URL"), versionString))
 	} else {
 		log.Info("upgrade checking disabled")
 	}
 
+	log.Info("starting controller runtime manager and will wait for signal to exit")
+
 	assertNoError(mgr.Start(ctx))
 	log.Info("signal received, exiting")
 }
 
 // addControllersToManager adds all PostgreSQL Operator controllers to the provided controller
 // runtime manager.
-func addControllersToManager(ctx context.Context, mgr manager.Manager) error {
+func addControllersToManager(mgr manager.Manager, openshift bool) error {
 	r := &postgrescluster.Reconciler{
 		Client:      mgr.GetClient(),
 		Owner:       postgrescluster.ControllerName,
 		Recorder:    mgr.GetEventRecorderFor(postgrescluster.ControllerName),
 		Tracer:      otel.Tracer(postgrescluster.ControllerName),
-		IsOpenShift: isOpenshift(ctx, mgr.GetConfig()),
+		IsOpenShift: openshift,
 	}
 	return r.SetupWithManager(mgr)
 }

internal/upgradecheck/http.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	"github.com/crunchydata/postgres-operator/internal/logging"
 )
@@ -51,10 +52,6 @@ const (
 	upgradeCheckURL = "https://operator-maestro.crunchydata.com/pgo-versions"
 )
 
-var (
-	upgradeCheckPeriod = 24 * time.Hour
-)
-
 type HTTPClient interface {
 	Do(req *http.Request) (*http.Response, error)
 }
@@ -135,48 +132,53 @@ func checkForUpgrades(ctx context.Context, url, versionString string, backoff wa
 	return string(bodyBytes), req.Header.Get(clientHeader), err
 }
 
-// CheckForUpgradesScheduler invokes the check func when the operator starts
-// and then on the given period schedule. It stops when the context is cancelled.
-func CheckForUpgradesScheduler(ctx context.Context,
-	versionString, url string, crclient crclient.Client,
-	cfg *rest.Config, isOpenShift bool,
-	cacheClient CacheWithWait,
-) {
-	log := logging.FromContext(ctx)
+type CheckForUpgradesScheduler struct {
+	Client crclient.Client
+	Config *rest.Config
+
+	OpenShift    bool
+	Refresh      time.Duration
+	URL, Version string
+}
 
+// ManagedScheduler creates a [CheckForUpgradesScheduler] and adds it to m.
+func ManagedScheduler(m manager.Manager, openshift bool, url, version string) error {
 	if url == "" {
 		url = upgradeCheckURL
 	}
 
-	// Since we pass the client to this function before we start the manager
-	// in cmd/postgres-operator/main.go, we want to make sure cache is synced
-	// before using the client.
-	// If the cache fails to sync, that probably indicates a more serious problem
-	// with the manager starting, so we don't have to worry about restarting or retrying
-	// this process -- simply log and return
-	if synced := cacheClient.WaitForCacheSync(ctx); !synced {
-		log.V(1).Info("unable to sync cache for upgrade check")
-		return
-	}
+	return m.Add(&CheckForUpgradesScheduler{
+		Client:    m.GetClient(),
+		Config:    m.GetConfig(),
+		OpenShift: openshift,
+		Refresh:   24 * time.Hour,
+		URL:       url,
+		Version:   version,
+	})
+}
+
+// NeedLeaderElection returns true so that s runs only on the single
+// [manager.Manager] that is elected leader in the Kubernetes cluster.
+func (s *CheckForUpgradesScheduler) NeedLeaderElection() bool { return true }
+
+// Start checks for upgrades periodically. It blocks until ctx is cancelled.
+func (s *CheckForUpgradesScheduler) Start(ctx context.Context) error {
+	s.check(ctx)
 
-	check(ctx, versionString, url, crclient, cfg, isOpenShift)
+	ticker := time.NewTicker(s.Refresh)
+	defer ticker.Stop()
 
-	ticker := time.NewTicker(upgradeCheckPeriod)
 	for {
 		select {
 		case <-ticker.C:
-			check(ctx, versionString, url, crclient, cfg, isOpenShift)
+			s.check(ctx)
 		case <-ctx.Done():
-			ticker.Stop()
-			return
+			return ctx.Err()
 		}
 	}
 }
 
-func check(ctx context.Context,
-	versionString, url string, crclient crclient.Client,
-	cfg *rest.Config, isOpenShift bool,
-) {
+func (s *CheckForUpgradesScheduler) check(ctx context.Context) {
 	log := logging.FromContext(ctx)
 
 	defer func() {
@@ -186,7 +188,7 @@ func check(ctx context.Context,
 	}()
 
 	info, header, err := checkForUpgrades(ctx,
-		url, versionString, backoff, crclient, cfg, isOpenShift)
+		s.URL, s.Version, backoff, s.Client, s.Config, s.OpenShift)
 
 	if err != nil {
 		log.V(1).Info("could not complete upgrade check", "response", err.Error())

internal/upgradecheck/http_test.go

@@ -30,6 +30,7 @@ import (
 	"gotest.tools/v3/assert"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	"github.com/crunchydata/postgres-operator/internal/logging"
 	"github.com/crunchydata/postgres-operator/internal/testing/cmp"
@@ -56,14 +57,6 @@ func (m *MockClient) Do(req *http.Request) (*http.Response, error) {
 	return funcFoo()
 }
 
-type MockCacheClient struct {
-	works bool
-}
-
-func (cc *MockCacheClient) WaitForCacheSync(ctx context.Context) bool {
-	return cc.works
-}
-
 func TestCheckForUpgrades(t *testing.T) {
 	fakeClient := setupFakeClientWithPGOScheme(t, false)
 	ctx := logging.NewContext(context.Background(), logging.Discard())
@@ -168,11 +161,9 @@ func TestCheckForUpgradesScheduler(t *testing.T) {
 	_, server := setupVersionServer(t, true)
 	defer server.Close()
 	cfg := &rest.Config{Host: server.URL}
-	const testUpgradeCheckURL = "http://localhost:8080"
 
 	t.Run("panic from checkForUpgrades doesn't bubble up", func(t *testing.T) {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
+		ctx := context.Background()
 
 		// capture logs
 		var calls []string
@@ -187,44 +178,18 @@ func TestCheckForUpgradesScheduler(t *testing.T) {
 			panic(fmt.Errorf("oh no!"))
 		}
 
-		go CheckForUpgradesScheduler(ctx, "4.7.3", testUpgradeCheckURL, fakeClient, cfg, false,
-			&MockCacheClient{works: true})
-		time.Sleep(1 * time.Second)
-		cancel()
+		s := CheckForUpgradesScheduler{
+			Client: fakeClient,
+			Config: cfg,
+		}
+		s.check(ctx)
 
-		// Sleeping leads to some non-deterministic results, but we expect at least 1 execution
-		// plus one log for the failure to apply the configmap
-		assert.Assert(t, len(calls) >= 2)
+		assert.Equal(t, len(calls), 2)
 		assert.Assert(t, cmp.Contains(calls[1], `encountered panic in upgrade check`))
 	})
 
-	t.Run("cache sync fail leads to log and exit", func(t *testing.T) {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-
-		// capture logs
-		var calls []string
-		ctx = logging.NewContext(ctx, funcr.NewJSON(func(object string) {
-			calls = append(calls, object)
-		}, funcr.Options{
-			Verbosity: 1,
-		}))
-
-		// Set loop time to 1s and sleep for 2s before sending the done signal -- though the cache sync
-		// failure will exit the func before the sleep ends
-		upgradeCheckPeriod = 1 * time.Second
-		go CheckForUpgradesScheduler(ctx, "4.7.3", testUpgradeCheckURL, fakeClient, cfg, false,
-			&MockCacheClient{works: false})
-		time.Sleep(2 * time.Second)
-		cancel()
-
-		assert.Assert(t, len(calls) == 1)
-		assert.Assert(t, cmp.Contains(calls[0], `unable to sync cache for upgrade check`))
-	})
-
 	t.Run("successful log each loop, ticker works", func(t *testing.T) {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
+		ctx := context.Background()
 
 		// capture logs
 		var calls []string
@@ -244,11 +209,14 @@ func TestCheckForUpgradesScheduler(t *testing.T) {
 		}
 
 		// Set loop time to 1s and sleep for 2s before sending the done signal
-		upgradeCheckPeriod = 1 * time.Second
-		go CheckForUpgradesScheduler(ctx, "4.7.3", testUpgradeCheckURL, fakeClient, cfg, false,
-			&MockCacheClient{works: true})
-		time.Sleep(2 * time.Second)
-		cancel()
+		ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
+		defer cancel()
+		s := CheckForUpgradesScheduler{
+			Client:  fakeClient,
+			Config:  cfg,
+			Refresh: 1 * time.Second,
+		}
+		assert.ErrorIs(t, context.DeadlineExceeded, s.Start(ctx))
 
 		// Sleeping leads to some non-deterministic results, but we expect at least 2 executions
 		// plus one log for the failure to apply the configmap
@@ -258,3 +226,11 @@ func TestCheckForUpgradesScheduler(t *testing.T) {
 		assert.Assert(t, cmp.Contains(calls[3], `{\"pgo_versions\":[{\"tag\":\"v5.0.4\"},{\"tag\":\"v5.0.3\"},{\"tag\":\"v5.0.2\"},{\"tag\":\"v5.0.1\"},{\"tag\":\"v5.0.0\"}]}`))
 	})
 }
+
+func TestCheckForUpgradesSchedulerLeaderOnly(t *testing.T) {
+	// CheckForUpgradesScheduler should implement this interface.
+	var s manager.LeaderElectionRunnable = new(CheckForUpgradesScheduler)
+
+	assert.Assert(t, s.NeedLeaderElection(),
+		"expected to only run on the leader")
+}