Use a Runner to watch for changes to the available Kubernetes APIs

Author: cbandy

cmd/postgres-operator/main.go

@@ -27,7 +27,6 @@ import (
 
 	"go.opentelemetry.io/otel"
 	"k8s.io/apimachinery/pkg/util/validation"
-	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 
@@ -157,6 +156,10 @@ func main() {
 	// deprecation warnings when using an older version of a resource for backwards compatibility).
 	rest.SetDefaultWarningHandler(rest.NoWarnings{})
 
+	apis, err := runtime.NewAPIDiscoveryRunner(cfg)
+	assertNoError(err)
+	assertNoError(apis.Read())
+
 	options, err := initManager()
 	assertNoError(err)
 
@@ -165,13 +168,17 @@ func main() {
 	options.BaseContext = func() context.Context {
 		ctx := context.Background()
 		ctx = feature.NewContext(ctx, features)
+		ctx = runtime.NewAPIContext(ctx, apis)
 		return ctx
 	}
 
 	mgr, err := runtime.NewManager(cfg, options)
 	assertNoError(err)
+	assertNoError(mgr.Add(apis))
 
-	openshift := isOpenshift(cfg)
+	openshift := apis.Has(runtime.API{
+		Group: "security.openshift.io", Kind: "SecurityContextConstraints",
+	})
 	if openshift {
 		log.Info("detected OpenShift environment")
 	}
@@ -275,33 +282,3 @@ func addControllersToManager(mgr runtime.Manager, openshift bool, log logging.Lo
 		os.Exit(1)
 	}
 }
-
-func isOpenshift(cfg *rest.Config) bool {
-	const sccGroupName, sccKind = "security.openshift.io", "SecurityContextConstraints"
-
-	client, err := discovery.NewDiscoveryClientForConfig(cfg)
-	assertNoError(err)
-
-	groups, err := client.ServerGroups()
-	if err != nil {
-		assertNoError(err)
-	}
-	for _, g := range groups.Groups {
-		if g.Name != sccGroupName {
-			continue
-		}
-		for _, v := range g.Versions {
-			resourceList, err := client.ServerResourcesForGroupVersion(v.GroupVersion)
-			if err != nil {
-				assertNoError(err)
-			}
-			for _, r := range resourceList.APIResources {
-				if r.Kind == sccKind {
-					return true
-				}
-			}
-		}
-	}
-
-	return false
-}

internal/controller/runtime/api_discovery.go

@@ -16,8 +16,17 @@ package runtime
 
 import (
 	"context"
+	"errors"
+	"sync"
+	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/discovery"
+	"k8s.io/client-go/rest"
+
+	"github.com/crunchydata/postgres-operator/internal/logging"
 )
 
 // API is a combination of Group, Version, and Kind that can be used to check
@@ -73,6 +82,145 @@ func (s APISet) HasOne(api ...API) bool {
 	return false
 }
 
+type APIDiscoveryRunner struct {
+	Client interface {
+		ServerGroups() (*metav1.APIGroupList, error)
+		ServerResourcesForGroupVersion(string) (*metav1.APIResourceList, error)
+	}
+
+	refresh time.Duration
+
+	want []API
+	have struct {
+		sync.RWMutex
+		APISet
+	}
+}
+
+// NewAPIDiscoveryRunner creates an [APIDiscoveryRunner] that periodically reads
+// what APIs are available in the Kubernetes at config.
+func NewAPIDiscoveryRunner(config *rest.Config) (*APIDiscoveryRunner, error) {
+	dc, err := discovery.NewDiscoveryClientForConfig(config)
+
+	runner := &APIDiscoveryRunner{
+		Client:  dc,
+		refresh: 10 * time.Minute,
+		want: []API{
+			{Group: "cert-manager.io", Kind: "Certificate"},
+			{Group: "gateway.networking.k8s.io", Kind: "ReferenceGrant"},
+			{Group: "security.openshift.io", Kind: "SecurityContextConstraints"},
+			{Group: "snapshot.storage.k8s.io", Kind: "VolumeSnapshot"},
+			{Group: "trust.cert-manager.io", Kind: "Bundle"},
+		},
+	}
+
+	return runner, err
+}
+
+// NeedLeaderElection returns false so that r runs on any [manager.Manager],
+// regardless of which is elected leader in the Kubernetes namespace.
+func (r *APIDiscoveryRunner) NeedLeaderElection() bool { return false }
+
+// Read fetches available APIs from Kubernetes.
+func (r *APIDiscoveryRunner) Read() error {
+
+	// Build an index of the APIs we want to know about.
+	wantAPIs := make(map[string]map[string]sets.Set[string])
+	for _, want := range r.want {
+		if wantAPIs[want.Group] == nil {
+			wantAPIs[want.Group] = make(map[string]sets.Set[string])
+		}
+		if wantAPIs[want.Group][want.Version] == nil {
+			wantAPIs[want.Group][want.Version] = sets.New[string]()
+		}
+		if want.Kind != "" {
+			wantAPIs[want.Group][want.Version].Insert(want.Kind)
+		}
+	}
+
+	// Fetch Groups and Versions from Kubernetes.
+	groups, err := r.Client.ServerGroups()
+	if err != nil {
+		return err
+	}
+
+	// Build an index of the Groups, GVs, GKs, and GVKs available in Kuberentes
+	// that we want to know about.
+	haveWantedAPIs := make(map[API]struct{})
+	for _, apiG := range groups.Groups {
+		var haveG string = apiG.Name
+		haveWantedAPIs[API{Group: haveG}] = struct{}{}
+
+		for _, apiGV := range apiG.Versions {
+			var haveV string = apiGV.Version
+			haveWantedAPIs[API{Group: haveG, Version: haveV}] = struct{}{}
+
+			// Only fetch Resources when there are Kinds we want to know about.
+			if wantAPIs[haveG][""].Len() == 0 && wantAPIs[haveG][haveV].Len() == 0 {
+				continue
+			}
+
+			resources, err := r.Client.ServerResourcesForGroupVersion(apiGV.GroupVersion)
+			if err != nil {
+				return err
+			}
+
+			for _, apiR := range resources.APIResources {
+				var haveK string = apiR.Kind
+				haveWantedAPIs[API{Group: haveG, Kind: haveK}] = struct{}{}
+				haveWantedAPIs[API{Group: haveG, Kind: haveK, Version: haveV}] = struct{}{}
+			}
+		}
+	}
+
+	r.have.Lock()
+	r.have.APISet = haveWantedAPIs
+	r.have.Unlock()
+
+	return nil
+}
+
+// Start periodically reads the Kuberentes API. It blocks until ctx is cancelled.
+func (r *APIDiscoveryRunner) Start(ctx context.Context) error {
+	ticker := time.NewTicker(r.refresh)
+	defer ticker.Stop()
+
+	log := logging.FromContext(ctx).WithValues("controller", "kubernetes")
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := r.Read(); err != nil {
+				log.Error(err, "Unable to detect Kubernetes APIs")
+			}
+		case <-ctx.Done():
+			// TODO(controller-runtime): Fixed in v0.19.0
+			// https://github.com/kubernetes-sigs/controller-runtime/issues/1927
+			if errors.Is(ctx.Err(), context.Canceled) {
+				return nil
+			}
+			return ctx.Err()
+		}
+	}
+}
+
+// Has returns true when api is available in Kuberentes.
+func (r *APIDiscoveryRunner) Has(api API) bool { return r.HasOne(api) }
+
+// HasAll returns true when every api is available in Kubernetes.
+func (r *APIDiscoveryRunner) HasAll(api ...API) bool {
+	r.have.RLock()
+	defer r.have.RUnlock()
+	return r.have.HasAll(api...)
+}
+
+// HasOne returns true when at least one api is available in Kubernetes.
+func (r *APIDiscoveryRunner) HasOne(api ...API) bool {
+	r.have.RLock()
+	defer r.have.RUnlock()
+	return r.have.HasOne(api...)
+}
+
 type apiContextKey struct{}
 
 // Kubernetes returns the APIs previously stored by [NewAPIContext].

internal/controller/runtime/api_discovery_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"gotest.tools/v3/assert"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 func TestAPISet(t *testing.T) {
@@ -74,3 +75,11 @@ func TestAPIContext(t *testing.T) {
 	set[API{Group: "snapshot.storage.k8s.io"}] = struct{}{}
 	assert.Assert(t, Kubernetes(ctx).Has(API{Group: "snapshot.storage.k8s.io"}))
 }
+
+func TestAPIDiscoveryRunnerInterfaces(t *testing.T) {
+	var _ APIs = new(APIDiscoveryRunner)
+	var _ manager.Runnable = new(APIDiscoveryRunner)
+
+	var runnable manager.LeaderElectionRunnable = new(APIDiscoveryRunner)
+	assert.Assert(t, false == runnable.NeedLeaderElection())
+}

internal/registration/runner.go

@@ -185,6 +185,7 @@ func (r *Runner) Start(ctx context.Context) error {
 				r.changed()
 			}
 		case <-ctx.Done():
+			// TODO(controller-runtime): Fixed in v0.19.0
 			// https://github.com/kubernetes-sigs/controller-runtime/issues/1927
 			if errors.Is(ctx.Err(), context.Canceled) {
 				return nil