Merge pull request #1624 from ankitathomas/registry_poll_interval_retry

Bug 1853601: Retry conflicting catalog source pod update after fetching latest pod spec

Author: openshift-merge-robot

cmd/catalog/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 	utilclock "k8s.io/apimachinery/pkg/util/clock"
+	k8sscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client"
@@ -172,7 +173,7 @@ func main() {
 	}
 
 	// Create a new instance of the operator.
-	op, err := catalog.NewOperator(ctx, *kubeConfigPath, utilclock.RealClock{}, logger, *wakeupInterval, *configmapServerImage, *utilImage, *catalogNamespace)
+	op, err := catalog.NewOperator(ctx, *kubeConfigPath, utilclock.RealClock{}, logger, *wakeupInterval, *configmapServerImage, *utilImage, *catalogNamespace, k8sscheme.Scheme)
 	if err != nil {
 		log.Panicf("error configuring operator: %s", err.Error())
 	}

pkg/controller/operators/catalog/operator.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"k8s.io/apimachinery/pkg/runtime"
 	"reflect"
 	"sort"
 	"strings"
@@ -49,6 +50,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/grpc"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/reconciler"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
+	controllerclient "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/controller-runtime/client"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/event"
 	index "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/index"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
@@ -74,6 +76,7 @@ const (
 	generatedByKey         = "olm.generated-by"
 	maxInstallPlanCount    = 5
 	maxDeletesPerSweep     = 5
+	RegistryFieldManager   = "olm.registry"
 )
 
 // Operator represents a Kubernetes operator that executes InstallPlans by
@@ -107,7 +110,7 @@ type Operator struct {
 type CatalogSourceSyncFunc func(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, syncError error)
 
 // NewOperator creates a new Catalog Operator.
-func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resync time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string) (*Operator, error) {
+func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resync time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string, scheme *runtime.Scheme) (*Operator, error) {
 	resyncPeriod := queueinformer.ResyncWithJitter(resync, 0.2)
 	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
 	if err != nil {
@@ -142,6 +145,11 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		return nil, err
 	}
 
+	ssaClient, err := controllerclient.NewForConfig(config, scheme, RegistryFieldManager)
+	if err != nil {
+		return nil, err
+	}
+
 	// Allocate the new instance of an Operator.
 	op := &Operator{
 		Operator:                 queueOperator,
@@ -162,9 +170,8 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		clientAttenuator:         scoped.NewClientAttenuator(logger, config, opClient, crClient, dynamicClient),
 	}
 	op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
-	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now)
+	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now, ssaClient)
 	res := resolver.NewOperatorStepResolver(lister, crClient, opClient.KubernetesInterface(), operatorNamespace, op.sources, logger)
-	op.resolver = resolver.NewOperatorStepResolver(lister, crClient, opClient.KubernetesInterface(), operatorNamespace, op.sources, logger)
 	op.resolver = resolver.NewInstrumentedResolver(res, metrics.RegisterDependencyResolutionSuccess, metrics.RegisterDependencyResolutionFailure)
 
 	// Wire OLM CR sharedIndexInformers

pkg/controller/operators/catalog/operator_test.go

@@ -50,6 +50,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/fakes"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/clientfake"
+	controllerclient "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/controller-runtime/client"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
@@ -1247,7 +1248,14 @@ func NewFakeOperator(ctx context.Context, namespace string, namespaces []string,
 	}
 	op.sources = grpc.NewSourceStore(config.logger, 1*time.Second, 5*time.Second, op.syncSourceState)
 	if op.reconciler == nil {
-		op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, op.opClient, "test:pod", op.now)
+		s := runtime.NewScheme()
+		err := k8sfake.AddToScheme(s)
+		if err != nil {
+			return nil, err
+		}
+		applier := controllerclient.NewFakeApplier(s, "testowner")
+
+		op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, op.opClient, "test:pod", op.now, applier)
 	}
 
 	op.RunInformers(ctx)

pkg/controller/registry/reconciler/grpc.go

@@ -3,8 +3,8 @@ package reconciler
 import (
 	"context"
 	"fmt"
-
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	controllerclient "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/controller-runtime/client"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
@@ -70,9 +70,10 @@ func (s *grpcCatalogSourceDecorator) Pod() *v1.Pod {
 }
 
 type GrpcRegistryReconciler struct {
-	now      nowFunc
-	Lister   operatorlister.OperatorLister
-	OpClient operatorclient.ClientInterface
+	now       nowFunc
+	Lister    operatorlister.OperatorLister
+	OpClient  operatorclient.ClientInterface
+	SSAClient *controllerclient.ServerSideApplier
 }
 
 var _ RegistryReconciler = &GrpcRegistryReconciler{}
@@ -194,16 +195,17 @@ func (c *GrpcRegistryReconciler) ensureUpdatePod(source grpcCatalogSourceDecorat
 			logrus.WithField("CatalogSource", source.GetName()).Info("detect image update for catalogsource pod")
 
 			updateFlag = true
-			updatePod.Labels[CatalogSourceLabelKey] = source.GetName()
-			updatePod.Labels[CatalogSourceUpdateKey] = ""
 
 			// Update the update pod to promote it to serving pod
-			_, err := c.OpClient.KubernetesInterface().CoreV1().Pods(source.GetNamespace()).Update(context.TODO(), updatePod, metav1.UpdateOptions{})
+			err := c.SSAClient.Apply(context.TODO(), updatePod, func(p *v1.Pod) error {
+				p.Labels[CatalogSourceLabelKey] = source.GetName()
+				p.Labels[CatalogSourceUpdateKey] = ""
+				return nil
+			})()
+
 			if err != nil {
-				return errors.Wrapf(err, "error creating new pod: %s", source.Pod().GetName())
+				return errors.Wrapf(err, "error updating catalog source pod: %s", source.Pod().GetName())
 			}
-
-			break
 		}
 	}
 

pkg/controller/registry/reconciler/reconciler.go

@@ -3,6 +3,7 @@ package reconciler
 
 import (
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	controllerclient "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/controller-runtime/client"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	v1 "k8s.io/api/core/v1"
@@ -46,6 +47,7 @@ type registryReconcilerFactory struct {
 	Lister               operatorlister.OperatorLister
 	OpClient             operatorclient.ClientInterface
 	ConfigMapServerImage string
+	SSAClient            *controllerclient.ServerSideApplier
 }
 
 // ReconcilerForSource returns a RegistryReconciler based on the configuration of the given CatalogSource.
@@ -62,9 +64,10 @@ func (r *registryReconcilerFactory) ReconcilerForSource(source *v1alpha1.Catalog
 	case v1alpha1.SourceTypeGrpc:
 		if source.Spec.Image != "" {
 			return &GrpcRegistryReconciler{
-				now:      r.now,
-				Lister:   r.Lister,
-				OpClient: r.OpClient,
+				now:       r.now,
+				Lister:    r.Lister,
+				OpClient:  r.OpClient,
+				SSAClient: r.SSAClient,
 			}
 		} else if source.Spec.Address != "" {
 			return &GrpcAddressRegistryReconciler{
@@ -76,12 +79,13 @@ func (r *registryReconcilerFactory) ReconcilerForSource(source *v1alpha1.Catalog
 }
 
 // NewRegistryReconcilerFactory returns an initialized RegistryReconcilerFactory.
-func NewRegistryReconcilerFactory(lister operatorlister.OperatorLister, opClient operatorclient.ClientInterface, configMapServerImage string, now nowFunc) RegistryReconcilerFactory {
+func NewRegistryReconcilerFactory(lister operatorlister.OperatorLister, opClient operatorclient.ClientInterface, configMapServerImage string, now nowFunc, ssaClient *controllerclient.ServerSideApplier) RegistryReconcilerFactory {
 	return &registryReconcilerFactory{
 		now:                  now,
 		Lister:               lister,
 		OpClient:             opClient,
 		ConfigMapServerImage: configMapServerImage,
+		SSAClient:            ssaClient,
 	}
 }
 

pkg/lib/controller-runtime/client/fake_ssa.go

@@ -0,0 +1,56 @@
+package client
+
+import (
+	"context"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	k8scontrollerclient "sigs.k8s.io/controller-runtime/pkg/client"
+	fakecontrollerclient "sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+// FakeApplier provides a wrapper around the fake k8s controller client to convert the unsupported apply-type patches to merge patches.
+func NewFakeApplier(scheme *runtime.Scheme, owner string, objs ...runtime.Object) *ServerSideApplier {
+	return &ServerSideApplier{
+		client: &fakeApplier{fakecontrollerclient.NewFakeClientWithScheme(scheme, objs...)},
+		Scheme: scheme,
+		Owner:  k8scontrollerclient.FieldOwner(owner),
+	}
+}
+
+type fakeApplier struct {
+	k8scontrollerclient.Client
+}
+
+func (c *fakeApplier) Patch(ctx context.Context, obj runtime.Object, patch k8scontrollerclient.Patch, opts ...k8scontrollerclient.PatchOption) error {
+	patch, opts = convertApplyToMergePatch(patch, opts...)
+	return c.Client.Patch(ctx, obj, patch, opts...)
+}
+
+func (c *fakeApplier) Status() k8scontrollerclient.StatusWriter {
+	return fakeStatusWriter{c.Client.Status()}
+}
+
+type fakeStatusWriter struct {
+	k8scontrollerclient.StatusWriter
+}
+
+func (c fakeStatusWriter) Patch(ctx context.Context, obj runtime.Object, patch k8scontrollerclient.Patch, opts ...k8scontrollerclient.PatchOption) error {
+	patch, opts = convertApplyToMergePatch(patch, opts...)
+	return c.StatusWriter.Patch(ctx, obj, patch, opts...)
+}
+
+func convertApplyToMergePatch(patch k8scontrollerclient.Patch, opts ...k8scontrollerclient.PatchOption) (k8scontrollerclient.Patch, []k8scontrollerclient.PatchOption) {
+	// Apply patch type is not supported on the fake controller
+	if patch.Type() == types.ApplyPatchType {
+		patch = k8scontrollerclient.Merge
+		patchOptions := make([]k8scontrollerclient.PatchOption, 0, len(opts))
+		for _, opt := range opts {
+			if opt == k8scontrollerclient.ForceOwnership {
+				continue
+			}
+			patchOptions = append(patchOptions, opt)
+		}
+		opts = patchOptions
+	}
+	return patch, opts
+}