unpack retry e2e tests

Signed-off-by: Ankita Thomas <[email protected]>

Author: ankitathomas

Diff for: pkg/controller/bundle/bundle_unpacker.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8slabels "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apiserver/pkg/storage/names"
 	"k8s.io/client-go/kubernetes"
 	listersbatchv1 "k8s.io/client-go/listers/batch/v1"
 	listerscorev1 "k8s.io/client-go/listers/core/v1"
@@ -695,10 +696,10 @@ func (c *ConfigMapUnpacker) ensureJob(cmRef *corev1.ObjectReference, bundlePath
 
 			if failed {
 				if time.Now().After(lastFailureTime.Add(unpackRetryInterval)) {
-					fresh.SetName(fmt.Sprintf("%s-%d", fresh.GetName(), len(jobs)))
+					fresh.SetName(names.SimpleNameGenerator.GenerateName(fresh.GetName()))
 					job, err = c.client.BatchV1().Jobs(fresh.GetNamespace()).Create(context.TODO(), fresh, metav1.CreateOptions{})
-					return
 				}
+				return
 			}
 		}
 	}
@@ -890,7 +891,7 @@ func OperatorGroupBundleUnpackRetryInterval(ogLister v1listers.OperatorGroupName
 
 	d, err := time.ParseDuration(timeoutStr)
 	if err != nil {
-		return 0, fmt.Errorf("failed to parse unpack timeout annotation(%s: %s): %w", BundleUnpackRetryMinimumIntervalAnnotationKey, timeoutStr, err)
+		return 0, fmt.Errorf("failed to parse unpack retry annotation(%s: %s): %w", BundleUnpackRetryMinimumIntervalAnnotationKey, timeoutStr, err)
 	}
 
 	return d, nil

Diff for: pkg/controller/bundle/bundle_unpacker_test.go

@@ -1839,3 +1839,115 @@ func TestOperatorGroupBundleUnpackTimeout(t *testing.T) {
 		})
 	}
 }
+
+func TestOperatorGroupBundleUnpackRetryInterval(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: 0,
+			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: 0,
+			expectedError:   errors.New("found 2 operatorGroups, expected 1"),
+		},
+		{
+			name: "One operator group exists with valid unpack retry 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{BundleUnpackRetryMinimumIntervalAnnotationKey: "1m"},
+					},
+				},
+			},
+			expectedTimeout: 1 * time.Minute,
+			expectedError:   nil,
+		},
+		{
+			name: "One operator group exists with no unpack retry annotation",
+			operatorGroups: []*operatorsv1.OperatorGroup{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       operatorsv1.OperatorGroupKind,
+						APIVersion: operatorsv1.GroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "og",
+						Namespace: nsName,
+					},
+				},
+			},
+			expectedTimeout: 0,
+			expectedError:   nil,
+		},
+		{
+			name: "One operator group exists with invalid unpack retry 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{BundleUnpackRetryMinimumIntervalAnnotationKey: "invalid"},
+					},
+				},
+			},
+			expectedTimeout: 0,
+			expectedError:   fmt.Errorf("failed to parse unpack retry annotation(operatorframework.io/bundle-unpack-min-retry-interval: 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 := OperatorGroupBundleUnpackRetryInterval(ogLister)
+
+			assert.Equal(t, tc.expectedTimeout, timeout)
+			assert.Equal(t, tc.expectedError, err)
+		})
+	}
+}

Diff for: test/e2e/fbc_provider.go

@@ -31,3 +31,9 @@ func NewFileBasedFiledBasedCatalogProvider(path string) (FileBasedCatalogProvide
 func (f *fileBasedFileBasedCatalogProvider) GetCatalog() string {
 	return f.fbc
 }
+
+func NewRawFileBasedCatalogProvider(data string) (FileBasedCatalogProvider, error) {
+	return &fileBasedFileBasedCatalogProvider{
+		fbc: string(data),
+	}, nil
+}

Diff for: test/e2e/subscription_e2e_test.go

@@ -2524,6 +2524,185 @@ var _ = Describe("Subscription", func() {
 			})
 		})
 	})
+	When("bundle unpack retries are enabled", func() {
+		FIt("should retry failing unpack jobs", func() {
+			By("Ensuring a registry to host bundle images")
+			local, err := Local(c)
+			Expect(err).NotTo(HaveOccurred(), "cannot determine if test running locally or on CI: %s", err)
+
+			var registryURL string
+			var copyImage func(dst, dstTag, src, srcTag string) error
+			if local {
+				registryURL, err = createDockerRegistry(c, generatedNamespace.GetName())
+				Expect(err).NotTo(HaveOccurred(), "error creating container registry: %s", err)
+				defer deleteDockerRegistry(c, generatedNamespace.GetName())
+
+				// ensure registry pod is ready before attempting port-forwarding
+				_ = awaitPod(GinkgoT(), c, generatedNamespace.GetName(), registryName, podReady)
+
+				err = registryPortForward(generatedNamespace.GetName())
+				Expect(err).NotTo(HaveOccurred(), "port-forwarding local registry: %s", err)
+				copyImage = func(dst, dstTag, src, srcTag string) error {
+					if !strings.HasPrefix(src, "docker://") {
+						src = fmt.Sprintf("docker://%s", src)
+					}
+					if !strings.HasPrefix(dst, "docker://") {
+						dst = fmt.Sprintf("docker://%s", dst)
+					}
+					_, err := skopeoLocalCopy(dst, dstTag, src, srcTag)
+					return err
+				}
+			} else {
+				registryURL = fmt.Sprintf("%s/%s", openshiftregistryFQDN, generatedNamespace.GetName())
+				registryAuth, err := openshiftRegistryAuth(c, generatedNamespace.GetName())
+				Expect(err).NotTo(HaveOccurred(), "error getting openshift registry authentication: %s", err)
+				copyImage = func(dst, dstTag, src, srcTag string) error {
+					if !strings.HasPrefix(src, "docker://") {
+						src = fmt.Sprintf("docker://%s", src)
+					}
+					if !strings.HasPrefix(dst, "docker://") {
+						dst = fmt.Sprintf("docker://%s", dst)
+					}
+					skopeoArgs := skopeoCopyCmd(dst, dstTag, src, srcTag, registryAuth)
+					err = createSkopeoPod(c, skopeoArgs, generatedNamespace.GetName())
+					if err != nil {
+						return fmt.Errorf("error creating skopeo pod: %v", err)
+					}
+
+					// wait for skopeo pod to exit successfully
+					awaitPod(GinkgoT(), c, generatedNamespace.GetName(), skopeo, func(pod *corev1.Pod) bool {
+						return pod.Status.Phase == corev1.PodSucceeded
+					})
+
+					if err := deleteSkopeoPod(c, generatedNamespace.GetName()); err != nil {
+						return fmt.Errorf("error deleting skopeo pod: %s", err)
+					}
+					return nil
+				}
+			}
+
+			// testImage is the name of the image used throughout the test - the image overwritten by skopeo
+			// the tag is generated randomly and appended to the end of the testImage
+			srcImage := "quay.io/olmtest/example-operator-bundle:"
+			srcTag := "0.1.0"
+			bundleImage := fmt.Sprint(registryURL, "/unpack-retry-bundle", ":")
+			bundleTag := genName("x")
+			//// hash hashes data with sha256 and returns the hex string.
+			//func hash(data string) string {
+			//	// A SHA256 hash is 64 characters, which is within the 253 character limit for kube resource names
+			//	h := fmt.Sprintf("%x", sha256.Sum256([]byte(data)))
+			//
+			//	// Make the hash 63 characters instead to comply with the 63 character limit for labels
+			//	return fmt.Sprintf(h[:len(h)-1])
+			//}
+			unpackRetryCatalog := fmt.Sprintf(`
+schema: olm.package
+name: unpack-retry-package
+defaultChannel: stable
+---
+schema: olm.channel
+package: unpack-retry-package
+name: stable
+entries:
+  - name: unpack-retry-operator.v2.0.0
+    skipRange: ">=0.1.0 <2.0.0"
+---
+schema: olm.bundle
+name: unpack-retry-operator.v1.0.0
+package: unpack-retry-package
+image: %s%s
+properties:
+  - type: olm.package
+    value:
+      packageName: unpack-retry-package
+      version: 1.0.0
+`, bundleImage, bundleTag)
+
+			By("creating a catalog referencing a non-existent bundle image")
+			unpackRetryProvider, err := NewRawFileBasedCatalogProvider(unpackRetryCatalog)
+			Expect(err).ToNot(HaveOccurred())
+			catalogSourceName := fmt.Sprintf("%s-catsrc", generatedNamespace.GetName())
+			magicCatalog := NewMagicCatalog(ctx.Ctx().Client(), generatedNamespace.GetName(), catalogSourceName, unpackRetryProvider)
+			Expect(magicCatalog.DeployCatalog(context.Background())).To(BeNil())
+
+			By("patching the OperatorGroup to reduce the bundle unpacking timeout")
+			ogNN := types.NamespacedName{Name: operatorGroup.GetName(), Namespace: generatedNamespace.GetName()}
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "1s")
+
+			By("creating a subscription for the missing bundle")
+			unpackRetrySubName := fmt.Sprintf("%s-unpack-retry-package-sub", generatedNamespace.GetName())
+			createSubscriptionForCatalog(crc, generatedNamespace.GetName(), unpackRetrySubName, catalogSourceName, "unpack-retry-package", stableChannel, "", operatorsv1alpha1.ApprovalAutomatic)
+
+			By("waiting for bundle unpack to fail")
+			Eventually(
+				func() error {
+					fetched, err := crc.OperatorsV1alpha1().Subscriptions(generatedNamespace.GetName()).Get(context.Background(), unpackRetrySubName, metav1.GetOptions{})
+					if err != nil {
+						return err
+					}
+					if cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpackFailed); cond.Status != corev1.ConditionTrue || cond.Reason != "BundleUnpackFailed" {
+						return fmt.Errorf("%s condition not found", v1alpha1.SubscriptionBundleUnpackFailed)
+					}
+					return nil
+				},
+				5*time.Minute,
+				interval,
+			).Should(Succeed())
+
+			By("pushing missing bundle image")
+			Expect(copyImage(bundleImage, bundleTag, srcImage, srcTag)).To(Succeed())
+
+			By("patching the OperatorGroup to increase the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "") // revert to default unpack timeout
+
+			By("patching operator group to enable unpack retries")
+			setBundleUnpackRetryMinimumIntervalAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "1s")
+
+			By("waiting for checking for installPlan to indicate unpack succeeds")
+			Expect(fetchSubscription(crc, magicCatalog.GetNamespace(), unpackRetrySubName, subscriptionHasInstallPlanChecker)).To(Succeed())
+			time.Sleep(20 * time.Minute)
+		})
+
+		It("should not retry successful unpack jobs", func() {
+			By("deploying the testing catalog")
+			provider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.1.0.yaml"))
+			Expect(err).To(BeNil())
+			catalogSourceName := fmt.Sprintf("%s-catsrc", generatedNamespace.GetName())
+			magicCatalog := NewMagicCatalog(ctx.Ctx().Client(), generatedNamespace.GetName(), catalogSourceName, provider)
+			Expect(magicCatalog.DeployCatalog(context.Background())).To(BeNil())
+
+			By("creating the testing subscription")
+			subName := fmt.Sprintf("%s-packagea-sub", generatedNamespace.GetName())
+			createSubscriptionForCatalog(crc, generatedNamespace.GetName(), subName, catalogSourceName, "packageA", stableChannel, "", operatorsv1alpha1.ApprovalAutomatic)
+
+			By("waiting until the subscription has an IP reference")
+			subscription, err := fetchSubscription(crc, generatedNamespace.GetName(), subName, subscriptionHasInstallPlanChecker)
+			Expect(err).Should(BeNil())
+
+			By("waiting for the v0.1.0 CSV to report a succeeded phase")
+			_, err = fetchCSV(crc, subscription.Status.CurrentCSV, generatedNamespace.GetName(), buildCSVConditionChecker(operatorsv1alpha1.CSVPhaseSucceeded))
+			Expect(err).ShouldNot(HaveOccurred())
+
+			By("patching operator group to enable unpack retries")
+			ogNN := types.NamespacedName{Name: operatorGroup.GetName(), Namespace: generatedNamespace.GetName()}
+			setBundleUnpackRetryMinimumIntervalAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "1s")
+
+			By("Ensuring successful bundle unpack jobs are not retried")
+			Consistently(func() error {
+				fetched, err := crc.OperatorsV1alpha1().Subscriptions(generatedNamespace.GetName()).Get(context.Background(), subName, metav1.GetOptions{})
+				if err != nil {
+					return err
+				}
+				if cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpacking); cond.Status == corev1.ConditionTrue {
+					return fmt.Errorf("unexpected condition status for %s on subscription %s", v1alpha1.SubscriptionBundleUnpacking, subName)
+				}
+				if cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpackFailed); cond.Status == corev1.ConditionTrue {
+					return fmt.Errorf("unexpected condition status for %s on subscription %s", v1alpha1.SubscriptionBundleUnpackFailed, subName)
+				}
+				return nil
+			}).Should(Succeed())
+		})
+	})
 })
 
 const (

Diff for: test/e2e/util.go

@@ -94,13 +94,25 @@ func objectRefToNamespacedName(ip *corev1.ObjectReference) types.NamespacedName
 // adding the "operatorframework.io/bundle-unpack-timeout" annotation to an OperatorGroup
 // resource.
 func addBundleUnpackTimeoutOGAnnotation(ctx context.Context, c k8scontrollerclient.Client, ogNN types.NamespacedName, timeout string) {
+	setOGAnnotation(ctx, c, ogNN, bundle.BundleUnpackTimeoutAnnotationKey, timeout)
+}
+
+func setBundleUnpackRetryMinimumIntervalAnnotation(ctx context.Context, c k8scontrollerclient.Client, ogNN types.NamespacedName, interval string) {
+	setOGAnnotation(ctx, c, ogNN, bundle.BundleUnpackRetryMinimumIntervalAnnotationKey, interval)
+}
+
+func setOGAnnotation(ctx context.Context, c k8scontrollerclient.Client, ogNN types.NamespacedName, key, value string) {
 	Eventually(func() error {
 		og := &operatorsv1.OperatorGroup{}
 		if err := c.Get(ctx, ogNN, og); err != nil {
 			return err
 		}
 		annotations := og.GetAnnotations()
-		annotations[bundle.BundleUnpackTimeoutAnnotationKey] = timeout
+		if len(value) == 0 {
+			delete(annotations, key)
+		} else {
+			annotations[key] = value
+		}
 		og.SetAnnotations(annotations)
 
 		return c.Update(ctx, og)