Fix bundle upload tests

This commit introduces a new condition to indicate whether the bundle has been uploaded successfully by the uploadmgr. Once the upload is successful, the reconciler further starts unpacking the contents. Signed-off-by: Varsha Prasad Narsing <[email protected]>
operator-framework · Jan 12, 2024 · 3fae782 · 3fae782
1 parent 72423a6
commit 3fae782
Show file tree

Hide file tree

Showing 6 changed files with 116 additions and 71 deletions.
diff --git a/api/v1alpha2/bundledeployment_types.go b/api/v1alpha2/bundledeployment_types.go
@@ -30,6 +30,8 @@ const (
  TypeHasValidBundle = "HasValidBundle"
  TypeHealthy = "Healthy"
  TypeInstalled = "Installed"
+ // TypeUploadStatus indicates the status of the bundle content upload by the uploadmgr.
+ TypeUploadStatus = "UploadStatus"
 
  ReasonBundleLoadFailed = "BundleLoadFailed"
  ReasonCreateDynamicWatchFailed = "CreateDynamicWatchFailed"
@@ -45,6 +47,8 @@ const (
  ReasonReconcileFailed = "ReconcileFailed"
  ReasonUnhealthy = "Unhealthy"
  ReasonUpgradeFailed = "UpgradeFailed"
+ ReasonUploadSuccessful = "UploadSuccessful"
+ ReasonUploadFailed = "UploadFailed"
 )
 
 // BundleDeploymentSpec defines the desired state of BundleDeployment

diff --git a/internal/rukpakctl/portforward.go b/internal/rukpakctl/portforward.go
@@ -95,7 +95,6 @@ func (pf *ServicePortForwarder) Start(ctx context.Context) error {
  path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/portforward", pf.serviceNamespace, podName)
  host := strings.TrimLeft(pf.cfg.Host, "htps:/")
  serverURL := url.URL{Scheme: "https", Path: path, Host: host}
-
  roundTripper, upgrader, err := spdy.RoundTripperFor(pf.cfg)
  if err != nil {
  return err

diff --git a/internal/source/upload.go b/internal/source/upload.go
@@ -7,6 +7,8 @@ import (
  "net/http"
 
  "github.com/nlepage/go-tarfs"
+ "k8s.io/apimachinery/pkg/api/meta"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
  rukpakv1alpha2 "github.com/operator-framework/rukpak/api/v1alpha2"
 )
@@ -25,6 +27,12 @@ func (b *Upload) Unpack(ctx context.Context, bundle *rukpakv1alpha2.BundleDeploy
  return nil, fmt.Errorf("cannot unpack source type %q with %q unpacker", bundle.Spec.Source.Type, rukpakv1alpha2.SourceTypeUpload)
  }
 
+ // Proceed with fetching contents from a web server, only if the bundle upload was successful.
+ // If upload is a failure, we have "TypeUploadState" explicitly set to false.
+ if !isBundleContentUploaded(bundle) {
+ return &Result{State: StatePending, Message: "pending unpacking contents from uploaded bundle"}, nil
+ }
+
  url := fmt.Sprintf("%s/uploads/%s.tgz", b.baseDownloadURL, bundle.Name)
  action := fmt.Sprintf("%s %s", http.MethodGet, url)
 
@@ -57,3 +65,12 @@ func (b *Upload) Unpack(ctx context.Context, bundle *rukpakv1alpha2.BundleDeploy
 
  return &Result{Bundle: bundleFS, ResolvedSource: bundle.Spec.Source.DeepCopy(), State: StateUnpacked, Message: message}, nil
 }
+
+func isBundleContentUploaded(bd *rukpakv1alpha2.BundleDeployment) bool {
+ if bd == nil {
+ return false
+ }
+
+ condition := meta.FindStatusCondition(bd.Status.Conditions, rukpakv1alpha2.TypeUploadStatus)
+ return condition != nil && condition.Status == metav1.ConditionTrue
+}
diff --git a/internal/uploadmgr/handler.go b/internal/uploadmgr/handler.go
@@ -14,6 +14,7 @@ import (
  "k8s.io/apimachinery/pkg/api/meta"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/types"
+ utilerrors "k8s.io/apimachinery/pkg/util/errors"
  "k8s.io/client-go/util/retry"
  "sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -95,9 +96,33 @@ func newPutHandler(cl client.Client, storageDir string) http.Handler {
  })
  return cl.Status().Update(r.Context(), bundledeployment)
  }); err != nil {
- http.Error(w, err.Error(), int(getCode(err)))
+ errs := []error{}
+ errs = append(errs, err)
+
+ meta.SetStatusCondition(&bundledeployment.Status.Conditions, metav1.Condition{
+ Type: rukpakv1alpha2.TypeUploadStatus,
+ Status: metav1.ConditionFalse,
+ Reason: rukpakv1alpha2.ReasonBundleLoadFailed,
+ Message: err.Error(),
+ })
+ if statusUpdateErr := cl.Status().Update(r.Context(), bundledeployment); statusUpdateErr != nil {
+ errs = append(errs, statusUpdateErr)
+ }
+ http.Error(w, utilerrors.NewAggregate(errs).Error(), int(getCode(err)))
  return
  }
+ meta.SetStatusCondition(&bundledeployment.Status.Conditions, metav1.Condition{
+ Type: rukpakv1alpha2.TypeUploadStatus,
+ Status: metav1.ConditionTrue,
+ Reason: rukpakv1alpha2.ReasonUploadSuccessful,
+ Message: "successfully uploaded bundle contents.",
+ })
+ if statusUpdateErr := cl.Status().Update(r.Context(), bundledeployment); statusUpdateErr != nil {
+ // Though this would not be the http error returned from uploading, it
+ // is required to error, as BundleDeployment reconciler is waiting for
+ // was a successful upload status.
+ http.Error(w, statusUpdateErr.Error(), int(getCode(statusUpdateErr)))
+ }
  w.WriteHeader(http.StatusCreated)
  })
 }
@@ -108,7 +133,7 @@ func isBundleDeploymentUnpacked(bd *rukpakv1alpha2.BundleDeployment) bool {
  }
 
  condition := meta.FindStatusCondition(bd.Status.Conditions, rukpakv1alpha2.TypeUnpacked)
- return condition.Status == metav1.ConditionTrue
+ return condition != nil && condition.Status == metav1.ConditionTrue
 }
 
 func getCode(err error) int32 {

diff --git a/test/e2e/crdvalidator_test.go b/test/e2e/crdvalidator_test.go
@@ -28,7 +28,7 @@ var _ = Describe("crdvalidator", func() {
  crd = testutil.NewTestingCRD("", testutil.DefaultGroup,
  []apiextensionsv1.CustomResourceDefinitionVersion{
  {
- Name: "v1alpha2",
+ Name: "v1alpha1",
  Served: true,
  Storage: true,
  Schema: &apiextensionsv1.CustomResourceValidation{
@@ -108,7 +108,7 @@ var _ = Describe("crdvalidator", func() {
  BeforeEach(func() {
  crd = testutil.NewTestingCRD("", testutil.DefaultGroup,
  []apiextensionsv1.CustomResourceDefinitionVersion{{
- Name: "v1alpha2",
+ Name: "v1alpha1",
  Served: true,
  Storage: true,
  Schema: &apiextensionsv1.CustomResourceValidation{
@@ -130,7 +130,7 @@ var _ = Describe("crdvalidator", func() {
  }).Should(Succeed(), "should be able to create a safe crd but was not")
 
  // Build up a CR to create out of unstructured.Unstructured
- sampleCR := testutil.NewTestingCR(testutil.DefaultCrName, testutil.DefaultGroup, "v1alpha2", crd.Spec.Names.Singular)
+ sampleCR := testutil.NewTestingCR(testutil.DefaultCrName, testutil.DefaultGroup, "v1alpha1", crd.Spec.Names.Singular)
  Eventually(func() error {
  return c.Create(ctx, sampleCR)
  }).Should(Succeed(), "should be able to create a cr for the sample crd but was not")
@@ -148,7 +148,7 @@ var _ = Describe("crdvalidator", func() {
  return err.Error()
  }
 
- // Update the v1alpha2 schema to invalidate existing CR created in BeforeEach()
+ // Update the v1alpha1 schema to invalidate existing CR created in BeforeEach()
  crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Required = []string{"sampleProperty"}
 
  err := c.Update(ctx, crd)
@@ -169,7 +169,7 @@ var _ = Describe("crdvalidator", func() {
  crd.Labels = map[string]string{}
  Expect(c.Update(ctx, crd)).To(Succeed())
 
- // Update the v1alpha2 schema to invalidate existing CR created in BeforeEach()
+ // Update the v1alpha1 schema to invalidate existing CR created in BeforeEach()
  crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Required = []string{"sampleProperty"}
 
  return c.Update(ctx, crd)
@@ -186,7 +186,7 @@ var _ = Describe("crdvalidator", func() {
  crd.Annotations = map[string]string{annotation.ValidationKey: annotation.Disabled}
  Expect(c.Update(ctx, crd)).To(Succeed())
 
- // Update the v1alpha2 schema to invalidate existing CR created in BeforeEach()
+ // Update the v1alpha1 schema to invalidate existing CR created in BeforeEach()
  crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Required = []string{"sampleProperty"}
 
  return c.Update(ctx, crd)
@@ -210,7 +210,7 @@ var _ = Describe("crdvalidator", func() {
  crd = testutil.NewTestingCRD("", testutil.DefaultGroup,
  []apiextensionsv1.CustomResourceDefinitionVersion{
  {
- Name: "v1alpha2",
+ Name: "v1alpha1",
  Served: true,
  Storage: true,
  Schema: &apiextensionsv1.CustomResourceValidation{

diff --git a/test/e2e/plain_provisioner_test.go b/test/e2e/plain_provisioner_test.go
@@ -912,67 +912,6 @@ var _ = Describe("plain provisioner bundle", func() {
  })
  })
 
- When("the bundle deployment is uploaded", func() {
- var (
- bundledeployment *rukpakv1alpha2.BundleDeployment
- ctx context.Context
- )
-
- BeforeEach(func() {
- ctx = context.Background()
-
- bundleFS := os.DirFS(filepath.Join(testdataDir, "bundles/plain-v0/valid"))
- bundledeployment = &rukpakv1alpha2.BundleDeployment{
- ObjectMeta: metav1.ObjectMeta{
- Name: fmt.Sprintf("valid-upload-%s", rand.String(8)),
- },
- Spec: rukpakv1alpha2.BundleDeploymentSpec{
- ProvisionerClassName: plain.ProvisionerID,
- Source: rukpakv1alpha2.BundleSource{
- Type: rukpakv1alpha2.SourceTypeUpload,
- Upload: &rukpakv1alpha2.UploadSource{},
- },
- },
- }
- err := c.Create(ctx, bundledeployment)
- Expect(err).ToNot(HaveOccurred())
-
- rootCAs, err := rukpakctl.GetClusterCA(ctx, c, types.NamespacedName{Namespace: defaultSystemNamespace, Name: "rukpak-ca"})
- Expect(err).ToNot(HaveOccurred())
-
- bu := rukpakctl.BundleUploader{
- UploadServiceName: defaultUploadServiceName,
- UploadServiceNamespace: defaultSystemNamespace,
- Cfg: cfg,
- RootCAs: rootCAs,
- }
- uploadCtx, cancel := context.WithTimeout(ctx, time.Second*5)
- defer cancel()
- _, err = bu.Upload(uploadCtx, bundledeployment.Name, bundleFS)
- Expect(err).ToNot(HaveOccurred())
- })
-
- AfterEach(func() {
- err := c.Delete(ctx, bundledeployment)
- Expect(client.IgnoreNotFound(err)).To(Succeed())
- })
-
- It("can unpack the bundle successfully", func() {
- Eventually(func() (*metav1.Condition, error) {
- if err := c.Get(ctx, client.ObjectKeyFromObject(bundledeployment), bundledeployment); err != nil {
- return nil, err
- }
- return meta.FindStatusCondition(bundledeployment.Status.Conditions, rukpakv1alpha2.TypeInstalled), nil
- }).Should(And(
- Not(BeNil()),
- WithTransform(func(c *metav1.Condition) string { return c.Type }, Equal(rukpakv1alpha2.TypeInstalled)),
- WithTransform(func(c *metav1.Condition) metav1.ConditionStatus { return c.Status }, Equal(metav1.ConditionTrue)),
- WithTransform(func(c *metav1.Condition) string { return c.Reason }, Equal(rukpakv1alpha2.ReasonInstallationSucceeded)),
- WithTransform(func(c *metav1.Condition) string { return c.Message }, ContainSubstring("Instantiated bundle")),
- ))
- })
- })
-
  When("the bundle is backed by an invalid configmap", func() {
  var (
  bundledeployment *rukpakv1alpha2.BundleDeployment
@@ -1050,6 +989,67 @@ var _ = Describe("plain provisioner bundle", func() {
  })
  })
 
+ When("the bundle deployment is uploaded", func() {
+ var (
+ bundledeployment *rukpakv1alpha2.BundleDeployment
+ ctx context.Context
+ )
+
+ BeforeEach(func() {
+ ctx = context.Background()
+
+ bundleFS := os.DirFS(filepath.Join(testdataDir, "bundles/plain-v0/valid"))
+ bundledeployment = &rukpakv1alpha2.BundleDeployment{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: fmt.Sprintf("valid-upload-%s", rand.String(8)),
+ },
+ Spec: rukpakv1alpha2.BundleDeploymentSpec{
+ ProvisionerClassName: plain.ProvisionerID,
+ Source: rukpakv1alpha2.BundleSource{
+ Type: rukpakv1alpha2.SourceTypeUpload,
+ Upload: &rukpakv1alpha2.UploadSource{},
+ },
+ },
+ }
+ err := c.Create(ctx, bundledeployment)
+ Expect(err).ToNot(HaveOccurred())
+
+ rootCAs, err := rukpakctl.GetClusterCA(ctx, c, types.NamespacedName{Namespace: defaultSystemNamespace, Name: "rukpak-ca"})
+ Expect(err).ToNot(HaveOccurred())
+
+ bu := rukpakctl.BundleUploader{
+ UploadServiceName: defaultUploadServiceName,
+ UploadServiceNamespace: defaultSystemNamespace,
+ Cfg: cfg,
+ RootCAs: rootCAs,
+ }
+ uploadCtx, cancel := context.WithTimeout(ctx, time.Second*5)
+ defer cancel()
+ _, err = bu.Upload(uploadCtx, bundledeployment.Name, bundleFS)
+ Expect(err).ToNot(HaveOccurred())
+ })
+
+ AfterEach(func() {
+ err := c.Delete(ctx, bundledeployment)
+ Expect(client.IgnoreNotFound(err)).To(Succeed())
+ })
+
+ It("can unpack the bundle successfully", func() {
+ Eventually(func() (*metav1.Condition, error) {
+ if err := c.Get(ctx, client.ObjectKeyFromObject(bundledeployment), bundledeployment); err != nil {
+ return nil, err
+ }
+ return meta.FindStatusCondition(bundledeployment.Status.Conditions, rukpakv1alpha2.TypeInstalled), nil
+ }).Should(And(
+ Not(BeNil()),
+ WithTransform(func(c *metav1.Condition) string { return c.Type }, Equal(rukpakv1alpha2.TypeInstalled)),
+ WithTransform(func(c *metav1.Condition) metav1.ConditionStatus { return c.Status }, Equal(metav1.ConditionTrue)),
+ WithTransform(func(c *metav1.Condition) string { return c.Reason }, Equal(rukpakv1alpha2.ReasonInstallationSucceeded)),
+ WithTransform(func(c *metav1.Condition) string { return c.Message }, ContainSubstring("Instantiated bundle")),
+ ))
+ })
+ })
+
  When("the bundle deployment is backed by an invalid upload", func() {
  var (
  bundledeployment *rukpakv1alpha2.BundleDeployment