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

@@ -33,6 +33,27 @@ func (b *BundleUnpackResult) Bundle() *api.Bundle {
 	return b.bundle
 }
 
+func (b *BundleUnpackResult) Name() string {
+	return b.name
+}
+
+// SetCondition replaces the existing BundleLookupCondition of the same type, or adds it if it was not found.
+func (b *BundleUnpackResult) SetCondition(cond operatorsv1alpha1.BundleLookupCondition) operatorsv1alpha1.BundleLookupCondition {
+	for i, existing := range b.Conditions {
+		if existing.Type != cond.Type {
+			continue
+		}
+		if existing.Status == cond.Status && existing.Reason == cond.Reason {
+			cond.LastTransitionTime = existing.LastTransitionTime
+		}
+		b.Conditions[i] = cond
+		return cond
+	}
+	b.Conditions = append(b.Conditions, cond)
+
+	return cond
+}
+
 var catalogSourceGVK = operatorsv1alpha1.SchemeGroupVersion.WithKind(operatorsv1alpha1.CatalogSourceKind)
 
 func newBundleUnpackResult(lookup *operatorsv1alpha1.BundleLookup) *BundleUnpackResult {
@@ -250,16 +271,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
@@ -303,7 +334,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
@@ -318,6 +349,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/manifests.go

@@ -0,0 +1,109 @@
+package catalog
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	"github.com/operator-framework/operator-registry/pkg/configmap"
+	errorwrap "github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	v1 "k8s.io/client-go/listers/core/v1"
+
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
+)
+
+// ManifestResolver can dereference a manifest for a step. Steps may embed manifests directly or reference content
+// in configmaps
+type ManifestResolver interface {
+	ManifestForStep(step *v1alpha1.Step) (string, error)
+}
+
+// manifestResolver caches manifest from unpacked bundles (via configmaps)
+type manifestResolver struct {
+	configMapLister v1.ConfigMapLister
+	unpackedSteps   map[string][]v1alpha1.StepResource
+	namespace       string
+	logger          logrus.FieldLogger
+}
+
+func newManifestResolver(namespace string, configMapLister v1.ConfigMapLister, logger logrus.FieldLogger) *manifestResolver {
+	return &manifestResolver{
+		namespace:       namespace,
+		configMapLister: configMapLister,
+		unpackedSteps:   map[string][]v1alpha1.StepResource{},
+		logger:          logger,
+	}
+}
+
+// ManifestForStep always returns the manifest that should be applied to the cluster for a given step
+// the manifest field in the installplan status can contain a reference to a configmap instead
+func (r *manifestResolver) ManifestForStep(step *v1alpha1.Step) (string, error) {
+	manifest := step.Resource.Manifest
+	ref := refForStep(step, r.logger)
+	if ref == nil {
+		return manifest, nil
+	}
+
+	log := r.logger.WithFields(logrus.Fields{"resolving": step.Resolving, "step": step.Resource.Name})
+	log.WithField("ref", ref).Debug("step is a reference to configmap")
+
+	usteps, err := r.unpackedStepsForBundle(step.Resolving, ref)
+	if err != nil  {
+		return "", err
+	}
+
+	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")
+			break
+		}
+	}
+	if manifest == step.Resource.Manifest {
+		return "", fmt.Errorf("couldn't find unpacked step for %v", step)
+	}
+	return manifest, nil
+}
+
+func (r *manifestResolver) unpackedStepsForBundle(bundleName string, ref *UnpackedBundleReference) ([]v1alpha1.StepResource, error) {
+	usteps, ok := r.unpackedSteps[bundleName]
+	if ok {
+		return usteps, nil
+	}
+	cm, err := r.configMapLister.ConfigMaps(ref.Namespace).Get(ref.Name)
+	if err != nil {
+		return nil, errorwrap.Wrapf(err, "error finding unpacked bundle configmap for ref %v", *ref)
+	}
+	loader := configmap.NewBundleLoader()
+	bundle, err := loader.Load(cm)
+	if err != nil {
+		return nil, errorwrap.Wrapf(err, "error loading unpacked bundle configmap for ref %v", *ref)
+	}
+	steps, err := resolver.NewStepResourceFromBundle(bundle, r.namespace, ref.Replaces, ref.CatalogSourceName, ref.CatalogSourceNamespace)
+	if err != nil {
+		return nil, errorwrap.Wrapf(err, "error calculating steps for ref %v", *ref)
+	}
+	r.unpackedSteps[bundleName] = steps
+	return steps, nil
+}
+
+func refForStep(step *v1alpha1.Step, log logrus.FieldLogger) *UnpackedBundleReference {
+	log = log.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
+}

pkg/controller/operators/catalog/operator.go

@@ -1098,6 +1098,16 @@ 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
@@ -1110,32 +1120,53 @@ 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 {
 			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
+		}
+		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
 	}
 
@@ -1436,6 +1467,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 	}
 
 	ensurer := newStepEnsurer(kubeclient, crclient, dynamicClient)
+	r := newManifestResolver(plan.GetNamespace(), o.lister.CoreV1().ConfigMapLister(), o.logger)
 
 	for i, step := range plan.Status.Plan {
 		switch step.Status {
@@ -1474,6 +1506,10 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				continue
 			}
 		case v1alpha1.StepStatusUnknown, v1alpha1.StepStatusNotPresent:
+			manifest, err := r.ManifestForStep(step)
+			if err != nil {
+				return err
+			}
 			o.logger.WithFields(logrus.Fields{"kind": step.Resource.Kind, "name": step.Resource.Name}).Debug("execute resource")
 			switch step.Resource.Kind {
 			case crdKind:
@@ -1546,7 +1582,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)
 				}
@@ -1579,7 +1615,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)
 				}
@@ -1612,7 +1648,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)
 				}
@@ -1627,7 +1663,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)
 				}
@@ -1642,7 +1678,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)
 				}
@@ -1665,7 +1701,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)
 				}
@@ -1688,7 +1724,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)
 				}
@@ -1711,7 +1747,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)
 				}
@@ -1739,7 +1775,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)