refactor(packageserver): plumb context throughout

Author: njhale

pkg/package-server/provider/registry.go

@@ -15,14 +15,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/kubernetes/pkg/util/labels"
 
 	operatorsv1alpha1 "github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/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"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
-	"github.com/operator-framework/operator-lifecycle-manager/pkg/metrics"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/package-server/apis/operators"
 )
 
@@ -51,7 +49,7 @@ func newRegistryClient(source *operatorsv1alpha1.CatalogSource, conn *grpc.Clien
 
 // RegistryProvider aggregates several `CatalogSources` and establishes gRPC connections to their registry servers.
 type RegistryProvider struct {
-	*queueinformer.Operator
+	queueinformer.Operator
 
 	mu              sync.RWMutex
 	globalNamespace string
@@ -60,30 +58,32 @@ type RegistryProvider struct {
 
 var _ PackageManifestProvider = &RegistryProvider{}
 
-func NewRegistryProvider(crClient versioned.Interface, operator *queueinformer.Operator, wakeupInterval time.Duration, watchedNamespaces []string, globalNamespace string) *RegistryProvider {
+func NewRegistryProvider(ctx context.Context, crClient versioned.Interface, operator queueinformer.Operator, wakeupInterval time.Duration, watchedNamespaces []string, globalNamespace string) (*RegistryProvider, error) {
 	p := &RegistryProvider{
 		Operator: operator,
 
 		globalNamespace: globalNamespace,
 		clients:         make(map[sourceKey]registryClient),
 	}
 
-	sourceHandlers := &cache.ResourceEventHandlerFuncs{
-		DeleteFunc: p.catalogSourceDeleted,
-	}
 	for _, namespace := range watchedNamespaces {
-		factory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
-		sourceInformer := factory.Operators().V1alpha1().CatalogSources()
+		informerFactory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
+		catsrcInformer := informerFactory.Operators().V1alpha1().CatalogSources()
 
 		// Register queue and QueueInformer
 		logrus.WithField("namespace", namespace).Info("watching catalogsources")
-		queueName := fmt.Sprintf("%s/catalogsources", namespace)
-		sourceQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName)
-		sourceQueueInformer := queueinformer.NewInformer(sourceQueue, sourceInformer.Informer(), p.syncCatalogSource, sourceHandlers, queueName, metrics.NewMetricsNil(), logrus.New())
-		p.RegisterQueueInformer(sourceQueueInformer)
+		catsrcQueueInformer, err := queueinformer.NewQueueInformer(
+			ctx,
+			queueinformer.WithInformer(catsrcInformer.Informer()),
+			queueinformer.WithSyncer(queueinformer.LegacySyncHandler(p.syncCatalogSource).ToSyncerWithDelete(p.catalogSourceDeleted)),
+		)
+		if err != nil {
+			return nil, err
+		}
+		p.RegisterQueueInformer(catsrcQueueInformer)
 	}
 
-	return p
+	return p, nil
 }
 
 func (p *RegistryProvider) getClient(key sourceKey) (registryClient, bool) {

pkg/package-server/provider/registry_test.go

@@ -2,6 +2,7 @@
 package provider
 
 import (
+	"context"
 	"encoding/json"
 	"net"
 	"os"
@@ -77,19 +78,19 @@ func packageManifest(value packageValue) operators.PackageManifest {
 	}
 }
 
-func NewFakeRegistryProvider(clientObjs []runtime.Object, k8sObjs []runtime.Object, watchedNamespaces []string, globalNamespace string, stopCh <-chan struct{}) (*RegistryProvider, error) {
+func NewFakeRegistryProvider(ctx context.Context, clientObjs []runtime.Object, k8sObjs []runtime.Object, watchedNamespaces []string, globalNamespace string) (*RegistryProvider, error) {
 	clientFake := fake.NewSimpleClientset(clientObjs...)
 	k8sClientFake := k8sfake.NewSimpleClientset(k8sObjs...)
 	opClientFake := operatorclient.NewClient(k8sClientFake, nil, nil)
 
-	operator, err := queueinformer.NewOperatorFromClient(opClientFake, logrus.StandardLogger())
+	op, err := queueinformer.NewOperatorFromClient(opClientFake.KubernetesInterface().Discovery(), logrus.StandardLogger())
 	if err != nil {
 		return nil, err
 	}
 
-	wakeupInterval := 5 * time.Minute
+	resyncInterval := 5 * time.Minute
 
-	return NewRegistryProvider(clientFake, operator, wakeupInterval, watchedNamespaces, globalNamespace), nil
+	return NewRegistryProvider(ctx, clientFake, op, resyncInterval, watchedNamespaces, globalNamespace)
 }
 
 func catalogSource(name, namespace string) *operatorsv1alpha1.CatalogSource {
@@ -386,9 +387,9 @@ func TestRegistryProviderGet(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			stopCh := make(chan struct{})
-			defer close(stopCh)
-			provider, err := NewFakeRegistryProvider(nil, nil, test.namespaces, test.globalNS, stopCh)
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+			provider, err := NewFakeRegistryProvider(ctx, nil, nil, test.namespaces, test.globalNS)
 			require.NoError(t, err)
 
 			for _, cs := range test.catalogSources {
@@ -587,9 +588,9 @@ func TestRegistryProviderList(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			stopCh := make(chan struct{})
-			defer close(stopCh)
-			provider, err := NewFakeRegistryProvider(nil, nil, test.namespaces, test.globalNS, stopCh)
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+			provider, err := NewFakeRegistryProvider(ctx, nil, nil, test.namespaces, test.globalNS)
 			require.NoError(t, err)
 
 			for _, cs := range test.catalogSources {

pkg/package-server/server/server.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -25,12 +26,12 @@ import (
 
 // NewCommandStartPackageServer provides a CLI handler for 'start master' command
 // with a default PackageServerOptions.
-func NewCommandStartPackageServer(defaults *PackageServerOptions, stopCh <-chan struct{}) *cobra.Command {
+func NewCommandStartPackageServer(ctx context.Context, defaults *PackageServerOptions) *cobra.Command {
 	cmd := &cobra.Command{
 		Short: "Launch a package API server",
 		Long:  "Launch a package API server",
 		RunE: func(c *cobra.Command, args []string) error {
-			if err := defaults.Run(stopCh); err != nil {
+			if err := defaults.Run(ctx); err != nil {
 				return err
 			}
 			return nil
@@ -128,7 +129,7 @@ func (o *PackageServerOptions) Config() (*apiserver.Config, error) {
 	}, nil
 }
 
-func (o *PackageServerOptions) Run(stopCh <-chan struct{}) error {
+func (o *PackageServerOptions) Run(ctx context.Context) error {
 	if o.Debug {
 		log.SetLevel(log.DebugLevel)
 	}
@@ -164,12 +165,15 @@ func (o *PackageServerOptions) Run(stopCh <-chan struct{}) error {
 		return err
 	}
 
-	queueOperator, err := queueinformer.NewOperator(o.Kubeconfig, log.New())
+	queueOperator, err := queueinformer.NewOperatorFromClient(crClient.Discovery(), log.New())
 	if err != nil {
 		return err
 	}
 
-	sourceProvider := provider.NewRegistryProvider(crClient, queueOperator, o.WakeupInterval, o.WatchedNamespaces, o.GlobalNamespace)
+	sourceProvider, err := provider.NewRegistryProvider(ctx, crClient, queueOperator, o.WakeupInterval, o.WatchedNamespaces, o.GlobalNamespace)
+	if err != nil {
+		return err
+	}
 	config.ProviderConfig.Provider = sourceProvider
 
 	// we should never need to resync, since we're not worried about missing events,
@@ -182,15 +186,11 @@ func (o *PackageServerOptions) Run(stopCh <-chan struct{}) error {
 		return err
 	}
 
-	// Ensure that provider stops after the apiserver gracefully shuts down
-	provCh := make(chan struct{})
-	ready, done, _ := sourceProvider.Run(provCh)
-	<-ready
-
-	err = server.GenericAPIServer.PrepareRun().Run(stopCh)
-	go func() { provCh <- struct{}{} }()
+	sourceProvider.Run(ctx)
+	<-sourceProvider.Ready()
 
-	<-done
+	err = server.GenericAPIServer.PrepareRun().Run(ctx.Done())
+	<-sourceProvider.Done()
 
 	return err
 }