*: detect when all objects are labelled, restart

When all of the k8s objects that need labels have them, we are good to
exit the process. The next Pod that start up will detect that all labels
are present and be able to filter informers going forward.

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

Author: stevekuznetsov

cmd/olm/main.go

@@ -173,7 +173,7 @@ func main() {
 		olm.WithExternalClient(crClient),
 		olm.WithMetadataClient(metadataClient),
 		olm.WithOperatorClient(opClient),
-		olm.WithRestConfig(config),
+		olm.WithRestConfig(validatingConfig),
 		olm.WithConfigClient(versionedConfigClient),
 		olm.WithProtectedCopiedCSVNamespaces(*protectedCopiedCSVNamespaces),
 	)

pkg/controller/install/apiservice.go

@@ -61,6 +61,7 @@ func (i *StrategyDeploymentInstaller) createOrUpdateAPIService(caPEM []byte, des
 	if err := ownerutil.AddOwnerLabels(apiService, i.owner); err != nil {
 		return err
 	}
+	apiService.Labels[OLMManagedLabelKey] = OLMManagedLabelValue
 
 	// Create a service for the deployment
 	containerPort := int32(443)

pkg/controller/install/deployment.go

@@ -152,11 +152,11 @@ func (i *StrategyDeploymentInstaller) deploymentForSpec(name string, spec appsv1
 	dep.Spec.Template.SetAnnotations(annotations)
 
 	// Set custom labels before CSV owner labels
+	dep.SetLabels(specLabels)
 	if dep.Labels == nil {
 		dep.Labels = map[string]string{}
 	}
 	dep.Labels[OLMManagedLabelKey] = OLMManagedLabelValue
-	dep.SetLabels(specLabels)
 
 	ownerutil.AddNonBlockingOwner(dep, i.owner)
 	ownerutil.AddOwnerLabelsForKind(dep, i.owner, v1alpha1.ClusterServiceVersionKind)

pkg/controller/install/deployment_test.go

@@ -353,6 +353,7 @@ func TestInstallStrategyDeploymentCheckInstallErrors(t *testing.T) {
 			dep.Spec.Template.SetAnnotations(map[string]string{"test": "annotation"})
 			dep.Spec.RevisionHistoryLimit = &revisionHistoryLimit
 			dep.SetLabels(labels.CloneAndAddLabel(dep.ObjectMeta.GetLabels(), DeploymentSpecHashLabelKey, HashDeploymentSpec(dep.Spec)))
+			dep.Labels[OLMManagedLabelKey] = OLMManagedLabelValue
 			dep.Status.Conditions = append(dep.Status.Conditions, appsv1.DeploymentCondition{
 				Type:   appsv1.DeploymentAvailable,
 				Status: corev1.ConditionTrue,

pkg/controller/operators/catalog/operator.go

@@ -207,10 +207,10 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		ogQueueSet:               queueinformer.NewEmptyResourceQueueSet(),
 		catalogSubscriberIndexer: map[string]cache.Indexer{},
 		serviceAccountQuerier:    scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
-		clientAttenuator:         scoped.NewClientAttenuator(logger, config, opClient),
+		clientAttenuator:         scoped.NewClientAttenuator(logger, validatingConfig, opClient),
 		installPlanTimeout:       installPlanTimeout,
 		bundleUnpackTimeout:      bundleUnpackTimeout,
-		clientFactory:            clients.NewFactory(config),
+		clientFactory:            clients.NewFactory(validatingConfig),
 	}
 	op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
 	op.sourceInvalidator = resolver.SourceProviderFromRegistryClientProvider(op.sources, logger)
@@ -380,10 +380,22 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	op.lister.RbacV1().RegisterRoleLister(metav1.NamespaceAll, roleInformer.Lister())
 	sharedIndexInformers = append(sharedIndexInformers, roleInformer.Informer())
 
-	labelObjects := func(gvr schema.GroupVersionResource, informer cache.SharedIndexInformer, sync queueinformer.LegacySyncHandler) error {
+	complete := map[schema.GroupVersionResource][]bool{}
+	completeLock := &sync.Mutex{}
+
+	labelObjects := func(gvr schema.GroupVersionResource, informer cache.SharedIndexInformer, sync func(done func() bool) queueinformer.LegacySyncHandler) error {
 		if canFilter {
 			return nil
 		}
+		var idx int
+		if _, exists := complete[gvr]; exists {
+			idx = len(complete[gvr])
+			complete[gvr] = append(complete[gvr], false)
+		} else {
+			idx = 0
+			complete[gvr] = []bool{false}
+		}
+
 		queue := workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workqueue.RateLimitingQueueConfig{
 			Name: gvr.String(),
 		})
@@ -392,7 +404,18 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 			queueinformer.WithQueue(queue),
 			queueinformer.WithLogger(op.logger),
 			queueinformer.WithInformer(informer),
-			queueinformer.WithSyncer(sync.ToSyncer()),
+			queueinformer.WithSyncer(sync(func() bool {
+				completeLock.Lock()
+				complete[gvr][idx] = true
+				allDone := true
+				for _, items := range complete {
+					for _, done := range items {
+						allDone = allDone && done
+					}
+				}
+				completeLock.Unlock()
+				return allDone
+			}).ToSyncer()),
 		)
 		if err != nil {
 			return err
@@ -408,6 +431,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	rolesgvk := rbacv1.SchemeGroupVersion.WithResource("roles")
 	if err := labelObjects(rolesgvk, roleInformer.Informer(), labeller.ObjectLabeler[*rbacv1.Role, *rbacv1applyconfigurations.RoleApplyConfiguration](
 		ctx, op.logger, labeller.Filter(rolesgvk),
+		roleInformer.Lister().List,
 		rbacv1applyconfigurations.Role,
 		func(namespace string, ctx context.Context, cfg *rbacv1applyconfigurations.RoleApplyConfiguration, opts metav1.ApplyOptions) (*rbacv1.Role, error) {
 			return op.opClient.KubernetesInterface().RbacV1().Roles(namespace).Apply(ctx, cfg, opts)
@@ -420,6 +444,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		func(role *rbacv1.Role) (string, error) {
 			return resolver.PolicyRuleHashLabelValue(role.Rules)
 		},
+		roleInformer.Lister().List,
 		rbacv1applyconfigurations.Role,
 		func(namespace string, ctx context.Context, cfg *rbacv1applyconfigurations.RoleApplyConfiguration, opts metav1.ApplyOptions) (*rbacv1.Role, error) {
 			return op.opClient.KubernetesInterface().RbacV1().Roles(namespace).Apply(ctx, cfg, opts)
@@ -436,6 +461,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	rolebindingsgvk := rbacv1.SchemeGroupVersion.WithResource("rolebindings")
 	if err := labelObjects(rolebindingsgvk, roleBindingInformer.Informer(), labeller.ObjectLabeler[*rbacv1.RoleBinding, *rbacv1applyconfigurations.RoleBindingApplyConfiguration](
 		ctx, op.logger, labeller.Filter(rolebindingsgvk),
+		roleBindingInformer.Lister().List,
 		rbacv1applyconfigurations.RoleBinding,
 		func(namespace string, ctx context.Context, cfg *rbacv1applyconfigurations.RoleBindingApplyConfiguration, opts metav1.ApplyOptions) (*rbacv1.RoleBinding, error) {
 			return op.opClient.KubernetesInterface().RbacV1().RoleBindings(namespace).Apply(ctx, cfg, opts)
@@ -448,6 +474,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		func(roleBinding *rbacv1.RoleBinding) (string, error) {
 			return resolver.RoleReferenceAndSubjectHashLabelValue(roleBinding.RoleRef, roleBinding.Subjects)
 		},
+		roleBindingInformer.Lister().List,
 		rbacv1applyconfigurations.RoleBinding,
 		func(namespace string, ctx context.Context, cfg *rbacv1applyconfigurations.RoleBindingApplyConfiguration, opts metav1.ApplyOptions) (*rbacv1.RoleBinding, error) {
 			return op.opClient.KubernetesInterface().RbacV1().RoleBindings(namespace).Apply(ctx, cfg, opts)
@@ -464,6 +491,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	serviceaccountsgvk := corev1.SchemeGroupVersion.WithResource("serviceaccounts")
 	if err := labelObjects(serviceaccountsgvk, serviceAccountInformer.Informer(), labeller.ObjectLabeler[*corev1.ServiceAccount, *corev1applyconfigurations.ServiceAccountApplyConfiguration](
 		ctx, op.logger, labeller.Filter(serviceaccountsgvk),
+		serviceAccountInformer.Lister().List,
 		corev1applyconfigurations.ServiceAccount,
 		func(namespace string, ctx context.Context, cfg *corev1applyconfigurations.ServiceAccountApplyConfiguration, opts metav1.ApplyOptions) (*corev1.ServiceAccount, error) {
 			return op.opClient.KubernetesInterface().CoreV1().ServiceAccounts(namespace).Apply(ctx, cfg, opts)
@@ -480,6 +508,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	servicesgvk := corev1.SchemeGroupVersion.WithResource("services")
 	if err := labelObjects(servicesgvk, serviceInformer.Informer(), labeller.ObjectLabeler[*corev1.Service, *corev1applyconfigurations.ServiceApplyConfiguration](
 		ctx, op.logger, labeller.Filter(servicesgvk),
+		serviceInformer.Lister().List,
 		corev1applyconfigurations.Service,
 		func(namespace string, ctx context.Context, cfg *corev1applyconfigurations.ServiceApplyConfiguration, opts metav1.ApplyOptions) (*corev1.Service, error) {
 			return op.opClient.KubernetesInterface().CoreV1().Services(namespace).Apply(ctx, cfg, opts)
@@ -505,6 +534,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	podsgvk := corev1.SchemeGroupVersion.WithResource("pods")
 	if err := labelObjects(podsgvk, csPodInformer.Informer(), labeller.ObjectLabeler[*corev1.Pod, *corev1applyconfigurations.PodApplyConfiguration](
 		ctx, op.logger, labeller.Filter(podsgvk),
+		csPodInformer.Lister().List,
 		corev1applyconfigurations.Pod,
 		func(namespace string, ctx context.Context, cfg *corev1applyconfigurations.PodApplyConfiguration, opts metav1.ApplyOptions) (*corev1.Pod, error) {
 			return op.opClient.KubernetesInterface().CoreV1().Pods(namespace).Apply(ctx, cfg, opts)
@@ -542,6 +572,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		ctx, op.logger, labeller.JobFilter(func(namespace, name string) (metav1.Object, error) {
 			return configMapInformer.Lister().ConfigMaps(namespace).Get(name)
 		}),
+		jobInformer.Lister().List,
 		batchv1applyconfigurations.Job,
 		func(namespace string, ctx context.Context, cfg *batchv1applyconfigurations.JobApplyConfiguration, opts metav1.ApplyOptions) (*batchv1.Job, error) {
 			return op.opClient.KubernetesInterface().BatchV1().Jobs(namespace).Apply(ctx, cfg, opts)
@@ -617,6 +648,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	customresourcedefinitionsgvk := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions")
 	if err := labelObjects(customresourcedefinitionsgvk, crdInformer, labeller.ObjectPatchLabeler(
 		ctx, op.logger, labeller.Filter(customresourcedefinitionsgvk),
+		crdLister.List,
 		op.opClient.ApiextensionsInterface().ApiextensionsV1().CustomResourceDefinitions().Patch,
 	)); err != nil {
 		return nil, err

pkg/controller/operators/labeller/filters.go

@@ -75,7 +75,7 @@ var filters = map[schema.GroupVersionResource]func(metav1.Object) bool{
 
 func Validate(ctx context.Context, logger *logrus.Logger, metadataClient metadata.Interface) (bool, error) {
 	okLock := sync.Mutex{}
-	var ok bool
+	ok := true
 	g, ctx := errgroup.WithContext(ctx)
 	allFilters := map[schema.GroupVersionResource]func(metav1.Object) bool{}
 	for gvr, filter := range filters {
@@ -124,5 +124,6 @@ func Validate(ctx context.Context, logger *logrus.Logger, metadataClient metadat
 	if err := g.Wait(); err != nil {
 		return false, err
 	}
+	logger.WithField("canFilter", ok).Info("detected ability to filter informers")
 	return ok, nil
 }

pkg/controller/operators/labeller/labels.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
 	"strings"
 
 	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/sirupsen/logrus"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 
@@ -40,28 +42,50 @@ func ObjectLabeler[T metav1.Object, A ApplyConfig[A]](
 	ctx context.Context,
 	logger *logrus.Logger,
 	check func(metav1.Object) bool,
+	list func(options labels.Selector) ([]T, error),
 	applyConfigFor func(name, namespace string) A,
 	apply func(namespace string, ctx context.Context, cfg A, opts metav1.ApplyOptions) (T, error),
-) queueinformer.LegacySyncHandler {
-	return func(obj interface{}) error {
-		cast, ok := obj.(T)
-		if !ok {
-			err := fmt.Errorf("wrong type %T, expected %T: %#v", obj, new(T), obj)
-			logger.WithError(err).Error("casting failed")
-			return fmt.Errorf("casting failed: %w", err)
-		}
+) func(done func() bool) queueinformer.LegacySyncHandler {
+	return func(done func() bool) queueinformer.LegacySyncHandler {
+		return func(obj interface{}) error {
+			cast, ok := obj.(T)
+			if !ok {
+				err := fmt.Errorf("wrong type %T, expected %T: %#v", obj, new(T), obj)
+				logger.WithError(err).Error("casting failed")
+				return fmt.Errorf("casting failed: %w", err)
+			}
 
-		if !check(cast) || hasLabel(cast) {
-			return nil
-		}
+			if !check(cast) || hasLabel(cast) {
+				// if the object we're processing does not need us to label it, it's possible that every object that requires
+				// the label already has it; in which case we should exit the process, so the Pod that succeeds us can filter
+				// the informers used to drive the controller and stop having to track extraneous objects
+				items, err := list(labels.Everything())
+				if err != nil {
+					logger.WithError(err).Warn("failed to list all objects to check for labelling completion")
+					return nil
+				}
+				gvrFullyLabelled := true
+				for _, item := range items {
+					gvrFullyLabelled = gvrFullyLabelled && (!check(item) || hasLabel(item))
+				}
+				if gvrFullyLabelled {
+					allObjectsLabelled := done()
+					if allObjectsLabelled {
+						logrus.Info("detected that every object is labelled, exiting to re-start the process...")
+						os.Exit(0)
+					}
+				}
+				return nil
+			}
 
-		cfg := applyConfigFor(cast.GetName(), cast.GetNamespace())
-		cfg.WithLabels(map[string]string{
-			install.OLMManagedLabelKey: install.OLMManagedLabelValue,
-		})
+			cfg := applyConfigFor(cast.GetName(), cast.GetNamespace())
+			cfg.WithLabels(map[string]string{
+				install.OLMManagedLabelKey: install.OLMManagedLabelValue,
+			})
 
-		_, err := apply(cast.GetNamespace(), ctx, cfg, metav1.ApplyOptions{})
-		return err
+			_, err := apply(cast.GetNamespace(), ctx, cfg, metav1.ApplyOptions{})
+			return err
+		}
 	}
 }
 
@@ -71,58 +95,77 @@ func ObjectPatchLabeler(
 	ctx context.Context,
 	logger *logrus.Logger,
 	check func(metav1.Object) bool,
+	list func(selector labels.Selector) (ret []*metav1.PartialObjectMetadata, err error),
 	patch func(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *apiextensionsv1.CustomResourceDefinition, err error),
-) func(
-	obj interface{},
-) error {
-	return func(obj interface{}) error {
-		cast, ok := obj.(*apiextensionsv1.CustomResourceDefinition)
-		if !ok {
-			err := fmt.Errorf("wrong type %T, expected %T: %#v", obj, new(*apiextensionsv1.CustomResourceDefinition), obj)
-			logger.WithError(err).Error("casting failed")
-			return fmt.Errorf("casting failed: %w", err)
-		}
+) func(done func() bool) queueinformer.LegacySyncHandler {
+	return func(done func() bool) queueinformer.LegacySyncHandler {
+		return func(obj interface{}) error {
+			cast, ok := obj.(*apiextensionsv1.CustomResourceDefinition)
+			if !ok {
+				err := fmt.Errorf("wrong type %T, expected %T: %#v", obj, new(*apiextensionsv1.CustomResourceDefinition), obj)
+				logger.WithError(err).Error("casting failed")
+				return fmt.Errorf("casting failed: %w", err)
+			}
 
-		if !check(cast) || hasLabel(cast) {
-			return nil
-		}
+			if !check(cast) || hasLabel(cast) {
+				// if the object we're processing does not need us to label it, it's possible that every object that requires
+				// the label already has it; in which case we should exit the process, so the Pod that succeeds us can filter
+				// the informers used to drive the controller and stop having to track extraneous objects
+				items, err := list(labels.Everything())
+				if err != nil {
+					logger.WithError(err).Warn("failed to list all objects to check for labelling completion")
+					return nil
+				}
+				gvrFullyLabelled := true
+				for _, item := range items {
+					gvrFullyLabelled = gvrFullyLabelled && (!check(item) || hasLabel(item))
+				}
+				if gvrFullyLabelled {
+					allObjectsLabelled := done()
+					if allObjectsLabelled {
+						logrus.Fatal("detected that every object is labelled, exiting...")
+					}
+				}
+				return nil
+			}
 
-		uid := cast.GetUID()
-		rv := cast.GetResourceVersion()
+			uid := cast.GetUID()
+			rv := cast.GetResourceVersion()
 
-		// to ensure they appear in the patch as preconditions
-		previous := cast.DeepCopy()
-		previous.SetUID("")
-		previous.SetResourceVersion("")
+			// to ensure they appear in the patch as preconditions
+			previous := cast.DeepCopy()
+			previous.SetUID("")
+			previous.SetResourceVersion("")
 
-		oldData, err := json.Marshal(previous)
-		if err != nil {
-			return fmt.Errorf("failed to Marshal old data for %s/%s: %w", previous.GetNamespace(), previous.GetName(), err)
-		}
+			oldData, err := json.Marshal(previous)
+			if err != nil {
+				return fmt.Errorf("failed to Marshal old data for %s/%s: %w", previous.GetNamespace(), previous.GetName(), err)
+			}
 
-		// to ensure they appear in the patch as preconditions
-		updated := cast.DeepCopy()
-		updated.SetUID(uid)
-		updated.SetResourceVersion(rv)
-		labels := updated.GetLabels()
-		if labels == nil {
-			labels = map[string]string{}
-		}
-		labels[install.OLMManagedLabelKey] = install.OLMManagedLabelValue
-		updated.SetLabels(labels)
+			// to ensure they appear in the patch as preconditions
+			updated := cast.DeepCopy()
+			updated.SetUID(uid)
+			updated.SetResourceVersion(rv)
+			labels := updated.GetLabels()
+			if labels == nil {
+				labels = map[string]string{}
+			}
+			labels[install.OLMManagedLabelKey] = install.OLMManagedLabelValue
+			updated.SetLabels(labels)
 
-		newData, err := json.Marshal(updated)
-		if err != nil {
-			return fmt.Errorf("failed to Marshal old data for %s/%s: %w", updated.GetNamespace(), updated.GetName(), err)
-		}
+			newData, err := json.Marshal(updated)
+			if err != nil {
+				return fmt.Errorf("failed to Marshal old data for %s/%s: %w", updated.GetNamespace(), updated.GetName(), err)
+			}
 
-		patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
-		if err != nil {
-			return fmt.Errorf("failed to create patch for %s/%s: %w", cast.GetNamespace(), cast.GetName(), err)
-		}
+			patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
+			if err != nil {
+				return fmt.Errorf("failed to create patch for %s/%s: %w", cast.GetNamespace(), cast.GetName(), err)
+			}
 
-		_, err = patch(ctx, cast.GetName(), types.MergePatchType, patchBytes, metav1.PatchOptions{})
-		return err
+			_, err = patch(ctx, cast.GetName(), types.MergePatchType, patchBytes, metav1.PatchOptions{})
+			return err
+		}
 	}
 }