operators/olm: use a partial object metadata watch for copied CSVs

All we ever ned to know about copied CSVs is their metadata. No need to
prune objects in memory, it's better to never allocate the memory to
deserilize them in the first place.

Signed-off-by: Steve Kuznetsov <[email protected]>

cmd/olm/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/metadata"
 	"k8s.io/klog"
 	ctrl "sigs.k8s.io/controller-runtime"
 
@@ -154,6 +155,10 @@ func main() {
 	if err != nil {
 		logger.WithError(err).Fatal("error configuring custom resource client")
 	}
+	metadataClient, err := metadata.NewForConfig(config)
+	if err != nil {
+		logger.WithError(err).Fatal("error configuring metadata client")
+	}
 
 	// Create a new instance of the operator.
 	op, err := olm.NewOperator(
@@ -162,6 +167,7 @@ func main() {
 		olm.WithWatchedNamespaces(namespaces...),
 		olm.WithResyncPeriod(queueinformer.ResyncWithJitter(*wakeupInterval, 0.2)),
 		olm.WithExternalClient(crClient),
+		olm.WithMetadataClient(metadataClient),
 		olm.WithOperatorClient(opClient),
 		olm.WithRestConfig(config),
 		olm.WithConfigClient(versionedConfigClient),

go.mod

@@ -40,11 +40,11 @@ require (
 	google.golang.org/grpc v1.53.0
 	gopkg.in/yaml.v2 v2.4.0
 	helm.sh/helm/v3 v3.12.2
-	k8s.io/api v0.27.2
+	k8s.io/api v0.27.4
 	k8s.io/apiextensions-apiserver v0.27.2
-	k8s.io/apimachinery v0.27.2
+	k8s.io/apimachinery v0.27.4
 	k8s.io/apiserver v0.27.2
-	k8s.io/client-go v0.27.2
+	k8s.io/client-go v0.27.4
 	k8s.io/code-generator v0.27.2
 	k8s.io/component-base v0.27.2
 	k8s.io/klog v1.0.0

go.sum

@@ -736,7 +736,7 @@ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
-github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rubenv/sql-migrate v1.3.1 h1:Vx+n4Du8X8VTYuXbhNxdEUoh6wiJERA0GlWocR5FrbA=
 github.com/rubenv/sql-migrate v1.3.1/go.mod h1:YzG/Vh82CwyhTFXy+Mf5ahAiiEOpAlHurg+23VEzcsk=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -1343,18 +1343,18 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.27.2 h1:+H17AJpUMvl+clT+BPnKf0E3ksMAzoBBg7CntpSuADo=
-k8s.io/api v0.27.2/go.mod h1:ENmbocXfBT2ADujUXcBhHV55RIT31IIEvkntP6vZKS4=
+k8s.io/api v0.27.4 h1:0pCo/AN9hONazBKlNUdhQymmnfLRbSZjd5H5H3f0bSs=
+k8s.io/api v0.27.4/go.mod h1:O3smaaX15NfxjzILfiln1D8Z3+gEYpjEpiNA/1EVK1Y=
 k8s.io/apiextensions-apiserver v0.27.2 h1:iwhyoeS4xj9Y7v8YExhUwbVuBhMr3Q4bd/laClBV6Bo=
 k8s.io/apiextensions-apiserver v0.27.2/go.mod h1:Oz9UdvGguL3ULgRdY9QMUzL2RZImotgxvGjdWRq6ZXQ=
-k8s.io/apimachinery v0.27.2 h1:vBjGaKKieaIreI+oQwELalVG4d8f3YAMNpWLzDXkxeg=
-k8s.io/apimachinery v0.27.2/go.mod h1:XNfZ6xklnMCOGGFNqXG7bUrQCoR04dh/E7FprV6pb+E=
+k8s.io/apimachinery v0.27.4 h1:CdxflD4AF61yewuid0fLl6bM4a3q04jWel0IlP+aYjs=
+k8s.io/apimachinery v0.27.4/go.mod h1:XNfZ6xklnMCOGGFNqXG7bUrQCoR04dh/E7FprV6pb+E=
 k8s.io/apiserver v0.27.2 h1:p+tjwrcQEZDrEorCZV2/qE8osGTINPuS5ZNqWAvKm5E=
 k8s.io/apiserver v0.27.2/go.mod h1:EsOf39d75rMivgvvwjJ3OW/u9n1/BmUMK5otEOJrb1Y=
 k8s.io/cli-runtime v0.27.2 h1:9HI8gfReNujKXt16tGOAnb8b4NZ5E+e0mQQHKhFGwYw=
 k8s.io/cli-runtime v0.27.2/go.mod h1:9UecpyPDTkhiYY4d9htzRqN+rKomJgyb4wi0OfrmCjw=
-k8s.io/client-go v0.27.2 h1:vDLSeuYvCHKeoQRhCXjxXO45nHVv2Ip4Fe0MfioMrhE=
-k8s.io/client-go v0.27.2/go.mod h1:tY0gVmUsHrAmjzHX9zs7eCjxcBsf8IiNe7KQ52biTcQ=
+k8s.io/client-go v0.27.4 h1:vj2YTtSJ6J4KxaC88P4pMPEQECWMY8gqPqsTgUKzvjk=
+k8s.io/client-go v0.27.4/go.mod h1:ragcly7lUlN0SRPk5/ZkGnDjPknzb37TICq07WhI6Xc=
 k8s.io/code-generator v0.27.2 h1:RmK0CnU5qRaK6WRtSyWNODmfTZNoJbrizpVcsgbtrvI=
 k8s.io/code-generator v0.27.2/go.mod h1:DPung1sI5vBgn4AGKtlPRQAyagj/ir/4jI55ipZHVww=
 k8s.io/component-base v0.27.2 h1:neju+7s/r5O4x4/txeUONNTS9r1HsPbyoPBAtHsDCpo=

pkg/controller/operators/olm/config.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
+	"k8s.io/client-go/metadata"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -29,6 +30,7 @@ type operatorConfig struct {
 	clock                        utilclock.Clock
 	logger                       *logrus.Logger
 	operatorClient               operatorclient.ClientInterface
+	metadataClient               metadata.Interface
 	externalClient               versioned.Interface
 	strategyResolver             install.StrategyResolverInterface
 	apiReconciler                APIIntersectionReconciler
@@ -159,6 +161,12 @@ func WithOperatorClient(operatorClient operatorclient.ClientInterface) OperatorO
 	}
 }
 
+func WithMetadataClient(metadataClient metadata.Interface) OperatorOption {
+	return func(config *operatorConfig) {
+		config.metadataClient = metadataClient
+	}
+}
+
 func WithExternalClient(externalClient versioned.Interface) OperatorOption {
 	return func(config *operatorConfig) {
 		config.externalClient = externalClient

pkg/controller/operators/olm/operator.go

@@ -24,6 +24,8 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/informers"
 	k8sscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/metadata/metadatainformer"
+	"k8s.io/client-go/metadata/metadatalister"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -35,12 +37,10 @@ import (
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
-	operatorsv1alpha1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/certs"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install"
-	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/internal/pruning"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/olm/overrides"
-	resolver "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/clients"
 	csvutility "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/csv"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/event"
@@ -75,7 +75,7 @@ type Operator struct {
 	client                       versioned.Interface
 	lister                       operatorlister.OperatorLister
 	protectedCopiedCSVNamespaces map[string]struct{}
-	copiedCSVLister              operatorsv1alpha1listers.ClusterServiceVersionLister
+	copiedCSVLister              metadatalister.Lister
 	ogQueueSet                   *queueinformer.ResourceQueueSet
 	csvQueueSet                  *queueinformer.ResourceQueueSet
 	olmConfigQueue               workqueue.RateLimitingInterface
@@ -211,51 +211,28 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 		// A separate informer solely for CSV copies. Fields
 		// are pruned from local copies of the objects managed
 		// by this informer in order to reduce cached size.
-		copiedCSVInformer := cache.NewSharedIndexInformer(
-			pruning.NewListerWatcher(
-				op.client,
-				namespace,
-				func(opts *metav1.ListOptions) {
-					opts.LabelSelector = v1alpha1.CopiedLabelKey
-				},
-				pruning.PrunerFunc(func(csv *v1alpha1.ClusterServiceVersion) {
-					nonstatus, status := copyableCSVHash(csv)
-					*csv = v1alpha1.ClusterServiceVersion{
-						TypeMeta:   csv.TypeMeta,
-						ObjectMeta: csv.ObjectMeta,
-						Status: v1alpha1.ClusterServiceVersionStatus{
-							Phase:  csv.Status.Phase,
-							Reason: csv.Status.Reason,
-						},
-					}
-					if csv.Annotations == nil {
-						csv.Annotations = make(map[string]string, 2)
-					}
-					// These annotation keys are
-					// intentionally invalid -- all writes
-					// to copied CSVs are regenerated from
-					// the corresponding non-copied CSV,
-					// so it should never be transmitted
-					// back to the API server.
-					csv.Annotations["$copyhash-nonstatus"] = nonstatus
-					csv.Annotations["$copyhash-status"] = status
-				}),
-			),
-			&v1alpha1.ClusterServiceVersion{},
+		gvr := v1alpha1.SchemeGroupVersion.WithResource("clusterserviceversions")
+		copiedCSVInformer := metadatainformer.NewFilteredMetadataInformer(
+			config.metadataClient,
+			gvr,
+			namespace,
 			config.resyncPeriod(),
 			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+			func(options *metav1.ListOptions) {
+				options.LabelSelector = v1alpha1.CopiedLabelKey
+			},
 		)
-		op.copiedCSVLister = operatorsv1alpha1listers.NewClusterServiceVersionLister(copiedCSVInformer.GetIndexer())
+		op.copiedCSVLister = metadatalister.New(copiedCSVInformer.Informer().GetIndexer(), gvr)
 
 		// Register separate queue for gcing copied csvs
 		copiedCSVGCQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("%s/csv-gc", namespace))
 		op.copiedCSVGCQueueSet.Set(namespace, copiedCSVGCQueue)
 		copiedCSVGCQueueInformer, err := queueinformer.NewQueueInformer(
 			ctx,
-			queueinformer.WithInformer(copiedCSVInformer),
+			queueinformer.WithInformer(copiedCSVInformer.Informer()),
 			queueinformer.WithLogger(op.logger),
 			queueinformer.WithQueue(copiedCSVGCQueue),
-			queueinformer.WithIndexer(copiedCSVInformer.GetIndexer()),
+			queueinformer.WithIndexer(copiedCSVInformer.Informer().GetIndexer()),
 			queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncGcCsv).ToSyncer()),
 		)
 		if err != nil {
@@ -1195,17 +1172,16 @@ func (a *Operator) handleClusterServiceVersionDeletion(obj interface{}) {
 	}
 }
 
-func (a *Operator) removeDanglingChildCSVs(csv *v1alpha1.ClusterServiceVersion) error {
+func (a *Operator) removeDanglingChildCSVs(csv *metav1.PartialObjectMetadata) error {
 	logger := a.logger.WithFields(logrus.Fields{
 		"id":          queueinformer.NewLoopID(),
 		"csv":         csv.GetName(),
 		"namespace":   csv.GetNamespace(),
-		"phase":       csv.Status.Phase,
 		"labels":      csv.GetLabels(),
 		"annotations": csv.GetAnnotations(),
 	})
 
-	if !csv.IsCopied() {
+	if !IsCopied(csv) {
 		logger.Warning("removeDanglingChild called on a parent. this is a no-op but should be avoided.")
 		return nil
 	}
@@ -1244,7 +1220,7 @@ func (a *Operator) removeDanglingChildCSVs(csv *v1alpha1.ClusterServiceVersion)
 	return nil
 }
 
-func (a *Operator) deleteChild(csv *v1alpha1.ClusterServiceVersion, logger *logrus.Entry) error {
+func (a *Operator) deleteChild(csv *metav1.PartialObjectMetadata, logger *logrus.Entry) error {
 	logger.Debug("gcing csv")
 	return a.client.OperatorsV1alpha1().ClusterServiceVersions(csv.GetNamespace()).Delete(context.TODO(), csv.GetName(), metav1.DeleteOptions{})
 }
@@ -1683,18 +1659,23 @@ func (a *Operator) createCSVCopyingDisabledEvent(csv *v1alpha1.ClusterServiceVer
 }
 
 func (a *Operator) syncGcCsv(obj interface{}) (syncError error) {
-	clusterServiceVersion, ok := obj.(*v1alpha1.ClusterServiceVersion)
+	clusterServiceVersion, ok := obj.(*metav1.PartialObjectMetadata)
 	if !ok {
 		a.logger.Debugf("wrong type: %#v", obj)
 		return fmt.Errorf("casting ClusterServiceVersion failed")
 	}
-	if clusterServiceVersion.IsCopied() {
+	if IsCopied(clusterServiceVersion) {
 		syncError = a.removeDanglingChildCSVs(clusterServiceVersion)
 		return
 	}
 	return
 }
 
+func IsCopied(o metav1.Object) bool {
+	_, ok := o.GetLabels()[v1alpha1.CopiedLabelKey]
+	return ok
+}
+
 // operatorGroupFromAnnotations returns the OperatorGroup for the CSV only if the CSV is active one in the group
 func (a *Operator) operatorGroupFromAnnotations(logger *logrus.Entry, csv *v1alpha1.ClusterServiceVersion) *operatorsv1.OperatorGroup {
 	annotations := csv.GetAnnotations()

pkg/controller/operators/olm/operatorgroup.go

@@ -12,7 +12,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	meta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/errors"
@@ -797,7 +797,7 @@ func (a *Operator) copyToNamespace(prototype *v1alpha1.ClusterServiceVersion, ns
 	prototype.ResourceVersion = ""
 	prototype.UID = ""
 
-	existing, err := a.copiedCSVLister.ClusterServiceVersions(nsTo).Get(prototype.GetName())
+	existing, err := a.copiedCSVLister.Namespace(nsTo).Get(prototype.GetName())
 	if apierrors.IsNotFound(err) {
 		created, err := a.client.OperatorsV1alpha1().ClusterServiceVersions(nsTo).Create(context.TODO(), prototype, metav1.CreateOptions{})
 		if err != nil {
@@ -824,38 +824,39 @@ func (a *Operator) copyToNamespace(prototype *v1alpha1.ClusterServiceVersion, ns
 	existingNonStatus := existing.Annotations["$copyhash-nonstatus"]
 	existingStatus := existing.Annotations["$copyhash-status"]
 
+	var updated *v1alpha1.ClusterServiceVersion
 	if existingNonStatus != nonstatus {
-		if existing, err = a.client.OperatorsV1alpha1().ClusterServiceVersions(nsTo).Update(context.TODO(), prototype, metav1.UpdateOptions{}); err != nil {
+		if updated, err = a.client.OperatorsV1alpha1().ClusterServiceVersions(nsTo).Update(context.TODO(), prototype, metav1.UpdateOptions{}); err != nil {
 			return nil, err
 		}
 	} else {
 		// Avoid mutating cached copied CSV.
-		existing = prototype
+		updated = prototype
 	}
 
 	if existingStatus != status {
-		existing.Status = prototype.Status
-		if _, err = a.client.OperatorsV1alpha1().ClusterServiceVersions(nsTo).UpdateStatus(context.TODO(), existing, metav1.UpdateOptions{}); err != nil {
+		updated.Status = prototype.Status
+		if _, err = a.client.OperatorsV1alpha1().ClusterServiceVersions(nsTo).UpdateStatus(context.TODO(), updated, metav1.UpdateOptions{}); err != nil {
 			return nil, err
 		}
 	}
 	return &v1alpha1.ClusterServiceVersion{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      existing.Name,
-			Namespace: existing.Namespace,
-			UID:       existing.UID,
+			Name:      updated.Name,
+			Namespace: updated.Namespace,
+			UID:       updated.UID,
 		},
 	}, nil
 }
 
 func (a *Operator) pruneFromNamespace(operatorGroupName, namespace string) error {
-	fetchedCSVs, err := a.copiedCSVLister.ClusterServiceVersions(namespace).List(labels.Everything())
+	fetchedCSVs, err := a.copiedCSVLister.Namespace(namespace).List(labels.Everything())
 	if err != nil {
 		return err
 	}
 
 	for _, csv := range fetchedCSVs {
-		if csv.IsCopied() && csv.GetAnnotations()[operatorsv1.OperatorGroupAnnotationKey] == operatorGroupName {
+		if IsCopied(csv) && csv.GetAnnotations()[operatorsv1.OperatorGroupAnnotationKey] == operatorGroupName {
 			a.logger.Debugf("Found CSV '%v' in namespace %v to delete", csv.GetName(), namespace)
 			if err := a.copiedCSVGCQueueSet.Requeue(csv.GetNamespace(), csv.GetName()); err != nil {
 				return err