Moves bundle unpack timeout into OperatorGroup

pkg/controller/bundle/bundle_unpacker.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/operator-framework/api/pkg/operators/reference"
 	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
+	v1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1"
 	listersoperatorsv1alpha1 "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/projection"
@@ -771,3 +772,30 @@ func getCondition(job *batchv1.Job, conditionType batchv1.JobConditionType) (con
 	}
 	return
 }
+
+// OperatorGroupBundleUnpackTimeout returns bundle timeout from annotation if specified.
+// If the timeout annotation is not set, return timeout < 0 which is subsequently ignored.
+// This is to overrides the --bundle-unpack-timeout flag value on per-OperatorGroup basis.
+func OperatorGroupBundleUnpackTimeout(ogLister v1listers.OperatorGroupNamespaceLister) (time.Duration, error) {
+	ignoreTimeout := -1 * time.Minute
+
+	ogs, err := ogLister.List(k8slabels.Everything())
+	if err != nil {
+		return ignoreTimeout, err
+	}
+	if len(ogs) != 1 {
+		return ignoreTimeout, fmt.Errorf("found %d operatorGroups, expected 1", len(ogs))
+	}
+
+	timeoutStr, ok := ogs[0].GetAnnotations()[BundleUnpackTimeoutAnnotationKey]
+	if !ok {
+		return ignoreTimeout, nil
+	}
+
+	d, err := time.ParseDuration(timeoutStr)
+	if err != nil {
+		return ignoreTimeout, fmt.Errorf("failed to parse unpack timeout annotation(%s: %s): %w", BundleUnpackTimeoutAnnotationKey, timeoutStr, err)
+	}
+
+	return d, nil
+}

pkg/controller/bundle/bundle_unpacker_test.go

@@ -2,10 +2,12 @@ package bundle
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -15,11 +17,14 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/informers"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/pointer"
 
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
 	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	crfake "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned/fake"
 	crinformers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
+	v1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install"
 	"github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/configmap"
@@ -1572,3 +1577,114 @@ func TestConfigMapUnpacker(t *testing.T) {
 		})
 	}
 }
+
+func TestOperatorGroupBundleUnpackTimeout(t *testing.T) {
+	nsName := "fake-ns"
+
+	for _, tc := range []struct {
+		name            string
+		operatorGroups  []*operatorsv1.OperatorGroup
+		expectedTimeout time.Duration
+		expectedError   error
+	}{
+		{
+			name:            "No operator groups exist",
+			expectedTimeout: -1 * time.Minute,
+			expectedError:   errors.New("found 0 operatorGroups, expected 1"),
+		},
+		{
+			name: "Multiple operator groups exist",
+			operatorGroups: []*operatorsv1.OperatorGroup{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "og1",
+						Namespace: nsName,
+					},
+				},
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "og2",
+						Namespace: nsName,
+					},
+				},
+			},
+			expectedTimeout: -1 * time.Minute,
+			expectedError:   errors.New("found 2 operatorGroups, expected 1"),
+		},
+		{
+			name: "One operator group exists with valid timeout annotation",
+			operatorGroups: []*operatorsv1.OperatorGroup{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:        "og",
+						Namespace:   nsName,
+						Annotations: map[string]string{BundleUnpackTimeoutAnnotationKey: "1m"},
+					},
+				},
+			},
+			expectedTimeout: 1 * time.Minute,
+			expectedError:   nil,
+		},
+		{
+			name: "One operator group exists with no timeout annotation",
+			operatorGroups: []*operatorsv1.OperatorGroup{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "og",
+						Namespace: nsName,
+					},
+				},
+			},
+			expectedTimeout: -1 * time.Minute,
+		},
+		{
+			name: "One operator group exists with invalid timeout annotation",
+			operatorGroups: []*operatorsv1.OperatorGroup{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:        "og",
+						Namespace:   nsName,
+						Annotations: map[string]string{BundleUnpackTimeoutAnnotationKey: "invalid"},
+					},
+				},
+			},
+			expectedTimeout: -1 * time.Minute,
+			expectedError:   fmt.Errorf("failed to parse unpack timeout annotation(operatorframework.io/bundle-unpack-timeout: invalid): %w", errors.New("time: invalid duration \"invalid\"")),
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ogIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+			ogLister := v1listers.NewOperatorGroupLister(ogIndexer).OperatorGroups(nsName)
+
+			for _, og := range tc.operatorGroups {
+				err := ogIndexer.Add(og)
+				assert.NoError(t, err)
+			}
+
+			timeout, err := OperatorGroupBundleUnpackTimeout(ogLister)
+
+			assert.Equal(t, tc.expectedTimeout, timeout)
+			assert.Equal(t, tc.expectedError, err)
+		})
+	}
+}

pkg/controller/operators/catalog/operator.go

@@ -1408,23 +1408,10 @@ type UnpackedBundleReference struct {
 	Properties             string `json:"properties"`
 }
 
-func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.InstallPlan, error) {
+func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan, unpackTimeout time.Duration) (bool, *v1alpha1.InstallPlan, error) {
 	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]
@@ -1599,7 +1586,6 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 
 	querier := o.serviceAccountQuerier.NamespaceQuerier(plan.GetNamespace())
 	ref, err := querier()
-
 	out := plan.DeepCopy()
 	if err != nil {
 		// Set status condition/message and retry sync if any error
@@ -1645,10 +1631,16 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 		}
 	}
 
+	ogLister := o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(plan.GetNamespace())
+	unpackTimeout, err := bundle.OperatorGroupBundleUnpackTimeout(ogLister)
+	if err != nil {
+		return err
+	}
+
 	// Attempt to unpack bundles before installing
 	// Note: This should probably use the attenuated client to prevent users from resolving resources they otherwise don't have access to.
 	if len(plan.Status.BundleLookups) > 0 {
-		unpacked, out, err := o.unpackBundles(plan)
+		unpacked, out, err := o.unpackBundles(plan, unpackTimeout)
 		if err != nil {
 			// If the error was fatal capture and fail
 			if fatal := olmerrors.IsFatal(err); fatal {

test/e2e/fail_forward_e2e_test.go

@@ -9,6 +9,7 @@ import (
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
@@ -27,6 +28,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 		ns       corev1.Namespace
 		crclient versioned.Interface
 		c        client.Client
+		ogName   string
 	)
 
 	BeforeEach(func() {
@@ -45,6 +47,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			},
 		}
 		ns = SetupGeneratedTestNamespaceWithOperatorGroup(namespaceName, og)
+		ogName = og.GetName()
 	})
 
 	AfterEach(func() {
@@ -61,9 +64,9 @@ var _ = Describe("Fail Forward Upgrades", func() {
 		)
 
 		BeforeEach(func() {
+			By("deploying the testing catalog")
 			provider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.1.0.yaml"))
 			Expect(err).To(BeNil())
-
 			catalogSourceName = genName("mc-ip-failed-")
 			magicCatalog = NewMagicCatalog(c, ns.GetName(), catalogSourceName, provider)
 			Expect(magicCatalog.DeployCatalog(context.Background())).To(BeNil())
@@ -93,6 +96,9 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			_, err = fetchCSV(crclient, subscription.Status.CurrentCSV, ns.GetName(), buildCSVConditionChecker(operatorsv1alpha1.CSVPhaseSucceeded))
 			Expect(err).ShouldNot(HaveOccurred())
 
+			By("patching the OperatorGroup to reduce the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "1s")
+
 			By("updating the catalog with a broken v0.2.0 bundle image")
 			brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0.yaml"))
 			Expect(err).To(BeNil())
@@ -104,9 +110,6 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanDifferentChecker(originalInstallPlanRef.Name))
 			Expect(err).Should(BeNil())
 
-			By("patching the installplan to reduce the bundle unpacking timeout")
-			addBundleUnpackTimeoutIPAnnotation(context.Background(), c, objectRefToNamespacedName(subscription.Status.InstallPlanRef), "1s")
-
 			By("waiting for the bad InstallPlan to report a failed installation state")
 			ref := subscription.Status.InstallPlanRef
 			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
@@ -129,18 +132,17 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.2.1"))
 			Expect(err).Should(BeNil())
 
-			By("patching the installplan to reduce the bundle unpacking timeout")
-			addBundleUnpackTimeoutIPAnnotation(context.Background(), c, objectRefToNamespacedName(subscription.Status.InstallPlanRef), "1s")
-
 			By("waiting for the bad v0.2.1 InstallPlan to report a failed installation state")
 			ref := subscription.Status.InstallPlanRef
 			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
 			Expect(err).To(BeNil())
 
+			By("patching the OperatorGroup to increase the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
+
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/multiple-bad-versions", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
-
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
@@ -150,10 +152,12 @@ var _ = Describe("Fail Forward Upgrades", func() {
 		})
 
 		It("eventually reports a successful state when using skip ranges", func() {
+			By("patching the OperatorGroup to increase the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
+
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/skip-range", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
-
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
@@ -162,10 +166,12 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			Expect(err).Should(BeNil())
 		})
 		It("eventually reports a successful state when using skips", func() {
+			By("patching the OperatorGroup to increase the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
+
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/skips", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
-
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
@@ -174,10 +180,12 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			Expect(err).Should(BeNil())
 		})
 		It("eventually reports a failed state when using replaces", func() {
+			By("patching the OperatorGroup to increase the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
+
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/replaces", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
-
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
@@ -200,9 +208,9 @@ var _ = Describe("Fail Forward Upgrades", func() {
 		)
 
 		BeforeEach(func() {
+			By("deploying the testing catalog")
 			provider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.1.0.yaml"))
 			Expect(err).To(BeNil())
-
 			catalogSourceName = genName("mc-csv-failed-")
 			magicCatalog = NewMagicCatalog(c, ns.GetName(), catalogSourceName, provider)
 			Expect(magicCatalog.DeployCatalog(context.Background())).To(BeNil())