✨ wire up ServiceAccount based caching layer (#1074)

* wire up serviceaccount based caching layer

Signed-off-by: everettraven <everettraven@gmail.com>

* remove body close

Signed-off-by: everettraven <everettraven@gmail.com>

---------

Signed-off-by: everettraven <everettraven@gmail.com>

Author: everettraven

cmd/manager/main.go

@@ -50,6 +50,7 @@ import (
 	"github.com/operator-framework/operator-controller/internal/authentication"
 	"github.com/operator-framework/operator-controller/internal/catalogmetadata/cache"
 	catalogclient "github.com/operator-framework/operator-controller/internal/catalogmetadata/client"
+	"github.com/operator-framework/operator-controller/internal/contentmanager"
 	"github.com/operator-framework/operator-controller/internal/controllers"
 	"github.com/operator-framework/operator-controller/internal/httputil"
 	"github.com/operator-framework/operator-controller/internal/labels"
@@ -180,12 +181,14 @@ func main() {
 			Name:      cExt.Spec.ServiceAccount.Name,
 			Namespace: cExt.Spec.InstallNamespace,
 		}
-		token, err := tokenGetter.Get(ctx, namespacedName)
-		if err != nil {
-			return nil, fmt.Errorf("failed to extract SA token, %w", err)
-		}
 		tempConfig := rest.AnonymousClientConfig(c)
-		tempConfig.BearerToken = token
+		tempConfig.WrapTransport = func(rt http.RoundTripper) http.RoundTripper {
+			return &authentication.TokenInjectingRoundTripper{
+				Tripper:     rt,
+				TokenGetter: tokenGetter,
+				Key:         namespacedName,
+			}
+		}
 		return tempConfig, nil
 	}
 	cfgGetter, err := helmclient.NewActionConfigGetter(mgr.GetConfig(), mgr.GetRESTMapper(),
@@ -201,7 +204,6 @@ func main() {
 	acg, err := action.NewWrappedActionClientGetter(cfgGetter,
 		helmclient.WithFailureRollbacks(false),
 	)
-
 	if err != nil {
 		setupLog.Error(err, "unable to create helm client")
 		os.Exit(1)
@@ -272,6 +274,7 @@ func main() {
 		InstalledBundleGetter: &controllers.DefaultInstalledBundleGetter{ActionClientGetter: acg},
 		Finalizers:            clusterExtensionFinalizers,
 		Preflights:            preflights,
+		Watcher:               contentmanager.New(restConfigMapper, mgr.GetConfig(), mgr.GetRESTMapper()),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ClusterExtension")
 		os.Exit(1)

config/base/rbac/role.yaml

@@ -4,12 +4,6 @@ kind: ClusterRole
 metadata:
   name: manager-role
 rules:
-- apiGroups:
-  - '*'
-  resources:
-  - '*'
-  verbs:
-  - '*'
 - apiGroups:
   - apiextensions.k8s.io
   resources:

hack/test/pre-upgrade-setup.sh

@@ -48,11 +48,60 @@ metadata:
   name: upgrade-e2e
 rules:
   - apiGroups:
-    - "*"
+    - ""
     resources:
-    - "*"
+    - "secrets"
+    - "services"
+    - "serviceaccounts"
     verbs:
-    - "*"
+    - "create"
+    - "update"
+    - "patch"
+    - "delete"
+    - "get"
+    - "list"
+    - "watch"
+  - apiGroups:
+    - "apps"
+    resources:
+    - "deployments"
+    verbs:
+    - "create"
+    - "update"
+    - "patch"
+    - "delete"
+    - "get"
+    - "list"
+    - "watch"
+  - apiGroups:
+    - "apiextensions.k8s.io"
+    resources:
+    - "customresourcedefinitions"
+    verbs:
+    - "create"
+    - "update"
+    - "patch"
+    - "delete"
+    - "get"
+    - "list"
+    - "watch"
+  - apiGroups:
+    - "rbac.authorization.k8s.io"
+    resources:
+    - "clusterroles"
+    - "clusterrolebindings"
+    - "roles"
+    - "rolebindings"
+    verbs:
+    - "create"
+    - "update"
+    - "patch"
+    - "delete"
+    - "get"
+    - "list"
+    - "watch"
+    - "bind"
+    - "escalate"
 EOF
 
 kubectl apply -f - <<EOF

internal/authentication/tokengetter.go

@@ -100,3 +100,9 @@ func (t *TokenGetter) reapExpiredTokens() {
 		}
 	}
 }
+
+func (t *TokenGetter) Delete(key types.NamespacedName) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	delete(t.tokens, key)
+}

internal/authentication/tripper.go

@@ -0,0 +1,38 @@
+package authentication
+
+import (
+	"fmt"
+	"net/http"
+
+	"k8s.io/apimachinery/pkg/types"
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+)
+
+var _ http.RoundTripper = (*TokenInjectingRoundTripper)(nil)
+
+type TokenInjectingRoundTripper struct {
+	Tripper     http.RoundTripper
+	TokenGetter *TokenGetter
+	Key         types.NamespacedName
+}
+
+func (tt *TokenInjectingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := tt.do(req)
+	if resp != nil && resp.StatusCode == http.StatusUnauthorized {
+		tt.TokenGetter.Delete(tt.Key)
+		resp, err = tt.do(req)
+	}
+	return resp, err
+}
+
+func (tt *TokenInjectingRoundTripper) do(req *http.Request) (*http.Response, error) {
+	reqClone := utilnet.CloneRequest(req)
+	token, err := tt.TokenGetter.Get(reqClone.Context(), tt.Key)
+	if err != nil {
+		return nil, err
+	}
+
+	// Always set the Authorization header to our retrieved token
+	reqClone.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
+	return tt.Tripper.RoundTrip(reqClone)
+}

internal/contentmanager/contentmanager.go

@@ -15,7 +15,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/operator-framework/operator-controller/api/v1alpha1"
@@ -156,7 +158,14 @@ func (i *instance) Watch(ctx context.Context, ctrl controller.Controller, ce *v1
 					scheme,
 					i.mapper,
 					ce,
+					handler.OnlyControllerOwner(),
 				),
+				predicate.Funcs{
+					CreateFunc:  func(tce event.TypedCreateEvent[client.Object]) bool { return false },
+					UpdateFunc:  func(tue event.TypedUpdateEvent[client.Object]) bool { return true },
+					DeleteFunc:  func(tde event.TypedDeleteEvent[client.Object]) bool { return true },
+					GenericFunc: func(tge event.TypedGenericEvent[client.Object]) bool { return true },
+				},
 			),
 		)
 		if err != nil {

internal/controllers/clusterextension_controller.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"io"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -38,7 +37,6 @@ import (
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	apimachyaml "k8s.io/apimachinery/pkg/util/yaml"
@@ -53,7 +51,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	catalogd "github.com/operator-framework/catalogd/api/core/v1alpha1"
@@ -64,6 +61,7 @@ import (
 	ocv1alpha1 "github.com/operator-framework/operator-controller/api/v1alpha1"
 	"github.com/operator-framework/operator-controller/internal/bundleutil"
 	"github.com/operator-framework/operator-controller/internal/conditionsets"
+	"github.com/operator-framework/operator-controller/internal/contentmanager"
 	"github.com/operator-framework/operator-controller/internal/labels"
 	"github.com/operator-framework/operator-controller/internal/resolve"
 	"github.com/operator-framework/operator-controller/internal/rukpak/convert"
@@ -82,8 +80,7 @@ type ClusterExtensionReconciler struct {
 	Resolver              resolve.Resolver
 	Unpacker              rukpaksource.Unpacker
 	ActionClientGetter    helmclient.ActionClientGetter
-	dynamicWatchMutex     sync.RWMutex
-	dynamicWatchGVKs      sets.Set[schema.GroupVersionKind]
+	Watcher               contentmanager.Watcher
 	controller            crcontroller.Controller
 	cache                 cache.Cache
 	InstalledBundleGetter InstalledBundleGetter
@@ -117,9 +114,6 @@ type Preflight interface {
 //+kubebuilder:rbac:groups=core,resources=serviceaccounts/token,verbs=create
 //+kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get
 
-// TODO: Remove these permissions as part of resolving https://github.com/operator-framework/operator-controller/issues/975
-//+kubebuilder:rbac:groups=*,resources=*,verbs=*
-
 //+kubebuilder:rbac:groups=catalogd.operatorframework.io,resources=clustercatalogs,verbs=list;watch
 //+kubebuilder:rbac:groups=catalogd.operatorframework.io,resources=catalogmetadata,verbs=list;watch
 
@@ -132,7 +126,7 @@ func (r *ClusterExtensionReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	l.V(1).Info("reconcile starting")
 	defer l.V(1).Info("reconcile ending")
 
-	var existingExt = &ocv1alpha1.ClusterExtension{}
+	existingExt := &ocv1alpha1.ClusterExtension{}
 	if err := r.Client.Get(ctx, req.NamespacedName, existingExt); err != nil {
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
@@ -401,36 +395,17 @@ func (r *ClusterExtensionReconciler) reconcile(ctx context.Context, ext *ocv1alp
 		return ctrl.Result{}, err
 	}
 
-	for _, obj := range relObjects {
-		if err := func() error {
-			r.dynamicWatchMutex.Lock()
-			defer r.dynamicWatchMutex.Unlock()
-
-			_, isWatched := r.dynamicWatchGVKs[obj.GetObjectKind().GroupVersionKind()]
-			if !isWatched {
-				if err := r.controller.Watch(
-					source.Kind(r.cache,
-						obj,
-						crhandler.EnqueueRequestForOwner(r.Scheme(), r.RESTMapper(), ext, crhandler.OnlyControllerOwner()),
-						predicate.Funcs{
-							CreateFunc:  func(ce event.CreateEvent) bool { return false },
-							UpdateFunc:  func(ue event.UpdateEvent) bool { return true },
-							DeleteFunc:  func(de event.DeleteEvent) bool { return true },
-							GenericFunc: func(ge event.GenericEvent) bool { return true },
-						},
-					),
-				); err != nil {
-					return err
-				}
-				r.dynamicWatchGVKs[obj.GetObjectKind().GroupVersionKind()] = sets.Empty{}
-			}
-			return nil
-		}(); err != nil {
+	// Only attempt to watch resources if we are
+	// installing / upgrading. Otherwise we may restart
+	// watches that have already been established
+	if state != stateUnchanged {
+		if err := r.Watcher.Watch(ctx, r.controller, ext, relObjects); err != nil {
 			ext.Status.InstalledBundle = nil
 			setInstalledStatusConditionFailed(ext, fmt.Sprintf("%s:%v", ocv1alpha1.ReasonInstallationFailed, err))
 			return ctrl.Result{}, err
 		}
 	}
+
 	ext.Status.InstalledBundle = bundleutil.MetadataFor(resolvedBundle.Name, *resolvedBundleVersion)
 	setInstalledStatusConditionSuccess(ext, fmt.Sprintf("Installed bundle %s successfully", resolvedBundle.Image))
 
@@ -531,13 +506,11 @@ func (r *ClusterExtensionReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				},
 			})).
 		Build(r)
-
 	if err != nil {
 		return err
 	}
 	r.controller = controller
 	r.cache = mgr.GetCache()
-	r.dynamicWatchGVKs = sets.New[schema.GroupVersionKind]()
 
 	return nil
 }