fix(installplan): don't store the manifest contents on the installplan

the installplan status.plan field stores a list of all manifests that
will be installed when the installplan is approved.

if the installplan is resolving a large set of dependencies, it is
possible for it to exceed various limits in kube (grpc request size
limit, etcd value size limit, etc).

this commit changes the way objects are tracked if they come from
bundle images. in that case, the manifests are already unpacked onto
the cluster into a configmap. the installplan now stores a reference
to the source configmap for each step, rather than storing the manifest
itself. this allows the installplan to scale to a large number of
resolved objects.

Author: ecordell

pkg/controller/bundle/bundle_unpacker.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"crypto/sha256"
 	"fmt"
-
 	"github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/configmap"
 	batchv1 "k8s.io/api/batch/v1"
@@ -34,6 +33,10 @@ func (b *BundleUnpackResult) Bundle() *api.Bundle {
 	return b.bundle
 }
 
+func (b *BundleUnpackResult) Name() string {
+	return b.name
+}
+
 var catalogSourceGVK = operatorsv1alpha1.SchemeGroupVersion.WithKind(operatorsv1alpha1.CatalogSourceKind)
 
 func newBundleUnpackResult(lookup *operatorsv1alpha1.BundleLookup) *BundleUnpackResult {
@@ -251,16 +254,26 @@ const (
 	CatalogSourceMissingMessage = "referenced catalogsource not found"
 	JobIncompleteReason         = "JobIncomplete"
 	JobIncompleteMessage        = "unpack job not completed"
+	JobNotStartedReason         = "JobNotStarted"
+	JobNotStartedMessage        = "unpack job not yet started"
+	NotUnpackedReason           = "BundleNotUnpacked"
+	NotUnpackedMessage          = "bundle contents have not yet been persisted to installplan status"
 )
 
 func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup) (result *BundleUnpackResult, err error) {
 	result = newBundleUnpackResult(lookup)
+
+	// if pending condition is missing, bundle has already been unpacked
 	cond := result.GetCondition(operatorsv1alpha1.BundleLookupPending)
+	if cond.Status == corev1.ConditionUnknown {
+		return result, nil
+	}
+
 	now := c.now()
 
 	var cs *operatorsv1alpha1.CatalogSource
 	if cs, err = c.csLister.CatalogSources(result.CatalogSourceRef.Namespace).Get(result.CatalogSourceRef.Name); err != nil {
-		if apierrors.IsNotFound(err) && cond.Status != corev1.ConditionTrue && cond.Reason != CatalogSourceMissingReason {
+		if apierrors.IsNotFound(err) && cond.Reason != CatalogSourceMissingReason {
 			cond.Status = corev1.ConditionTrue
 			cond.Reason = CatalogSourceMissingReason
 			cond.Message = CatalogSourceMissingMessage
@@ -304,7 +317,7 @@ func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup)
 		return
 	}
 
-	if !jobConditionTrue(job, batchv1.JobComplete) && cond.Status != corev1.ConditionTrue && cond.Reason != JobIncompleteReason {
+	if !jobConditionTrue(job, batchv1.JobComplete) && (cond.Status != corev1.ConditionTrue || cond.Reason != JobIncompleteReason) {
 		cond.Status = corev1.ConditionTrue
 		cond.Reason = JobIncompleteReason
 		cond.Message = JobIncompleteMessage
@@ -319,6 +332,10 @@ func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup)
 		return
 	}
 
+	if result.Bundle() == nil || len(result.Bundle().GetObject()) == 0 {
+		return
+	}
+
 	// A successful load should remove the pending condition
 	result.RemoveCondition(operatorsv1alpha1.BundleLookupPending)
 

pkg/controller/bundle/bundle_unpacker_test.go

@@ -71,6 +71,14 @@ func TestConfigMapUnpacker(t *testing.T) {
 						Namespace: "ns-a",
 						Name:      "src-a",
 					},
+					Conditions: []operatorsv1alpha1.BundleLookupCondition{
+						{
+							Type:    operatorsv1alpha1.BundleLookupPending,
+							Status:  corev1.ConditionTrue,
+							Reason:  JobNotStartedReason,
+							Message: JobNotStartedMessage,
+						},
+					},
 				},
 			},
 			expected: expected{
@@ -116,6 +124,14 @@ func TestConfigMapUnpacker(t *testing.T) {
 						Namespace: "ns-a",
 						Name:      "src-a",
 					},
+					Conditions: []operatorsv1alpha1.BundleLookupCondition{
+						{
+							Type:    operatorsv1alpha1.BundleLookupPending,
+							Status:  corev1.ConditionTrue,
+							Reason:  JobNotStartedReason,
+							Message: JobNotStartedMessage,
+						},
+					},
 				},
 			},
 			expected: expected{

pkg/controller/operators/catalog/operator.go

@@ -53,6 +53,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/scoped"
 	sharedtime "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/time"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/metrics"
+	"github.com/operator-framework/operator-registry/pkg/configmap"
 )
 
 const (
@@ -1105,9 +1106,20 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 	return reference.GetReference(res)
 }
 
+type UnpackedBundleReference struct {
+	Kind                   string `json:"kind"`
+	Name                   string `json:"name"`
+	Namespace              string `json:"namespace"`
+	CatalogSourceName      string `json:"catalogSourceName"`
+	CatalogSourceNamespace string `json:"catalogSourceNamespace"`
+	Replaces               string `json:"replaces"`
+}
+
+// unpackBundles makes one walk through the bundlelookups and attempts to progress them
 func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.InstallPlan, error) {
 	out := plan.DeepCopy()
 	unpacked := true
+	unpackedCount := len(out.Status.BundleLookups)
 
 	var errs []error
 	for i := 0; i < len(out.Status.BundleLookups); i++ {
@@ -1117,32 +1129,55 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.In
 			errs = append(errs, err)
 			continue
 		}
+		out.Status.BundleLookups[i] = *res.BundleLookup
 
-		if res == nil {
+		// if pending condition is present, still waiting for the job to unpack to configmap
+		if res.GetCondition(v1alpha1.BundleLookupPending).Status == corev1.ConditionTrue {
 			unpacked = false
 			continue
 		}
 
-		out.Status.BundleLookups[i] = *res.BundleLookup
-		if res.Bundle() == nil || len(res.Bundle().GetObject()) == 0 {
-			unpacked = false
+		// if packed condition is missing, bundle has already been unpacked into steps, continue
+		if res.GetCondition(resolver.BundleLookupConditionPacked).Status == corev1.ConditionUnknown {
+			unpackedCount--
 			continue
 		}
 
+		// Ensure that bundle can be applied by the current version of OLM by converting to steps
 		steps, err := resolver.NewStepsFromBundle(res.Bundle(), out.GetNamespace(), res.Replaces, res.CatalogSourceRef.Name, res.CatalogSourceRef.Namespace)
 		if err != nil {
-			errs = append(errs, fmt.Errorf("failed to turn bundle into steps: %s", err.Error()))
+			errs = append(errs, fmt.Errorf("failed to turn bundle into steps: %v", err))
 			unpacked = false
 			continue
 		}
 
-		// Add steps and remove resolved bundle lookup
+		// step manifests are replaced with references to the configmap containing them
+		for i, s := range steps {
+			ref := UnpackedBundleReference{
+				Kind:                   "ConfigMap",
+				Namespace:              res.CatalogSourceRef.Namespace,
+				Name:                   res.Name(),
+				CatalogSourceName:      res.CatalogSourceRef.Name,
+				CatalogSourceNamespace: res.CatalogSourceRef.Namespace,
+				Replaces:               res.Replaces,
+			}
+			r, err := json.Marshal(&ref)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("failed to generate reference for configmap: %v", err))
+				unpacked = false
+				continue
+			}
+			s.Resource.Manifest = string(r)
+			steps[i] = s
+		}
+		unpackedCount--
+		res.RemoveCondition(resolver.BundleLookupConditionPacked)
+		out.Status.BundleLookups[i] = *res.BundleLookup
 		out.Status.Plan = append(out.Status.Plan, steps...)
-		out.Status.BundleLookups = append(out.Status.BundleLookups[:i], out.Status.BundleLookups[i+1:]...)
-		i--
 	}
 
 	if err := utilerrors.NewAggregate(errs); err != nil {
+		o.logger.Debugf("failed to unpack bundles: %v", err)
 		return false, nil, err
 	}
 
@@ -1418,9 +1453,79 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 	ensurer := newStepEnsurer(kubeclient, crclient, dynamicClient)
 	b := newBuilder(kubeclient, dynamicClient, o.csvProvidedAPIsIndexer, o.logger)
 
+	refForStep := func(step *v1alpha1.Step) *UnpackedBundleReference {
+		log := o.logger.WithFields(logrus.Fields{"resolving": step.Resolving, "step": step.Resource.Name})
+		var ref UnpackedBundleReference
+		if err := json.Unmarshal([]byte(step.Resource.Manifest), &ref); err != nil {
+			log.Debug("step is not a reference to an unpacked bundle (this is not an error if the step is a manifest)")
+			return nil
+		}
+		log = log.WithField("ref", ref)
+		if ref.Kind != "ConfigMap" || ref.Name == "" || ref.Namespace == "" || ref.CatalogSourceName == "" || ref.CatalogSourceNamespace == "" {
+			log.Debug("step is not a reference to an unpacked bundle (this is not an error if the step is a manifest)")
+			return nil
+		}
+		return &ref
+	}
+
+	// for each step that points to a configmap instead of holding the full step manifest, find the steps and cache them
+	unpackedSteps := map[string][]v1alpha1.StepResource{}
+	loader := configmap.NewBundleLoader()
+	for _, step := range plan.Status.Plan {
+		ref := refForStep(step)
+		if ref == nil {
+			continue
+		}
+		cm, err := o.lister.CoreV1().ConfigMapLister().ConfigMaps(ref.Namespace).Get(ref.Name)
+		if err != nil {
+			return errorwrap.Wrapf(err, "error finding unpacked bundle configmap for ref %s", step.Resource.Manifest)
+		}
+		b, err := loader.Load(cm)
+		if err != nil {
+			return errorwrap.Wrapf(err, "error loading unpacked bundle configmap for ref %s", step.Resource.Manifest)
+		}
+		steps, err := resolver.NewStepResourceFromBundle(b, plan.GetNamespace(), ref.Replaces, ref.CatalogSourceName, ref.CatalogSourceNamespace)
+		if err != nil {
+			return errorwrap.Wrapf(err, "error calculating steps for ref %s", step.Resource.Manifest)
+		}
+		unpackedSteps[step.Resolving] = steps
+	}
+
+	manifestForStep := func(step *v1alpha1.Step) (string, error) {
+		manifest := step.Resource.Manifest
+		ref := refForStep(step)
+		if ref != nil {
+			log := o.logger.WithFields(logrus.Fields{"resolving": step.Resolving, "step": step.Resource.Name})
+			log.WithField("ref", ref).Debug("step is a reference to configmap")
+			if usteps, ok := unpackedSteps[step.Resolving]; ok {
+				log.Debugf("checking cache for unpacked step")
+				// need to find the real manifest from the unpacked steps
+				for _, u := range usteps {
+					if u.Name == step.Resource.Name &&
+						u.Kind == step.Resource.Kind &&
+						u.Version == step.Resource.Version &&
+						u.Group == step.Resource.Group {
+						manifest = u.Manifest
+						log.WithField("manifest", manifest).Debug("step replaced with unpacked value")
+					}
+				}
+			}
+			if manifest == step.Resource.Manifest {
+				return "", errorwrap.Wrapf(err, "couldn't find unpacked step for %v", step)
+			}
+		}
+		return manifest, nil
+	}
+
 	for i, step := range plan.Status.Plan {
+		// TODO: should lazy load only when the manifest is actually needed
+		manifest, err := manifestForStep(step)
+		if err != nil {
+			return err
+		}
+
 		doStep := true
-		s, err := b.create(step)
+		s, err := b.create(step, manifest)
 		if err != nil {
 			if _, ok := err.(notSupportedStepperErr); ok {
 				// stepper not implemented for this type yet
@@ -1448,7 +1553,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case v1alpha1.ClusterServiceVersionKind:
 				// Marshal the manifest into a CSV instance.
 				var csv v1alpha1.ClusterServiceVersion
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &csv)
+				err := json.Unmarshal([]byte(manifest), &csv)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1481,7 +1586,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case v1alpha1.SubscriptionKind:
 				// Marshal the manifest into a subscription instance.
 				var sub v1alpha1.Subscription
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &sub)
+				err := json.Unmarshal([]byte(manifest), &sub)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1505,7 +1610,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 
 			case resolver.BundleSecretKind:
 				var s corev1.Secret
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &s)
+				err := json.Unmarshal([]byte(manifest), &s)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1544,7 +1649,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case clusterRoleKind:
 				// Marshal the manifest into a ClusterRole instance.
 				var cr rbacv1.ClusterRole
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &cr)
+				err := json.Unmarshal([]byte(manifest), &cr)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1559,7 +1664,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case clusterRoleBindingKind:
 				// Marshal the manifest into a RoleBinding instance.
 				var rb rbacv1.ClusterRoleBinding
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &rb)
+				err := json.Unmarshal([]byte(manifest), &rb)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1574,7 +1679,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case roleKind:
 				// Marshal the manifest into a Role instance.
 				var r rbacv1.Role
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &r)
+				err := json.Unmarshal([]byte(manifest), &r)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1597,7 +1702,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case roleBindingKind:
 				// Marshal the manifest into a RoleBinding instance.
 				var rb rbacv1.RoleBinding
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &rb)
+				err := json.Unmarshal([]byte(manifest), &rb)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1620,7 +1725,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case serviceAccountKind:
 				// Marshal the manifest into a ServiceAccount instance.
 				var sa corev1.ServiceAccount
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &sa)
+				err := json.Unmarshal([]byte(manifest), &sa)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1643,7 +1748,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 			case serviceKind:
 				// Marshal the manifest into a Service instance
 				var s corev1.Service
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &s)
+				err := json.Unmarshal([]byte(manifest), &s)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1673,7 +1778,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 
 			case configMapKind:
 				var cfg corev1.ConfigMap
-				err := json.Unmarshal([]byte(step.Resource.Manifest), &cfg)
+				err := json.Unmarshal([]byte(manifest), &cfg)
 				if err != nil {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
@@ -1709,7 +1814,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				}
 
 				// Marshal the manifest into an unstructured object
-				dec := yaml.NewYAMLOrJSONDecoder(strings.NewReader(step.Resource.Manifest), 10)
+				dec := yaml.NewYAMLOrJSONDecoder(strings.NewReader(manifest), 10)
 				unstructuredObject := &unstructured.Unstructured{}
 				if err := dec.Decode(unstructuredObject); err != nil {
 					return errorwrap.Wrapf(err, "error decoding %s object to an unstructured object", step.Resource.Name)

pkg/controller/operators/catalog/operator_test.go

@@ -269,13 +269,13 @@ func TestExecutePlan(t *testing.T) {
 							Version:                "v1",
 							Kind:                   "ConfigMap",
 							Name:                   "cfg",
-							Manifest:               toManifest(t, configmap("cfg", namespace)),
+							Manifest:               toManifest(t, configMap("cfg", namespace)),
 						},
 						Status: v1alpha1.StepStatusUnknown,
 					},
 				},
 			),
-			want: []runtime.Object{configmap("cfg", namespace)},
+			want: []runtime.Object{configMap("cfg", namespace)},
 			err:  nil,
 		},
 		{
@@ -1307,7 +1307,7 @@ func serviceAccount(name, namespace, generateName string, secretRef *corev1.Obje
 	}
 }
 
-func configmap(name, namespace string) *corev1.ConfigMap {
+func configMap(name, namespace string) *corev1.ConfigMap {
 	return &corev1.ConfigMap{
 		TypeMeta:   metav1.TypeMeta{Kind: configMapKind},
 		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace},