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 <skuznets@redhat.com>

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),

pkg/controller/operators/catalog/operator.go

@@ -204,29 +204,31 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
  // Fields are pruned from local copies of the objects managed
  // by this informer in order to reduce cached size.
  prunedCSVInformer := cache.NewSharedIndexInformer(
- pruning.NewListerWatcher(op.client, metav1.NamespaceAll, func(options *metav1.ListOptions) {
- options.LabelSelector = fmt.Sprintf("!%s", v1alpha1.CopiedLabelKey)
- }, pruning.PrunerFunc(func(csv *v1alpha1.ClusterServiceVersion) {
- *csv = v1alpha1.ClusterServiceVersion{
- TypeMeta: csv.TypeMeta,
- ObjectMeta: metav1.ObjectMeta{
- Name: csv.Name,
- Namespace: csv.Namespace,
- Labels: csv.Labels,
- Annotations: csv.Annotations,
- },
- Spec: v1alpha1.ClusterServiceVersionSpec{
- CustomResourceDefinitions: csv.Spec.CustomResourceDefinitions,
- APIServiceDefinitions: csv.Spec.APIServiceDefinitions,
- Replaces: csv.Spec.Replaces,
- Version: csv.Spec.Version,
- },
- Status: v1alpha1.ClusterServiceVersionStatus{
- Phase: csv.Status.Phase,
- Reason: csv.Status.Reason,
- },
- }
- })),
+ pruning.NewListerWatcher(op.client, metav1.NamespaceAll,
+ func(options *metav1.ListOptions) {
+ options.LabelSelector = fmt.Sprintf("!%s", v1alpha1.CopiedLabelKey)
+ },
+ pruning.PrunerFunc(func(csv *v1alpha1.ClusterServiceVersion) {
+ *csv = v1alpha1.ClusterServiceVersion{
+ TypeMeta: csv.TypeMeta,
+ ObjectMeta: metav1.ObjectMeta{
+ Name: csv.Name,
+ Namespace: csv.Namespace,
+ Labels: csv.Labels,
+ Annotations: csv.Annotations,
+ },
+ Spec: v1alpha1.ClusterServiceVersionSpec{
+ CustomResourceDefinitions: csv.Spec.CustomResourceDefinitions,
+ APIServiceDefinitions: csv.Spec.APIServiceDefinitions,
+ Replaces: csv.Spec.Replaces,
+ Version: csv.Spec.Version,
+ },
+ Status: v1alpha1.ClusterServiceVersionStatus{
+ Phase: csv.Status.Phase,
+ Reason: csv.Status.Reason,
+ },
+ }
+ })),
  &v1alpha1.ClusterServiceVersion{},
  resyncPeriod(),
  cache.Indexers{

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
@@ -127,6 +127,9 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
  if err := k8sscheme.AddToScheme(scheme); err != nil {
  return nil, err
  }
+ if err := metav1.AddMetaToScheme(scheme); err != nil {
+ return nil, err
+ }
 
  op := &Operator{
  Operator: queueOperator,
@@ -208,44 +211,20 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
  return nil, err
  }
 
- // A separate informer solely for CSV copies. Fields
- // are pruned from local copies of the objects managed
+ // A separate informer solely for CSV copies. Object metadata requests are used
  // 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},
- )
- op.copiedCSVLister = operatorsv1alpha1listers.NewClusterServiceVersionLister(copiedCSVInformer.GetIndexer())
+ func(options *metav1.ListOptions) {
+ options.LabelSelector = v1alpha1.CopiedLabelKey
+ },
+ ).Informer()
+ op.copiedCSVLister = metadatalister.New(copiedCSVInformer.GetIndexer(), gvr)
 
  // Register separate queue for gcing copied csvs
  copiedCSVGCQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("%s/csv-gc", namespace))
@@ -1195,17 +1174,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 !v1alpha1.IsCopied(csv) {
  logger.Warning("removeDanglingChild called on a parent. this is a no-op but should be avoided.")
  return nil
  }
@@ -1244,7 +1222,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,12 +1661,12 @@ 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 v1alpha1.IsCopied(clusterServiceVersion) {
  syncError = a.removeDanglingChildCSVs(clusterServiceVersion)
  return
  }