feat(registry): when syncing catalogsources, generate pods that serve

the operator registry api for configmaps

cmd/catalog/main.go

@@ -69,7 +69,7 @@ func main() {
 	go http.ListenAndServe(":8080", nil)
 
 	// Create a new instance of the operator.
-	catalogOperator, err := catalog.NewOperator(*kubeConfigPath, *wakeupInterval, *catalogNamespace, strings.Split(*watchedNamespaces, ",")...)
+	catalogOperator, err := catalog.NewOperator(*kubeConfigPath, *wakeupInterval, *configmapServerImage, *catalogNamespace, strings.Split(*watchedNamespaces, ",")...)
 	if err != nil {
 		log.Panicf("error configuring operator: %s", err.Error())
 	}

deploy/chart/templates/0000_30_05-catalogsource.crd.yaml

@@ -43,10 +43,6 @@ spec:
   validation:
     openAPIV3Schema:
       properties:
-        type: object
-        description: Represents a catalog of operator data. OLM will query catalog sources to resolve dependencies.
-        required:
-        - spec
         spec:
           type: object
           description: Spec for a catalog source.

pkg/controller/operators/catalog/operator.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"k8s.io/client-go/informers"
 	"sync"
 	"time"
 
@@ -52,10 +53,11 @@ type Operator struct {
 	sourcesLastUpdate  metav1.Time
 	dependencyResolver resolver.DependencyResolver
 	subQueue           workqueue.RateLimitingInterface
+	configmapRegistryReconciler *registry.ConfigMapRegistryReconciler
 }
 
 // NewOperator creates a new Catalog Operator.
-func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNamespace string, watchedNamespaces ...string) (*Operator, error) {
+func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, configmapRegistryImage string, operatorNamespace string, watchedNamespaces ...string) (*Operator, error) {
 	// Default to watching all namespaces.
 	if watchedNamespaces == nil {
 		watchedNamespaces = []string{metav1.NamespaceAll}
@@ -141,57 +143,135 @@ func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNa
 		op.RegisterQueueInformer(informer)
 	}
 
+	// Creates registry pods in response to configmaps
+	informerFactory := informers.NewSharedInformerFactory(op.OpClient.KubernetesInterface(), wakeupInterval)
+	roleInformer := informerFactory.Rbac().V1().Roles()
+	roleBindingInformer := informerFactory.Rbac().V1().RoleBindings()
+	serviceAccountInformer := informerFactory.Core().V1().ServiceAccounts()
+	serviceInformer := informerFactory.Core().V1().Services()
+	podInformer := informerFactory.Core().V1().Pods()
+	configMapInformer := informerFactory.Core().V1().ConfigMaps()
+	op.configmapRegistryReconciler = &registry.ConfigMapRegistryReconciler{
+		Image: configmapRegistryImage,
+		OpClient: op.OpClient,
+		RoleLister: roleInformer.Lister(),
+		RoleBindingLister: roleBindingInformer.Lister(),
+		ServiceAccountLister: serviceAccountInformer.Lister(),
+		ServiceLister: serviceInformer.Lister(),
+		PodLister: podInformer.Lister(),
+		ConfigMapLister: configMapInformer.Lister(),
+	}
+
+	// register informers for configmapRegistryReconciler
+	registryInformers := []cache.SharedIndexInformer{
+		roleInformer.Informer(),
+		roleBindingInformer.Informer(),
+		serviceAccountInformer.Informer(),
+		serviceInformer.Informer(),
+		podInformer.Informer(),
+		configMapInformer.Informer(),
+	}
+
+	// TODO: won't this possibly conflict since GVK isn't part of the queue entry?
+	registryQueueInformers := queueinformer.New(
+		workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "registry"),
+		registryInformers,
+		op.syncRegistry,
+		nil,
+		"registry",
+		metrics.NewMetricsNil(),
+	)
+	for _, informer := range registryQueueInformers {
+		op.RegisterQueueInformer(informer)
+	}
 	return op, nil
 }
 
+func (o *Operator) syncRegistry(obj interface{}) (syncError error) {
+	switch obj.(type) {
+	case *corev1.ConfigMap:
+		// requeue catalogsource
+	}
+	return nil
+}
+
 func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 	catsrc, ok := obj.(*v1alpha1.CatalogSource)
 	if !ok {
 		log.Debugf("wrong type: %#v", obj)
 		return fmt.Errorf("casting CatalogSource failed")
 	}
 
-	// Get the catalog source's config map
-	configMap, err := o.OpClient.KubernetesInterface().CoreV1().ConfigMaps(catsrc.GetNamespace()).Get(catsrc.Spec.ConfigMap, metav1.GetOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to get catalog config map %s when updating status: %s", catsrc.Spec.ConfigMap, err)
+	logger := log.WithFields(log.Fields{
+		"source":      catsrc.GetName(),
+	})
+
+	if catsrc.Spec.SourceType == v1alpha1.SourceTypeInternal || catsrc.Spec.SourceType == v1alpha1.SourceTypeConfigmap {
+		return o.syncConfigMapSource(logger, catsrc)
 	}
 
-	o.sourcesLock.Lock()
-	defer o.sourcesLock.Unlock()
-	sourceKey := registry.ResourceKey{Name: catsrc.GetName(), Namespace: catsrc.GetNamespace()}
-	_, ok = o.sources[sourceKey]
+	logger.WithField("sourceType", catsrc.Spec.SourceType).Warn("unknown source type")
 
-	// Check for catalog source changes
-	if ok && catsrc.Status.ConfigMapResource != nil && catsrc.Status.ConfigMapResource.Name == configMap.GetName() && catsrc.Status.ConfigMapResource.ResourceVersion == configMap.GetResourceVersion() {
-		return nil
-	}
+	// TODO: write status about invalid source type
 
-	// Update status subresource
-	out := catsrc.DeepCopy()
-	out.Status.ConfigMapResource = &v1alpha1.ConfigMapResourceReference{
-		Name:            configMap.GetName(),
-		Namespace:       configMap.GetNamespace(),
-		UID:             configMap.GetUID(),
-		ResourceVersion: configMap.GetResourceVersion(),
-	}
-	out.Status.LastSync = timeNow()
+	return nil
+}
 
-	_, err = o.client.OperatorsV1alpha1().CatalogSources(out.GetNamespace()).UpdateStatus(out)
+func (o *Operator) syncConfigMapSource(logger *log.Entry, catsrc *v1alpha1.CatalogSource) (syncError error) {
+
+	// Get the catalog source's config map
+	configMap, err := o.configmapRegistryReconciler.ConfigMapLister.ConfigMaps(catsrc.GetNamespace()).Get(catsrc.Spec.ConfigMap)
 	if err != nil {
-		return fmt.Errorf("failed to update catalog source %s status: %s", out.GetName(), err)
+		return fmt.Errorf("failed to get catalog config map %s: %s", catsrc.Spec.ConfigMap, err)
 	}
 
-	// Create a new in-mem registry
-	src, err := registry.NewInMemoryFromConfigMap(o.OpClient, out.GetNamespace(), out.Spec.ConfigMap)
-	if err != nil {
-		return fmt.Errorf("failed to create catalog source from ConfigMap %s: %s", out.Spec.ConfigMap, err)
+	if catsrc.Status.ConfigMapResource == nil || catsrc.Status.ConfigMapResource.UID == configMap.GetUID() || catsrc.Status.ConfigMapResource.ResourceVersion != configMap.ResourceVersion {
+		// configmap ref nonexistant or updated, write out the new configmap ref to status and exit
+		out := catsrc.DeepCopy()
+		out.Status.ConfigMapResource = &v1alpha1.ConfigMapResourceReference{
+			Name:            configMap.GetName(),
+			Namespace:       configMap.GetNamespace(),
+			UID:             configMap.GetUID(),
+			ResourceVersion: configMap.GetResourceVersion(),
+		}
+		out.Status.LastSync = timeNow()
+
+		// update source map
+		o.sourcesLock.Lock()
+		defer o.sourcesLock.Unlock()
+		sourceKey := registry.ResourceKey{Name: catsrc.GetName(), Namespace: catsrc.GetNamespace()}
+		src, err := registry.NewInMemoryFromConfigMap(o.OpClient, out.GetNamespace(), out.Spec.ConfigMap)
+		o.sources[sourceKey] = src
+		if err != nil {
+			return err
+		}
+
+		// update status
+		if _, err = o.client.OperatorsV1alpha1().CatalogSources(out.GetNamespace()).UpdateStatus(out); err!= nil {
+			return err
+		}
+		o.sourcesLastUpdate = timeNow()
+		return nil
 	}
 
-	// Update sources map
-	o.sources[sourceKey] = src
-	o.sourcesLastUpdate = timeNow()
+	// configmap ref is up to date, continue parsing
+	if catsrc.Status.RegistryServiceStatus == nil || catsrc.Status.RegistryServiceStatus.CreatedAt.Before(&catsrc.Status.LastSync) {
+		// if registry pod hasn't been created or hasn't been updated since the last configmap update, recreate it
+
+		out := catsrc.DeepCopy()
+		if err := o.configmapRegistryReconciler.EnsureRegistryServer(out); err != nil {
+			logger.WithError(err).Warn("couldn't ensure registry server")
+			return err
+		}
+		// update status
+		if _, err = o.client.OperatorsV1alpha1().CatalogSources(out.GetNamespace()).UpdateStatus(out); err!= nil {
+			return err
+		}
+		o.sourcesLastUpdate = timeNow()
+		return nil
+	}
 
+	// no updates
 	return nil
 }
 

pkg/controller/operators/catalog/operator_test.go

@@ -2,7 +2,11 @@ package catalog
 
 import (
 	"errors"
+	"fmt"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 	"testing"
+	"time"
 
 	"github.com/ghodss/yaml"
 
@@ -139,6 +143,32 @@ func TestSyncCatalogSources(t *testing.T) {
 		expectedStatus    *v1alpha1.CatalogSourceStatus
 		expectedError     error
 	}{
+		{
+			testName:          "CatalogSourceWithInvalidSourceType",
+			operatorNamespace: "cool-namespace",
+			catalogSource: &v1alpha1.CatalogSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "cool-catalog",
+					Namespace: "cool-namespace",
+					UID:       types.UID("catalog-uid"),
+				},
+				Spec: v1alpha1.CatalogSourceSpec{
+					ConfigMap: "cool-configmap",
+					SourceType: "nope",
+				},
+			},
+			configMap: &corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:            "cool-configmap",
+					Namespace:       "cool-namespace",
+					UID:             types.UID("configmap-uid"),
+					ResourceVersion: "resource-version",
+				},
+				Data: fakeConfigMapData(),
+			},
+			expectedStatus: nil,
+			expectedError: nil,
+		},
 		{
 			testName:          "CatalogSourceWithBackingConfigMap",
 			operatorNamespace: "cool-namespace",
@@ -150,6 +180,7 @@ func TestSyncCatalogSources(t *testing.T) {
 				},
 				Spec: v1alpha1.CatalogSourceSpec{
 					ConfigMap: "cool-configmap",
+					SourceType: v1alpha1.SourceTypeInternal,
 				},
 			},
 			configMap: &corev1.ConfigMap{
@@ -182,6 +213,7 @@ func TestSyncCatalogSources(t *testing.T) {
 				},
 				Spec: v1alpha1.CatalogSourceSpec{
 					ConfigMap: "cool-configmap",
+					SourceType: v1alpha1.SourceTypeConfigmap,
 				},
 				Status: v1alpha1.CatalogSourceStatus{
 					ConfigMapResource: &v1alpha1.ConfigMapResourceReference{
@@ -222,6 +254,7 @@ func TestSyncCatalogSources(t *testing.T) {
 				},
 				Spec: v1alpha1.CatalogSourceSpec{
 					ConfigMap: "cool-configmap",
+					SourceType: v1alpha1.SourceTypeConfigmap,
 				},
 			},
 			configMap: &corev1.ConfigMap{
@@ -234,7 +267,7 @@ func TestSyncCatalogSources(t *testing.T) {
 				Data: map[string]string{},
 			},
 			expectedStatus: nil,
-			expectedError:  errors.New("failed to create catalog source from ConfigMap cool-configmap: error parsing ConfigMap cool-configmap: no valid resources found"),
+			expectedError:  errors.New("error parsing ConfigMap cool-configmap: no valid resources found"),
 		},
 		{
 			testName:          "CatalogSourceWithMissingConfigMap",
@@ -247,21 +280,25 @@ func TestSyncCatalogSources(t *testing.T) {
 				},
 				Spec: v1alpha1.CatalogSourceSpec{
 					ConfigMap: "cool-configmap",
+					SourceType: v1alpha1.SourceTypeConfigmap,
 				},
 			},
 			configMap:      &corev1.ConfigMap{},
 			expectedStatus: nil,
-			expectedError:  errors.New("failed to get catalog config map cool-configmap when updating status: configmaps \"cool-configmap\" not found"),
+			expectedError:  errors.New("failed to get catalog config map cool-configmap: configmap \"cool-configmap\" not found"),
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.testName, func(t *testing.T) {
+			stopc := make(chan struct{})
+			defer close(stopc)
+
 			// Create existing objects
 			clientObjs := []runtime.Object{tt.catalogSource}
 			k8sObjs := []runtime.Object{tt.configMap}
 
 			// Create test operator
-			op, err := NewFakeOperator(clientObjs, k8sObjs, nil, nil, resolver, tt.operatorNamespace)
+			op, err := NewFakeOperator(clientObjs, k8sObjs, nil, nil, resolver, tt.operatorNamespace, stopc)
 			require.NoError(t, err)
 
 			// Run sync
@@ -373,8 +410,8 @@ func fakeConfigMapData() map[string]string {
 	return data
 }
 
-// NewFakeOprator creates a new operator using fake clients
-func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extObjs []runtime.Object, regObjs []runtime.Object, resolver resolver.DependencyResolver, namespace string) (*Operator, error) {
+// NewFakeOperator creates a new operator using fake clients
+func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extObjs []runtime.Object, regObjs []runtime.Object, resolver resolver.DependencyResolver, namespace string, stopc <- chan struct{}) (*Operator, error) {
 	// Create client fakes
 	clientFake := fake.NewSimpleClientset(clientObjs...)
 	opClientFake := operatorclient.NewClient(k8sfake.NewSimpleClientset(k8sObjs...), apiextensionsfake.NewSimpleClientset(extObjs...), apiregistrationfake.NewSimpleClientset(regObjs...))
@@ -385,6 +422,15 @@ func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extO
 		return nil, err
 	}
 
+	// Creates registry pods in response to configmaps
+	informerFactory := informers.NewSharedInformerFactory(opClientFake.KubernetesInterface(), 5*time.Second)
+	roleInformer := informerFactory.Rbac().V1().Roles()
+	roleBindingInformer := informerFactory.Rbac().V1().RoleBindings()
+	serviceAccountInformer := informerFactory.Core().V1().ServiceAccounts()
+	serviceInformer := informerFactory.Core().V1().Services()
+	podInformer := informerFactory.Core().V1().Pods()
+	configMapInformer := informerFactory.Core().V1().ConfigMaps()
+
 	// Create the new operator
 	queueOperator, err := queueinformer.NewOperatorFromClient(opClientFake)
 	op := &Operator{
@@ -395,6 +441,38 @@ func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extO
 		dependencyResolver: resolver,
 	}
 
+	op.configmapRegistryReconciler = &registry.ConfigMapRegistryReconciler{
+		Image: "test:pod",
+		OpClient: op.OpClient,
+		RoleLister: roleInformer.Lister(),
+		RoleBindingLister: roleBindingInformer.Lister(),
+		ServiceAccountLister: serviceAccountInformer.Lister(),
+		ServiceLister: serviceInformer.Lister(),
+		PodLister: podInformer.Lister(),
+		ConfigMapLister: configMapInformer.Lister(),
+	}
+
+	// register informers for configmapRegistryReconciler
+	registryInformers := []cache.SharedIndexInformer{
+		roleInformer.Informer(),
+		roleBindingInformer.Informer(),
+		serviceAccountInformer.Informer(),
+		serviceInformer.Informer(),
+		podInformer.Informer(),
+		configMapInformer.Informer(),
+	}
+
+	var hasSyncedCheckFns []cache.InformerSynced
+	for _, informer := range registryInformers {
+		op.RegisterInformer(informer)
+		hasSyncedCheckFns = append(hasSyncedCheckFns, informer.HasSynced)
+		go informer.Run(stopc)
+	}
+
+	if ok := cache.WaitForCacheSync(stopc, hasSyncedCheckFns...); !ok {
+		return nil, fmt.Errorf("failed to wait for caches to sync")
+	}
+
 	return op, nil
 }
 

pkg/controller/operators/olm/operator.go

@@ -56,8 +56,6 @@ type Operator struct {
 	client   versioned.Interface
 	resolver install.StrategyResolverInterface
 	lister   operatorlister.OperatorLister
-	// csvLister           map[string]csvlister.ClusterServiceVersionLister
-	// operatorGroupLister map[string]operatorgrouplister.OperatorGroupLister
 	annotator   *annotator.Annotator
 	recorder    record.EventRecorder
 	cleanupFunc func()