Parse timeout annotation before UnpackBundle()

Author: hasbro17

cmd/catalog/main.go

@@ -71,7 +71,7 @@ var (
 		"profiling", false, "serve profiling data (on port 8080)")
 
 	installPlanTimeout  = flag.Duration("install-plan-retry-timeout", 1*time.Minute, "time since first attempt at which plan execution errors are considered fatal")
-	bundleUnpackTimeout = flag.Duration("bundle-unpack-timeout", 10*time.Minute, "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.")
+	bundleUnpackTimeout = flag.Duration("bundle-unpack-timeout", 10*time.Minute, "The time limit for bundle unpacking, after which InstallPlan execution is considered to have failed. 0 is considered as having no timeout.")
 )
 
 func init() {

pkg/controller/bundle/bundle_unpacker.go

@@ -81,7 +81,7 @@ func newBundleUnpackResult(lookup *operatorsv1alpha1.BundleLookup) *BundleUnpack
 	}
 }
 
-func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string, secrets []corev1.LocalObjectReference, ipAnnotations map[string]string) *batchv1.Job {
+func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string, secrets []corev1.LocalObjectReference, annotationUnpackTimeout time.Duration) *batchv1.Job {
 	job := &batchv1.Job{
 		Spec: batchv1.JobSpec{
 			//ttlSecondsAfterFinished: 0 // can use in the future to not have to clean up job
@@ -190,7 +190,6 @@ func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string
 	// We want to fail faster than that when we have repeated failures from the bundle unpack pod
 	// so we set it to 3 which is ~1m of waiting time
 	// See: https://kubernetes.io/docs/concepts/workloads/controllers/job/#pod-backoff-failure-policy
-	// TODO (haseeb): Should this be configurable as well?
 	backOffLimit := int32(3)
 	job.Spec.BackoffLimit = &backOffLimit
 
@@ -201,22 +200,18 @@ func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string
 		job.Spec.ActiveDeadlineSeconds = &t
 	}
 
-	// The bundle timeout annotation if specified overrides the --bundle-unpack-timeout flag value
-	timeoutStr, ok := ipAnnotations[BundleUnpackTimeoutAnnotationKey]
-	if !ok {
+	// Check annotationUnpackTimeout which is the annotation override for the default unpack timeout
+	// A negative timeout means the annotation was unset or malformed so we ignore it
+	if annotationUnpackTimeout < time.Duration(0) {
 		return job
 	}
-	d, err := time.ParseDuration(timeoutStr)
-	if err != nil {
-		// TODO(haseeb): log this error
-		return job
-	}
-	if d == time.Duration(0) {
-		// 0 means no timeout
+	// // 0 means no timeout so we unset ActiveDeadlineSeconds
+	if annotationUnpackTimeout == time.Duration(0) {
+		job.Spec.ActiveDeadlineSeconds = nil
 		return job
 	}
 
-	timeoutSeconds := int64(d.Seconds())
+	timeoutSeconds := int64(annotationUnpackTimeout.Seconds())
 	job.Spec.ActiveDeadlineSeconds = &timeoutSeconds
 
 	return job
@@ -225,7 +220,7 @@ func (c *ConfigMapUnpacker) job(cmRef *corev1.ObjectReference, bundlePath string
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . Unpacker
 
 type Unpacker interface {
-	UnpackBundle(lookup *operatorsv1alpha1.BundleLookup, ipAnnotations map[string]string) (result *BundleUnpackResult, err error)
+	UnpackBundle(lookup *operatorsv1alpha1.BundleLookup, annotationUnpackTimeout time.Duration) (result *BundleUnpackResult, err error)
 }
 
 type ConfigMapUnpacker struct {
@@ -372,7 +367,7 @@ const (
 	NotUnpackedMessage          = "bundle contents have not yet been persisted to installplan status"
 )
 
-func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup, ipAnnotations map[string]string) (result *BundleUnpackResult, err error) {
+func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup, annotationUnpackTimeout time.Duration) (result *BundleUnpackResult, err error) {
 
 	result = newBundleUnpackResult(lookup)
 
@@ -435,7 +430,7 @@ func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup,
 		secrets = append(secrets, corev1.LocalObjectReference{Name: secretName})
 	}
 	var job *batchv1.Job
-	job, err = c.ensureJob(cmRef, result.Path, secrets, ipAnnotations)
+	job, err = c.ensureJob(cmRef, result.Path, secrets, annotationUnpackTimeout)
 	if err != nil {
 		return
 	}
@@ -559,8 +554,8 @@ func (c *ConfigMapUnpacker) ensureConfigmap(csRef *corev1.ObjectReference, name
 	return
 }
 
-func (c *ConfigMapUnpacker) ensureJob(cmRef *corev1.ObjectReference, bundlePath string, secrets []corev1.LocalObjectReference, ipAnnotations map[string]string) (job *batchv1.Job, err error) {
-	fresh := c.job(cmRef, bundlePath, secrets, ipAnnotations)
+func (c *ConfigMapUnpacker) ensureJob(cmRef *corev1.ObjectReference, bundlePath string, secrets []corev1.LocalObjectReference, annotationUnpackTimeout time.Duration) (job *batchv1.Job, err error) {
+	fresh := c.job(cmRef, bundlePath, secrets, annotationUnpackTimeout)
 	job, err = c.jobLister.Jobs(fresh.GetNamespace()).Get(fresh.GetName())
 	if err != nil {
 		if apierrors.IsNotFound(err) {

pkg/controller/bundle/bundle_unpacker_test.go

@@ -795,7 +795,8 @@ func TestConfigMapUnpacker(t *testing.T) {
 			)
 			require.NoError(t, err)
 
-			res, err := unpacker.UnpackBundle(tt.args.lookup, map[string]string{})
+			// The annotationUnpackTimeout arg is negative so it is ignored for these tests
+			res, err := unpacker.UnpackBundle(tt.args.lookup, -1*time.Minute)
 			require.Equal(t, tt.expected.err, err)
 
 			if tt.expected.res == nil {

pkg/controller/bundle/bundlefakes/fake_unpacker.go

@@ -1183,10 +1183,23 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.In
 	out := plan.DeepCopy()
 	unpacked := true
 
+	// The bundle timeout annotation if specified overrides the --bundle-unpack-timeout flag value
+	// If the timeout cannot be parsed it's set to < 0 and subsequently ignored
+	unpackTimeout := -1 * time.Minute
+	timeoutStr, ok := plan.GetAnnotations()[bundle.BundleUnpackTimeoutAnnotationKey]
+	if ok {
+		d, err := time.ParseDuration(timeoutStr)
+		if err != nil {
+			o.logger.Errorf("failed to parse unpack timeout annotation(%s: %s): %v", bundle.BundleUnpackTimeoutAnnotationKey, timeoutStr, err)
+		} else {
+			unpackTimeout = d
+		}
+	}
+
 	var errs []error
 	for i := 0; i < len(out.Status.BundleLookups); i++ {
 		lookup := out.Status.BundleLookups[i]
-		res, err := o.bundleUnpacker.UnpackBundle(&lookup, plan.GetAnnotations())
+		res, err := o.bundleUnpacker.UnpackBundle(&lookup, unpackTimeout)
 		if err != nil {
 			errs = append(errs, err)
 			continue

pkg/controller/operators/catalog/operator.go

@@ -3149,14 +3149,14 @@ var _ = Describe("Install Plan", func() {
 			Expect(ctx.Ctx().Client().Status().Update(context.Background(), ip)).To(Succeed())
 
 			// The InstallPlan should fail after the unpack pod keeps failing and exceeds the job's
-			// BackoffLimit(set to 3), which with an exponential backoff (10s + 20s + 40s)= 1m10s
+			// BackoffLimit(set to 3), which for 4 failures is an exponential backoff (10s + 20s + 40s + 80s)= 2m30s
 			// so we wait a little over that.
 			Eventually(
 				func() (*operatorsv1alpha1.InstallPlan, error) {
 					err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(ip), ip)
 					return ip, err
 				},
-				2*time.Minute,
+				5*time.Minute,
 			).Should(HavePhase(operatorsv1alpha1.InstallPlanPhaseFailed))
 		})