add catalog: support bundle injection

Author: joelanford

internal/pkg/action/catalog_add.go

@@ -2,26 +2,38 @@ package action
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-registry/pkg/image"
 	"github.com/operator-framework/operator-registry/pkg/image/containerdregistry"
 	"github.com/spf13/pflag"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"github.com/joelanford/kubectl-operator/internal/pkg/catalog"
 	"github.com/joelanford/kubectl-operator/internal/pkg/log"
 )
 
+const grpcPort = "50051"
+
 type CatalogAdd struct {
 	config *Configuration
 
 	CatalogSourceName string
 	IndexImage        string
+	InjectBundles     []string
+	InjectBundleMode  string
 	DisplayName       string
 	Publisher         string
 	AddTimeout        time.Duration
@@ -43,6 +55,10 @@ func (a *CatalogAdd) BindFlags(fs *pflag.FlagSet) {
 	fs.StringVarP(&a.Publisher, "publisher", "p", "", "publisher of the index")
 	fs.DurationVarP(&a.AddTimeout, "timeout", "t", time.Minute, "the amount of time to wait before cancelling the catalog addition")
 	fs.DurationVar(&a.CleanupTimeout, "cleanup-timeout", time.Minute, "the amount to time to wait before cancelling cleanup")
+
+	fs.StringArrayVarP(&a.InjectBundles, "inject-bundles", "b", nil, "inject extra bundles into the index at runtime")
+	fs.StringVarP(&a.InjectBundleMode, "inject-bundle-mode", "m", "", "mode to use to inject bundles")
+	_ = fs.MarkHidden("inject-bundle-mode")
 }
 
 func (a *CatalogAdd) Run(ctx context.Context) (*v1alpha1.CatalogSource, error) {
@@ -70,13 +86,37 @@ func (a *CatalogAdd) Run(ctx context.Context) (*v1alpha1.CatalogSource, error) {
 
 	a.setDefaults(labels)
 
+	var registryPod *corev1.Pod
+	if len(a.InjectBundles) > 0 {
+		if registryPod, err = a.createRegistryPod(ctx); err != nil {
+			return nil, err
+		}
+	}
+
 	opts := []catalog.Option{
-		catalog.Image(a.IndexImage),
 		catalog.DisplayName(a.DisplayName),
 		catalog.Publisher(a.Publisher),
 	}
+
+	if registryPod == nil {
+		opts = append(opts, catalog.Image(a.IndexImage))
+	} else {
+		address := fmt.Sprintf("%s:%s", registryPod.Status.PodIP, grpcPort)
+		injectedBundlesJSON, err := json.Marshal(a.InjectBundles)
+		if err != nil {
+			return nil, fmt.Errorf("json marshal injected bundles: %v", err)
+		}
+		annotations := map[string]string{
+			"operators.operatorframework.io/injected-bundles": string(injectedBundlesJSON),
+		}
+		opts = append(opts,
+			catalog.Address(address),
+			catalog.Annotations(annotations),
+		)
+	}
+
 	cs := catalog.Build(csKey, opts...)
-	if err := a.add(ctx, cs); err != nil {
+	if err := a.add(ctx, cs, registryPod); err != nil {
 		defer a.cleanup(cs)
 		return nil, err
 	}
@@ -106,13 +146,85 @@ func (a *CatalogAdd) setDefaults(labels map[string]string) {
 			a.Publisher = v
 		}
 	}
+	if a.InjectBundleMode == "" {
+		if strings.HasPrefix(a.IndexImage, "quay.io/operator-framework/upstream-opm-builder") {
+			a.InjectBundleMode = "semver"
+		} else {
+			a.InjectBundleMode = "replaces"
+		}
+	}
+}
+
+func (a *CatalogAdd) createRegistryPod(ctx context.Context) (*corev1.Pod, error) {
+	command := []string{
+		"/bin/sh",
+		"-c",
+		fmt.Sprintf(`mkdir -p /database && \
+/bin/opm registry add   -d /database/index.db --mode=%s -b %s && \
+/bin/opm registry serve -d /database/index.db -p %s`, a.InjectBundleMode, strings.Join(a.InjectBundles, ","), grpcPort),
+	}
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-%s", a.CatalogSourceName, rand.String(4)),
+			Namespace: a.config.Namespace,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:    "registry",
+					Image:   a.IndexImage,
+					Command: command,
+				},
+			},
+		},
+	}
+	if err := a.config.Client.Create(ctx, pod); err != nil {
+		return nil, err
+	}
+
+	if err := wait.PollImmediateUntil(time.Millisecond*250, func() (bool, error) {
+		podKey, err := client.ObjectKeyFromObject(pod)
+		if err != nil {
+			return false, fmt.Errorf("get pod key: %v", err)
+		}
+		if err := a.config.Client.Get(ctx, podKey, pod); err != nil {
+			return false, err
+		}
+		if pod.Status.Phase == corev1.PodRunning && pod.Status.PodIP != "" {
+			return true, nil
+		}
+		return false, nil
+	}, ctx.Done()); err != nil {
+		return nil, fmt.Errorf("registry pod not ready: %v", err)
+	}
+	return pod, nil
 }
 
-func (a *CatalogAdd) add(ctx context.Context, cs *v1alpha1.CatalogSource) error {
+func (a *CatalogAdd) add(ctx context.Context, cs *v1alpha1.CatalogSource, pod *corev1.Pod) error {
 	if err := a.config.Client.Create(ctx, cs); err != nil {
 		return fmt.Errorf("create catalogsource: %v", err)
 	}
 
+	if pod != nil {
+		retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+			podKey, err := client.ObjectKeyFromObject(pod)
+			if err != nil {
+				return fmt.Errorf("get pod key: %v", err)
+			}
+			if err := a.config.Client.Get(ctx, podKey, pod); err != nil {
+				return fmt.Errorf("get registry pod: %v", err)
+			}
+			if err := controllerutil.SetOwnerReference(cs, pod, a.config.Scheme); err != nil {
+				return fmt.Errorf("set registry pod owner reference: %v", err)
+			}
+			if err := a.config.Client.Update(ctx, pod); err != nil {
+				return fmt.Errorf("update registry pod owner reference: %w", err)
+			}
+			return nil
+		})
+	}
+
 	csKey, err := client.ObjectKeyFromObject(cs)
 	if err != nil {
 		return fmt.Errorf("get catalogsource key: %v", err)
@@ -136,7 +248,7 @@ func (a *CatalogAdd) add(ctx context.Context, cs *v1alpha1.CatalogSource) error
 func (a *CatalogAdd) cleanup(cs *v1alpha1.CatalogSource) {
 	ctx, cancel := context.WithTimeout(context.Background(), a.CleanupTimeout)
 	defer cancel()
-	if err := a.config.Client.Delete(ctx, cs); err != nil {
+	if err := a.config.Client.Delete(ctx, cs); err != nil && !apierrors.IsNotFound(err) {
 		log.Printf("delete catalogsource %q: %v", cs.Name, err)
 	}
 }

internal/pkg/action/config.go

@@ -19,6 +19,7 @@ type Configuration struct {
 	RESTConfig *rest.Config
 	Client     client.Client
 	Namespace  string
+	Scheme     *runtime.Scheme
 
 	overrides *clientcmd.ConfigOverrides
 }
@@ -77,6 +78,7 @@ func (c *Configuration) Load() error {
 		return err
 	}
 
+	c.Scheme = sch
 	c.Client = &operatorClient{cl}
 	c.Namespace = ns
 	c.RESTConfig = cc

internal/pkg/catalog/catalog.go

@@ -8,6 +8,12 @@ import (
 
 type Option func(*v1alpha1.CatalogSource)
 
+func Annotations(v map[string]string) Option {
+	return func(cs *v1alpha1.CatalogSource) {
+		cs.ObjectMeta.Annotations = v
+	}
+}
+
 func DisplayName(v string) Option {
 	return func(cs *v1alpha1.CatalogSource) {
 		cs.Spec.DisplayName = v
@@ -25,6 +31,12 @@ func Publisher(v string) Option {
 	}
 }
 
+func Address(address string) Option {
+	return func(cs *v1alpha1.CatalogSource) {
+		cs.Spec.Address = address
+	}
+}
+
 func Build(key types.NamespacedName, opts ...Option) *v1alpha1.CatalogSource {
 	cs := &v1alpha1.CatalogSource{
 		ObjectMeta: metav1.ObjectMeta{