Fail InstallPlan on bundle unpack timeout

The InstallPlan sync can stay stalled on the Installing phase if the bundle cannot be successfully unpacked.
Adding a configurable timeout for the duration of the bundle unpack Job helps identify if an unpack Job is
stalled, and the InstallPlan is then transitioned to Failed with the unpack Job's failure condition
propagated to the InstallPlan condition.

Author: hasbro17

cmd/catalog/main.go

@@ -69,6 +69,9 @@ var (
 
 	profiling = flag.Bool(
 		"profiling", false, "serve profiling data (on port 8080)")
+
+	// TODO(haseeb): Change to a more realistic timeout for bundle unpack jobs
+	bundleUnpackTimeout = flag.Duration("bundle-unpack-timeout", 60*time.Second, "The time duration after which the bundle unpack Job for an installplan will be aborted and the installplan will be Failed. 0 is considered as having no timeout.")
 )
 
 func init() {
@@ -173,7 +176,7 @@ func main() {
 	}
 
 	// Create a new instance of the operator.
-	op, err := catalog.NewOperator(ctx, *kubeConfigPath, utilclock.RealClock{}, logger, *wakeupInterval, *configmapServerImage, *utilImage, *catalogNamespace, k8sscheme.Scheme)
+	op, err := catalog.NewOperator(ctx, *kubeConfigPath, utilclock.RealClock{}, logger, *wakeupInterval, *configmapServerImage, *utilImage, *catalogNamespace, k8sscheme.Scheme, *bundleUnpackTimeout)
 	if err != nil {
 		log.Panicf("error configuring operator: %s", err.Error())
 	}

pkg/controller/bundle/bundle_unpacker.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"fmt"
+	"time"
 
 	"github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/configmap"
@@ -146,6 +147,12 @@ func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string
 	job.SetName(cmRef.Name)
 	job.SetOwnerReferences([]metav1.OwnerReference{ownerRef(cmRef)})
 
+	// Don't set a timeout if it is 0
+	if c.unpackTimeout != time.Duration(0) {
+		t := int64(c.unpackTimeout.Seconds())
+		job.Spec.ActiveDeadlineSeconds = &t
+	}
+
 	return job
 }
 
@@ -156,16 +163,17 @@ type Unpacker interface {
 }
 
 type ConfigMapUnpacker struct {
-	opmImage   string
-	utilImage  string
-	client     kubernetes.Interface
-	csLister   listersoperatorsv1alpha1.CatalogSourceLister
-	cmLister   listerscorev1.ConfigMapLister
-	jobLister  listersbatchv1.JobLister
-	roleLister listersrbacv1.RoleLister
-	rbLister   listersrbacv1.RoleBindingLister
-	loader     *configmap.BundleLoader
-	now        func() metav1.Time
+	opmImage      string
+	utilImage     string
+	client        kubernetes.Interface
+	csLister      listersoperatorsv1alpha1.CatalogSourceLister
+	cmLister      listerscorev1.ConfigMapLister
+	jobLister     listersbatchv1.JobLister
+	roleLister    listersrbacv1.RoleLister
+	rbLister      listersrbacv1.RoleBindingLister
+	loader        *configmap.BundleLoader
+	now           func() metav1.Time
+	unpackTimeout time.Duration
 }
 
 type ConfigMapUnpackerOption func(*ConfigMapUnpacker)
@@ -183,6 +191,12 @@ func NewConfigmapUnpacker(options ...ConfigMapUnpackerOption) (*ConfigMapUnpacke
 	return unpacker, nil
 }
 
+func WithUnpackTimeout(timeout time.Duration) ConfigMapUnpackerOption {
+	return func(unpacker *ConfigMapUnpacker) {
+		unpacker.unpackTimeout = timeout
+	}
+}
+
 func WithOPMImage(opmImage string) ConfigMapUnpackerOption {
 	return func(unpacker *ConfigMapUnpacker) {
 		unpacker.opmImage = opmImage
@@ -342,7 +356,22 @@ func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup)
 		return
 	}
 
-	if !jobConditionTrue(job, batchv1.JobComplete) && (cond.Status != corev1.ConditionTrue || cond.Reason != JobIncompleteReason) {
+	// Check if bundle unpack job has failed due a timeout
+	// Return a BundleJobError so we can mark the InstallPlan as Failed
+	isFailed, jobCond := jobConditionTrue(job, batchv1.JobFailed)
+	if isFailed {
+		cond.Status = corev1.ConditionTrue
+		cond.Reason = jobCond.Reason
+		cond.Message = jobCond.Message
+		cond.LastTransitionTime = &now
+		result.SetCondition(cond)
+
+		err = NewBundleJobError(fmt.Sprintf("Bundle extract Job failed with Reason: %v, and Message: %v", jobCond.Reason, jobCond.Message))
+		return
+	}
+
+	isComplete, _ := jobConditionTrue(job, batchv1.JobComplete)
+	if !isComplete && (cond.Status != corev1.ConditionTrue || cond.Reason != JobIncompleteReason) {
 		cond.Status = corev1.ConditionTrue
 		cond.Reason = JobIncompleteReason
 		cond.Message = JobIncompleteMessage
@@ -522,15 +551,16 @@ func ownerRef(ref *corev1.ObjectReference) metav1.OwnerReference {
 }
 
 // jobConditionTrue returns true if the given job has the given condition with the given condition type true, and returns false otherwise.
-func jobConditionTrue(job *batchv1.Job, conditionType batchv1.JobConditionType) bool {
+// Also returns the condition if true
+func jobConditionTrue(job *batchv1.Job, conditionType batchv1.JobConditionType) (bool, *batchv1.JobCondition) {
 	if job == nil {
-		return false
+		return false, nil
 	}
 
 	for _, cond := range job.Status.Conditions {
 		if cond.Type == conditionType && cond.Status == corev1.ConditionTrue {
-			return true
+			return true, &cond
 		}
 	}
-	return false
+	return false, nil
 }

pkg/controller/bundle/errors.go

@@ -0,0 +1,28 @@
+package bundle
+
+// bundleJobError is an error type returned by UnpackBundle() when the
+// bundle unpack job fails (e.g due to a timeout)
+type bundleJobError struct {
+	s string
+}
+
+func NewBundleJobError(s string) error {
+	return bundleJobError{s: s}
+}
+
+func (e bundleJobError) Error() string {
+	return e.s
+}
+
+func (e bundleJobError) IsBundleJobError() bool {
+	return true
+}
+
+// IsBundleJobError checks if an error is an error due to the bundle extract job failing.
+func IsBundleJobError(err error) bool {
+	type bundleJobError interface {
+		IsBundleJobError() bool
+	}
+	ogErr, ok := err.(bundleJobError)
+	return ok && ogErr.IsBundleJobError()
+}

pkg/controller/operators/catalog/operator.go

@@ -105,12 +105,13 @@ type Operator struct {
 	clientAttenuator         *scoped.ClientAttenuator
 	serviceAccountQuerier    *scoped.UserDefinedServiceAccountQuerier
 	bundleUnpacker           bundle.Unpacker
+	bundleUnpackTimeout      time.Duration
 }
 
 type CatalogSourceSyncFunc func(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, syncError error)
 
 // NewOperator creates a new Catalog Operator.
-func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resync time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string, scheme *runtime.Scheme) (*Operator, error) {
+func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resync time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string, scheme *runtime.Scheme, bundleUnpackTimeout time.Duration) (*Operator, error) {
 	resyncPeriod := queueinformer.ResyncWithJitter(resync, 0.2)
 	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
 	if err != nil {
@@ -168,6 +169,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		catalogSubscriberIndexer: map[string]cache.Indexer{},
 		serviceAccountQuerier:    scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
 		clientAttenuator:         scoped.NewClientAttenuator(logger, config, opClient, crClient, dynamicClient),
+		bundleUnpackTimeout:      bundleUnpackTimeout,
 	}
 	op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now, ssaClient)
@@ -341,6 +343,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		bundle.WithOPMImage(configmapRegistryImage),
 		bundle.WithUtilImage(utilImage),
 		bundle.WithNow(op.now),
+		bundle.WithUnpackTimeout(op.bundleUnpackTimeout),
 	)
 	if err != nil {
 		return nil, err
@@ -1182,6 +1185,10 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.In
 		lookup := out.Status.BundleLookups[i]
 		res, err := o.bundleUnpacker.UnpackBundle(&lookup)
 		if err != nil {
+			// If the bundle unpack job fails abort early to fail the InstallPlan
+			if bundle.IsBundleJobError(err) {
+				return false, nil, err
+			}
 			errs = append(errs, err)
 			continue
 		}
@@ -1349,20 +1356,10 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 		// Mark the InstallPlan as failed for a fatal Operator Group related error
 		logger.Infof("attenuated service account query failed - %v", err)
 		ipFailError := fmt.Errorf("invalid operator group - %v", err)
-		now := o.now()
-		out := plan.DeepCopy()
-		out.Status.SetCondition(v1alpha1.ConditionFailed(v1alpha1.InstallPlanInstalled,
-			v1alpha1.InstallPlanReasonInstallCheckFailed, ipFailError.Error(), &now))
-		out.Status.Phase = v1alpha1.InstallPlanPhaseFailed
-
-		logger.Info("transitioning InstallPlan to failed")
-		if _, err := o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(context.TODO(), out, metav1.UpdateOptions{}); err != nil {
-			updateErr := errors.New("error updating InstallPlan status: " + err.Error())
-			logger = logger.WithField("updateError", updateErr)
-			logger.Errorf("error transitioning InstallPlan to failed")
-
-			// retry sync with error to update InstallPlan status
-			syncError = fmt.Errorf("InstallPlan failed: %s and error updating InstallPlan status as failed: %s", ipFailError, updateErr)
+
+		if err := o.transitionInstallPlanToFailed(plan, logger, v1alpha1.InstallPlanReasonInstallCheckFailed, ipFailError.Error()); err != nil {
+			// retry for failure to update status
+			syncError = err
 			return
 		}
 
@@ -1392,7 +1389,23 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 	if len(plan.Status.BundleLookups) > 0 {
 		unpacked, out, err := o.unpackBundles(plan)
 		if err != nil {
-			syncError = fmt.Errorf("bundle unpacking failed: %v", err)
+			// Retry sync if non-fatal error
+			if !bundle.IsBundleJobError(err) {
+				syncError = fmt.Errorf("bundle unpacking failed: %v", err)
+				return
+			}
+
+			// Mark the InstallPlan as failed for a fatal bundle unpack error
+			logger.Infof("bundle unpacking failed: %v", err)
+
+			if err := o.transitionInstallPlanToFailed(plan, logger, v1alpha1.InstallPlanReasonInstallCheckFailed, err.Error()); err != nil {
+				// retry for failure to update status
+				syncError = err
+				return
+			}
+
+			// Requeue subscription to propagate SubscriptionInstallPlanFailed condtion to subscription
+			o.requeueSubscriptionForInstallPlan(plan, logger)
 			return
 		}
 
@@ -1445,6 +1458,28 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 	return
 }
 
+func (o *Operator) transitionInstallPlanToFailed(plan *v1alpha1.InstallPlan, logger *logrus.Entry, reason v1alpha1.InstallPlanConditionReason, message string) (syncError error) {
+	now := o.now()
+	out := plan.DeepCopy()
+	out.Status.SetCondition(v1alpha1.ConditionFailed(v1alpha1.InstallPlanInstalled,
+		reason, message, &now))
+	out.Status.Phase = v1alpha1.InstallPlanPhaseFailed
+
+	logger.Info("transitioning InstallPlan to failed")
+	_, err := o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(context.TODO(), out, metav1.UpdateOptions{})
+	if err == nil {
+		return
+	}
+
+	updateErr := errors.New("error updating InstallPlan status: " + err.Error())
+	logger = logger.WithField("updateError", updateErr)
+	logger.Errorf("error transitioning InstallPlan to failed")
+
+	// retry sync with error to update InstallPlan status
+	syncError = fmt.Errorf("InstallPlan failed: %s and error updating InstallPlan status as failed: %s", message, updateErr)
+	return
+}
+
 func (o *Operator) requeueSubscriptionForInstallPlan(plan *v1alpha1.InstallPlan, logger *logrus.Entry) {
 	// Notify subscription loop of installplan changes
 	owners := ownerutil.GetOwnersByKind(plan, v1alpha1.SubscriptionKind)

test/e2e/installplan_e2e_test.go

@@ -36,6 +36,7 @@ import (
 
 	opver "github.com/operator-framework/api/pkg/lib/version"
 	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/catalog"
@@ -2913,6 +2914,111 @@ var _ = Describe("Install Plan", func() {
 		ctx.Ctx().Logf(fmt.Sprintf("Install plan %s fetched with conditions %+v", fetchedInstallPlan.GetName(), fetchedInstallPlan.Status.Conditions))
 	})
 
+	It("should fail an InstallPlan when an InstallPlan has a bundle unpack job timeout", func() {
+		ns := &corev1.Namespace{}
+		ns.SetName(genName("ns-"))
+
+		c := newKubeClient()
+		crc := newCRClient()
+		now := metav1.Now()
+
+		// Create a namespace
+		ns, err := c.KubernetesInterface().CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
+		Expect(err).NotTo(HaveOccurred())
+		deleteOpts := &metav1.DeleteOptions{}
+		defer func() {
+			err = c.KubernetesInterface().CoreV1().Namespaces().Delete(context.TODO(), ns.GetName(), *deleteOpts)
+			Expect(err).ToNot(HaveOccurred())
+		}()
+
+		// Create the single (kiali) bundle catalog source
+		catsrc := &operatorsv1alpha1.CatalogSource{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      genName("kiali-"),
+				Namespace: ns.GetName(),
+				Labels:    map[string]string{"olm.catalogSource": "kaili-catalog"},
+			},
+			Spec: operatorsv1alpha1.CatalogSourceSpec{
+				Image:      "quay.io/olmtest/single-bundle-index:1.0.0",
+				SourceType: operatorsv1alpha1.SourceTypeGrpc,
+			},
+		}
+		catsrc, err = crc.OperatorsV1alpha1().CatalogSources(catsrc.GetNamespace()).Create(context.TODO(), catsrc, metav1.CreateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		// Wait for the CatalogSource to be ready
+		catsrc, err = fetchCatalogSourceOnStatus(crc, catsrc.GetName(), catsrc.GetNamespace(), catalogSourceRegistryPodSynced)
+		Expect(err).ToNot(HaveOccurred())
+
+		// Create the OperatorGroup
+		og := &operatorsv1.OperatorGroup{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "og",
+			},
+			Spec: operatorsv1.OperatorGroupSpec{
+				TargetNamespaces: []string{ns.GetName()},
+			},
+		}
+		_, err = crc.OperatorsV1().OperatorGroups(ns.GetName()).Create(context.TODO(), og, metav1.CreateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		// Create an InstallPlan in the Planning phase with status.bundleLookups specified for a non-existing bundle
+		ip := &operatorsv1alpha1.InstallPlan{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "ip",
+				Namespace: ns.GetName(),
+			},
+			Spec: operatorsv1alpha1.InstallPlanSpec{
+				ClusterServiceVersionNames: []string{"foobar"},
+				Approval:                   v1alpha1.ApprovalAutomatic,
+				Approved:                   true,
+				Generation:                 1,
+			},
+			Status: operatorsv1alpha1.InstallPlanStatus{
+				Phase:          operatorsv1alpha1.InstallPlanPhasePlanning,
+				CatalogSources: []string{},
+				BundleLookups: []operatorsv1alpha1.BundleLookup{
+					{
+						Path:       "quay.io/foo/bar:v0.0.1",
+						Identifier: "foobar.v0.0.1",
+						CatalogSourceRef: &corev1.ObjectReference{
+							Namespace: ns.GetName(),
+							Name:      catsrc.GetName(),
+						},
+						Conditions: []operatorsv1alpha1.BundleLookupCondition{
+							{
+								Type:               operatorsv1alpha1.BundleLookupPending,
+								Status:             corev1.ConditionTrue,
+								Reason:             "JobIncomplete",
+								Message:            "unpack job not completed",
+								LastTransitionTime: &now,
+							},
+						},
+					},
+				},
+			},
+		}
+
+		outIP, err := crc.OperatorsV1alpha1().InstallPlans(ns.GetName()).Create(context.TODO(), ip, metav1.CreateOptions{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(outIP).NotTo(BeNil())
+
+		// The status gets ignored on create so we need to update it else the InstallPlan sync ignores
+		// InstallPlans without any steps or bundle lookups
+		outIP.Status = ip.Status
+		_, err = crc.OperatorsV1alpha1().InstallPlans(ns.GetName()).UpdateStatus(context.TODO(), outIP, metav1.UpdateOptions{})
+		Expect(err).NotTo(HaveOccurred())
+
+		// The InstallPlan should be Failed due to the bundle unpack job timeout
+		// TODO(haseeb): Use Eventually() to set a Duration just over the default bundle-unpack-timeout duration. Else this could poll for a long time.
+		fetchedInstallPlan, err := fetchInstallPlanWithNamespace(GinkgoT(), crc, ip.Name, ns.GetName(), buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
+		Expect(err).NotTo(HaveOccurred())
+		Expect(fetchedInstallPlan).NotTo(BeNil())
+		ctx.Ctx().Logf(fmt.Sprintf("Install plan %s fetched with phase %s", fetchedInstallPlan.GetName(), fetchedInstallPlan.Status.Phase))
+		ctx.Ctx().Logf(fmt.Sprintf("Install plan %s fetched with conditions %+v", fetchedInstallPlan.GetName(), fetchedInstallPlan.Status.Conditions))
+
+	})
+
 	It("compresses installplan step resource manifests to configmap references", func() {
 		// Test ensures that all steps for index-based catalogs are references to configmaps. This avoids the problem
 		// of installplans growing beyond the etcd size limit when manifests are written to the ip status.