Move server side apply patch code to library, use ssa to update catalog source pods

Author: Ankita Thomas

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())
 	}

go.mod

@@ -18,11 +18,13 @@ require (
 	github.com/otiai10/copy v1.0.1
 	github.com/pkg/errors v0.8.1
 	github.com/prometheus/client_golang v1.2.1
+	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/sirupsen/logrus v1.4.2
 	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.4.0
 	golang.org/x/sys v0.0.0-20191210023423-ac6580df4449 // indirect
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/grpc v1.24.0
 	gopkg.in/yaml.v2 v2.2.4
 	helm.sh/helm/v3 v3.0.1
@@ -37,7 +39,10 @@ require (
 	k8s.io/kube-aggregator v0.0.0
 	k8s.io/kube-openapi v0.0.0-20190918143330-0270cf2f1c1d
 	k8s.io/kubernetes v1.16.0
+	sigs.k8s.io/controller-runtime v0.5.1 // indirect
 	sigs.k8s.io/controller-tools v0.2.4
+	sigs.k8s.io/testing_frameworks v0.1.2 // indirect
+	sigs.k8s.io/yaml v1.1.0 // indirect
 )
 
 replace (

go.sum

@@ -11,6 +11,8 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime"
+
 	errorwrap "github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/connectivity"
@@ -47,6 +49,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"
 	index "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/index"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
@@ -70,6 +73,7 @@ const (
 	generatedByKey         = "olm.generated-by"
 	maxInstallPlanCount    = 5
 	maxDeletesPerSweep     = 5
+	RegistryFieldManager   = "olm.registry"
 )
 
 // Operator represents a Kubernetes operator that executes InstallPlans by
@@ -102,7 +106,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 {
@@ -131,6 +135,10 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	// Create an OperatorLister
 	lister := operatorlister.NewLister()
 
+	ssaClient, err := controllerclient.NewForConfig(config, scheme, RegistryFieldManager)
+	if err != nil {
+		return nil, err
+	}
 	// Allocate the new instance of an Operator.
 	op := &Operator{
 		Operator:                 queueOperator,
@@ -151,7 +159,7 @@ 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)
 
 	// Wire OLM CR sharedIndexInformers
 	crInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, resyncPeriod())

pkg/controller/operators/catalog/operator.go

@@ -49,6 +49,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"
@@ -1301,7 +1302,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/operators/catalog/operator_test.go

@@ -1,9 +1,11 @@
 package reconciler
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/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"
@@ -69,9 +71,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{}
@@ -193,16 +196,16 @@ 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(updatePod)
+			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/grpc.go

@@ -3,6 +3,7 @@ package reconciler
 
 import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/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/controller/registry/reconciler/reconciler.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
+}