[validate] Add logic for validation and test cases

This commit adds:
1. Logic for validation of helm, registryV1 and plain bundles.
2. Tests for image source and a mock store implementation for use.

Signed-off-by: Varsha Prasad Narsing <[email protected]>

api/v1alpha2/bundledeployment_types.go

@@ -25,6 +25,15 @@ var (
 	BundleDeploymentKind = BundleDeploymentGVK.Kind
 )
 
+const (
+	TypeUnpacked = "Unpacked"
+
+	ReasonUnpackFailed     = "UnpackFailed"
+	ReasonUnpackPending    = "UnpackPending"
+	ReasonUnpacking        = "Unpacking"
+	ReasonUnpackSuccessful = "UnpackSuccessful"
+)
+
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:resource:scope=Cluster,shortName={"bd","bds"}
@@ -111,3 +120,7 @@ type BundleDeploymentList struct {
 
 	Items []BundleDeployment `json:"items"`
 }
+
+func init() {
+	SchemeBuilder.Register(&BundleDeployment{}, &BundleDeploymentList{})
+}

internal/v1alpha2/controllers/bundledeployment/bundledeployment.go

@@ -18,9 +18,12 @@ package bundledeployment
 
 import (
 	"context"
+	"fmt"
 	"sync"
+	"time"
 
 	"github.com/operator-framework/rukpak/api/v1alpha2"
+	source "github.com/operator-framework/rukpak/internal/v1alpha2/source"
 	"github.com/operator-framework/rukpak/internal/v1alpha2/store"
 	"github.com/spf13/afero"
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -46,11 +49,26 @@ type bundleDeploymentReconciler struct {
 	Scheme   *runtime.Scheme
 	Recorder record.EventRecorder
 
-	controller        crcontroller.Controller
+	controller crcontroller.Controller
+
+	// unpacker knows how to unpack and store the contents locally
+	// on filesystem for the bundle types which would be handled
+	// by this reconciler.
+	unpacker source.Unpacker
+
 	dynamicWatchMutex sync.RWMutex
 	dynamicWatchGVKs  map[schema.GroupVersionKind]struct{}
 }
 
+// Options to configure bundleDeploymentReconciler
+type Option func(bd *bundleDeploymentReconciler)
+
+func WithUnpacker(u *source.Unpacker) Option {
+	return func(bd *bundleDeploymentReconciler) {
+		bd.unpacker = *u
+	}
+}
+
 //+kubebuilder:rbac:groups=core.rukpak.io,resources=bundledeployments,verbs=list;watch
 //+kubebuilder:rbac:groups=core.rukpak.io,resources=bundledeployments/status,verbs=update;patch
 //+kubebuilder:rbac:groups=core.rukpak.io,resources=bundledeployments/finalizers,verbs=update
@@ -64,7 +82,7 @@ func (b *bundleDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	log := log.FromContext(ctx)
 
 	existingBD := &v1alpha2.BundleDeployment{}
-	err := b.Get(ctx, req.NamespacedName, existingBD)
+	err := b.Client.Get(ctx, req.NamespacedName, existingBD)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			log.Info("bundledeployment resource not found. Ignoring since object must be deleted.")
@@ -81,7 +99,16 @@ func (b *bundleDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	}
 
 	reconciledBD := existingBD.DeepCopy()
-	res, reconcileErr := b.reconcile(ctx, reconciledBD)
+
+	// Establish a filesystem view on the local system, similar to 'chroot', with the root directory
+	// determined by the bundle deployment name. All contents from various sources will be extracted and
+	// placed within this specified root directory.
+	bundledeploymentStore, err := store.NewBundleDeploymentStore(unpackpath, reconciledBD.GetName(), afero.NewOsFs())
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	res, reconcileErr := b.reconcile(ctx, reconciledBD, bundledeploymentStore)
 
 	// The controller is not updating spec, we only update the status. Hence sending
 	// a status update should be enough.
@@ -97,15 +124,57 @@ func (b *bundleDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req
 // It further validates if the unpacked content conforms to the format specified in the bundledeployment.
 // If the validation is successful, it further deploys the objects on cluster. If there is an error
 // encountered in this process, an appropriate result is returned.
-func (b *bundleDeploymentReconciler) reconcile(ctx context.Context, bundleDeployment *v1alpha2.BundleDeployment) (ctrl.Result, error) {
+func (b *bundleDeploymentReconciler) reconcile(ctx context.Context, bundleDeployment *v1alpha2.BundleDeployment, bundledeploymentStore store.Store) (ctrl.Result, error) {
 
-	// Establish a filesystem view on the local system, similar to 'chroot', with the root directory
-	// determined by the bundle deployment name. All contents from various sources will be extracted and
-	// placed within this specified root directory.
-	_, err := store.NewBundleDeploymentStore(unpackpath, bundleDeployment.GetName(), afero.NewOsFs())
-	if err != nil {
+	res, err := b.unpackContents(ctx, bundleDeployment, bundledeploymentStore)
+	// result can be nil, when there is an error during unpacking. This indicates
+	// that unpacking was failed. Update the status accordingly.
+	if res == nil || err != nil {
+		setUnpackStatusFailing(&bundleDeployment.Status.Conditions, fmt.Sprintf("unpack unsuccessful %v", err), bundleDeployment.Generation)
 		return ctrl.Result{}, err
 	}
 
+	switch res.State {
+	case source.StateUnpackPending:
+		setUnpackStatusPending(&bundleDeployment.Status.Conditions, fmt.Sprintf("pending unpack"), bundleDeployment.Generation)
+		// Requeing after 5 sec for now since the average time to unpack an registry bundle locally
+		// was around ~4sec.
+		// Warning: This could end up requeing indefinitely, till an error has occured.
+		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+	case source.StateUnpacking:
+		setUnpackStatusPending(&bundleDeployment.Status.Conditions, fmt.Sprintf("unpacking in progress"), bundleDeployment.Generation)
+	case source.StateUnpacked:
+		setUnpackStatusSuccessful(&bundleDeployment.Status.Conditions, fmt.Sprintf("unpacked successfully"), bundleDeployment.Generation)
+	default:
+		return ctrl.Result{}, fmt.Errorf("unkown unpack state %q for bundle deployment %s: %v", res.State, bundleDeployment.GetName(), bundleDeployment.Generation)
+	}
+
 	return ctrl.Result{}, nil
 }
+
+// unpackContents unpacks contents from all the sources, and stores under a directory referenced by the bundle deployment name.
+// It returns the consolidated state on whether contents from all the sources have been unpacked.
+func (b *bundleDeploymentReconciler) unpackContents(ctx context.Context, bundledeployment *v1alpha2.BundleDeployment, store store.Store) (*source.Result, error) {
+	unpackResult := make([]source.Result, 0)
+
+	// unpack each of the sources individually, and consolidate all their results into one.
+	for _, src := range bundledeployment.Spec.Sources {
+		res, err := b.unpacker.Unpack(ctx, &src, store, source.UnpackOption{
+			BundleDeploymentUID: bundledeployment.UID,
+		})
+		if err != nil {
+			return nil, err
+		}
+		unpackResult = append(unpackResult, *res)
+	}
+	// Even if one source has not unpacked, update Bundle Deployment status accordingly.
+	// In this case the status will contain the result from the first source
+	// which is still waiting to be unpacked.
+	for _, res := range unpackResult {
+		if res.State != source.StateUnpacked {
+			return &res, nil
+		}
+	}
+	// TODO: capture the list of resolved sources for all the successful entry points.
+	return &source.Result{State: source.StateUnpacked, Message: "Successfully unpacked"}, nil
+}

internal/v1alpha2/controllers/bundledeployment/bundledeployment_status.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2023.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bundledeployment
+
+import (
+	"github.com/operator-framework/rukpak/api/v1alpha2"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// setUnpackStatusFailing sets the unpack status condition to failing.
+func setUnpackStatusFailing(conditions *[]metav1.Condition, message string, generation int64) {
+	apimeta.SetStatusCondition(conditions, metav1.Condition{
+		Type:               v1alpha2.TypeUnpacked,
+		Status:             metav1.ConditionFalse,
+		Reason:             v1alpha2.ReasonUnpackFailed,
+		Message:            message,
+		ObservedGeneration: generation,
+	})
+}
+
+// setUnpackStatusPending sets the unpack status condition to pending.
+func setUnpackStatusPending(conditions *[]metav1.Condition, message string, generation int64) {
+	apimeta.SetStatusCondition(conditions, metav1.Condition{
+		Type:               v1alpha2.TypeUnpacked,
+		Status:             metav1.ConditionFalse,
+		Reason:             v1alpha2.ReasonUnpackPending,
+		Message:            message,
+		ObservedGeneration: generation,
+	})
+}
+
+// setUnpackStatusSuccessful sets the unpack status condition to success.
+func setUnpackStatusSuccessful(conditions *[]metav1.Condition, message string, generation int64) {
+	apimeta.SetStatusCondition(conditions, metav1.Condition{
+		Type:               v1alpha2.TypeUnpacked,
+		Status:             metav1.ConditionTrue,
+		Reason:             v1alpha2.ReasonUnpackSuccessful,
+		Message:            message,
+		ObservedGeneration: generation,
+	})
+}

internal/v1alpha2/controllers/bundledeployment/bundledeployment_suite_test.go

@@ -0,0 +1,76 @@
+/*
+Copyright 2023.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bundledeployment
+
+import (
+	"path/filepath"
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/operator-framework/rukpak/api/v1alpha2"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+var (
+	cfg     *rest.Config
+	cl      client.Client
+	sch     *runtime.Scheme
+	testEnv *envtest.Environment
+)
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "..", "..", "manifests", "base", "apis", "crds")},
+		ErrorIfCRDPathMissing: true,
+	}
+
+	var err error
+	// cfg is defined in this file globally.
+	cfg, err = testEnv.Start()
+	Expect(err).NotTo(HaveOccurred())
+	Expect(cfg).NotTo(BeNil())
+
+	sch = runtime.NewScheme()
+	err = corev1.AddToScheme(sch)
+	Expect(err).NotTo(HaveOccurred())
+	err = v1alpha2.AddToScheme(sch)
+	Expect(err).NotTo(HaveOccurred())
+
+	cl, err = client.New(cfg, client.Options{Scheme: sch})
+	Expect(err).NotTo(HaveOccurred())
+	Expect(cl).NotTo(BeNil())
+})
+
+var _ = AfterSuite(func() {
+	err := testEnv.Stop()
+	Expect(err).NotTo(HaveOccurred())
+})
+
+func TestAPIs(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Controller Suite")
+}

internal/v1alpha2/controllers/bundledeployment/bundledeployment_test.go

@@ -0,0 +1,53 @@
+/*
+Copyright 2023.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bundledeployment
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+var _ = Describe("Bundledeployment controller test", func() {
+	var (
+		ctx        context.Context
+		reconciler bundleDeploymentReconciler
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		reconciler = bundleDeploymentReconciler{
+			Client: cl,
+			Scheme: sch,
+		}
+	})
+
+	When("the bundle deployment does not exist", func() {
+		It("does not return an error", func() {
+			res, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: "non-existent"}})
+			Expect(res).To(Equal(ctrl.Result{}))
+			Expect(err).NotTo(HaveOccurred())
+		})
+	})
+
+	When("the bundle deployment exists on cluster", func() {
+	})
+
+})