(feature) Implement polling image source in intervals

Implements https://docs.google.com/document/d/1iWSrWL9pYRJ5Ua3VYErkK1Q2lAusBUeDCh66Ew4lDbQ/edit?usp=sharing

Closes #180

api/core/v1alpha1/catalog_types.go

@@ -70,6 +70,7 @@ type CatalogList struct {
 }
 
 // CatalogSpec defines the desired state of Catalog
+// +kubebuilder:validation:XValidation:rule="!has(self.source.image.pollInterval) || (self.source.image.ref.find('sha256') != \"\" && self.source.image.pollInterval == \"\") || (self.source.image.ref.find('sha256') == \"\" && self.source.image.pollInterval == \"\") || (self.source.image.ref.find('sha256') == \"\" && self.source.image.pollInterval != \"\") ",message="cannot specify PollInterval while using tagged image"
 type CatalogSpec struct {
 	// Source is the source of a Catalog that contains Operators' metadata in the FBC format
 	// https://olm.operatorframework.io/docs/reference/file-based-catalogs/#docs
@@ -102,6 +103,12 @@ type ImageSource struct {
 	Ref string `json:"ref"`
 	// PullSecret contains the name of the image pull secret in the namespace that catalogd is deployed.
 	PullSecret string `json:"pullSecret,omitempty"`
+	// PollInterval indicates the interval at which the image source should be polled for new content.
+	// Eg 5m, 1h, 24h etc. A default value of 24h is used if a pollInterval is not mentioned.
+	// To switch off polling, specify "0s" as the value.
+	PollInterval *metav1.Duration `json:"pollInterval,omitempty"`
+	// LastPollAttempt logs the time when the image source was polled for new content.
+	LastPollAttempt metav1.Time `json:"lastPollAttempt,omitempty"`
 }
 
 func init() {

config/crd/bases/catalogd.operatorframework.io_catalogs.yaml

@@ -49,6 +49,18 @@ spec:
                     description: Image is the catalog image that backs the content
                       of this catalog.
                     properties:
+                      lastPollAttempt:
+                        description: LastPollAttempt logs the time when the image
+                          source was polled for new content.
+                        format: date-time
+                        type: string
+                      pollInterval:
+                        description: PollInterval indicates the interval at which
+                          the image source should be polled for new content. Eg 5m,
+                          1h, 24h etc. A default value of 24h is used if a pollInterval
+                          is not mentioned. To switch off polling, specify "0s" as
+                          the value.
+                        type: string
                       pullSecret:
                         description: PullSecret contains the name of the image pull
                           secret in the namespace that catalogd is deployed.
@@ -69,6 +81,12 @@ spec:
             required:
             - source
             type: object
+            x-kubernetes-validations:
+            - message: cannot specify PollInterval while using tagged image
+              rule: '!has(self.source.image.pollInterval) || (self.source.image.ref.find(''sha256'')
+                != "" && self.source.image.pollInterval == "") || (self.source.image.ref.find(''sha256'')
+                == "" && self.source.image.pollInterval == "") || (self.source.image.ref.find(''sha256'')
+                == "" && self.source.image.pollInterval != "") '
           status:
             description: CatalogStatus defines the observed state of Catalog
             properties:
@@ -156,6 +174,18 @@ spec:
                     description: Image is the catalog image that backs the content
                       of this catalog.
                     properties:
+                      lastPollAttempt:
+                        description: LastPollAttempt logs the time when the image
+                          source was polled for new content.
+                        format: date-time
+                        type: string
+                      pollInterval:
+                        description: PollInterval indicates the interval at which
+                          the image source should be polled for new content. Eg 5m,
+                          1h, 24h etc. A default value of 24h is used if a pollInterval
+                          is not mentioned. To switch off polling, specify "0s" as
+                          the value.
+                        type: string
                       pullSecret:
                         description: PullSecret contains the name of the image pull
                           secret in the namespace that catalogd is deployed.

config/samples/core_v1alpha1_catalog.yaml

@@ -6,4 +6,5 @@ spec:
   source:
     type: image
     image:
+      pollInterval: 1h
       ref: quay.io/operatorhubio/catalog:latest

internal/source/image_registry_client.go

@@ -9,12 +9,15 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/containerd/containerd/archive"
 	"github.com/google/go-containerregistry/pkg/authn/k8schain"
 	gcrkube "github.com/google/go-containerregistry/pkg/authn/kubernetes"
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	catalogdv1alpha1 "github.com/operator-framework/catalogd/api/core/v1alpha1"
@@ -106,17 +109,29 @@ func (i *ImageRegistry) Cleanup(_ context.Context, catalog *catalogdv1alpha1.Cat
 }
 
 func unpackedResult(fsys fs.FS, catalog *catalogdv1alpha1.Catalog, ref string) *Result {
-	return &Result{
+	result := &Result{
 		FS: fsys,
 		ResolvedSource: &catalogdv1alpha1.CatalogSource{
 			Type: catalogdv1alpha1.SourceTypeImage,
 			Image: &catalogdv1alpha1.ImageSource{
-				Ref:        ref,
-				PullSecret: catalog.Spec.Source.Image.PullSecret,
+				Ref:             ref,
+				PullSecret:      catalog.Spec.Source.Image.PullSecret,
+				LastPollAttempt: metav1.Now(),
 			},
 		},
 		State: StateUnpacked,
 	}
+	defaultDuration, _ := time.ParseDuration("24h")
+	disabledDuration, _ := time.ParseDuration("0h0m0s")
+	if catalog.Spec.Source.Image.PollInterval == nil {
+		result.RequeueAfter = &metav1.Duration{Duration: wait.Jitter(defaultDuration, 5.0)}
+		result.ResolvedSource.Image.PollInterval = &metav1.Duration{Duration: defaultDuration}
+	} else if catalog.Spec.Source.Image.PollInterval.Duration == disabledDuration {
+		result.RequeueAfter = nil
+	} else {
+		result.RequeueAfter = &metav1.Duration{Duration: wait.Jitter(catalog.Spec.Source.Image.PollInterval.Duration, 5.0)}
+	}
+	return result
 }
 
 // unpackImage unpacks a catalog image reference to the provided unpackPath,

internal/source/image_registry_client_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/registry"
@@ -361,3 +362,88 @@ func TestImageRegistry(t *testing.T) {
 		}
 	})
 }
+
+func TestPolling(t *testing.T) {
+	ctx := context.Background()
+	testCache := t.TempDir()
+	imgReg := &source.ImageRegistry{
+		BaseCachePath: testCache,
+	}
+	// Start a new server running an image registry
+	srv := httptest.NewServer(registry.New())
+	defer srv.Close()
+
+	// parse the server url so we can grab just the host
+	url, err := url.Parse(srv.URL)
+	assert.NoError(t, err)
+	duration0s, _ := time.ParseDuration("0s")
+	duration24h, _ := time.ParseDuration("24h")
+	t.Run("Polling tests", func(t *testing.T) {
+		for _, tt := range []struct {
+			testName                           string
+			inputCatalog                       v1alpha1.Catalog
+			expectedResolvedSourcePollInterval *metav1.Duration
+			shouldError                        bool
+		}{
+			{
+				testName: "When a poll interval is not specified, a default value of 24h is set in the resolved source as poll interval",
+				inputCatalog: v1alpha1.Catalog{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1alpha1.CatalogSpec{
+						Source: v1alpha1.CatalogSource{
+							Type: v1alpha1.SourceTypeImage,
+							Image: &v1alpha1.ImageSource{
+								Ref: fmt.Sprintf("%s/%s", url.Host, "test-image:latest"),
+							},
+						},
+					},
+				},
+				expectedResolvedSourcePollInterval: &metav1.Duration{Duration: duration24h},
+				shouldError:                        false,
+			},
+			{
+				testName: "When a poll interval of 0s is specified, no poll interval is set in the resolved source",
+				inputCatalog: v1alpha1.Catalog{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1alpha1.CatalogSpec{
+						Source: v1alpha1.CatalogSource{
+							Type: v1alpha1.SourceTypeImage,
+							Image: &v1alpha1.ImageSource{
+								Ref:          fmt.Sprintf("%s/%s", url.Host, "test-image:latest"),
+								PollInterval: &metav1.Duration{Duration: duration0s},
+							},
+						},
+					},
+				},
+				expectedResolvedSourcePollInterval: nil,
+				shouldError:                        false,
+			},
+		} {
+			t.Run(t.Name(), func(t *testing.T) {
+				img, err := random.Image(20, 3)
+				assert.NoError(t, err)
+				img, err = mutate.Config(img, v1.Config{
+					Labels: map[string]string{
+						source.ConfigDirLabel: "/configs",
+					},
+				})
+				assert.NoError(t, err)
+				ref, err := name.ParseReference(tt.inputCatalog.Spec.Source.Image.Ref)
+
+				assert.NoError(t, err)
+				remote.Write(ref, img)
+				res, err := imgReg.Unpack(ctx, &tt.inputCatalog)
+				if tt.shouldError {
+					assert.Error(t, err)
+				} else {
+					assert.NoError(t, err)
+				}
+				assert.Equal(t, res.ResolvedSource.Image.PollInterval, tt.expectedResolvedSourcePollInterval)
+			})
+		}
+	})
+}

internal/source/unpacker.go

@@ -8,6 +8,8 @@ import (
 	"path"
 
 	catalogdv1alpha1 "github.com/operator-framework/catalogd/api/core/v1alpha1"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // TODO: This package is almost entirely copy/pasted from rukpak. We should look
@@ -57,6 +59,10 @@ type Result struct {
 	// Message is contextual information about the progress of unpacking the
 	// catalog content.
 	Message string
+
+	// RequeueAfter contains the duration after which the controller should set
+	// up for Unpacking again
+	RequeueAfter *metav1.Duration
 }
 
 type State string

pkg/controllers/core/catalog_controller.go

@@ -151,7 +151,10 @@ func (r *CatalogReconciler) reconcile(ctx context.Context, catalog *v1alpha1.Cat
 		}
 
 		updateStatusUnpacked(&catalog.Status, unpackResult, contentURL)
-		return ctrl.Result{}, nil
+		if unpackResult.RequeueAfter == nil {
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{RequeueAfter: unpackResult.RequeueAfter.Duration}, nil
 	default:
 		return ctrl.Result{}, updateStatusUnpackFailing(&catalog.Status, fmt.Errorf("unknown unpack state %q: %v", unpackResult.State, err))
 	}

pkg/controllers/core/catalog_controller_test.go

@@ -135,43 +135,6 @@ var _ = Describe("Catalogd Controller Test", func() {
 		BeforeEach(func() {
 			catalogKey = types.NamespacedName{Name: fmt.Sprintf("catalogd-test-%s", rand.String(8))}
 		})
-
-		When("the catalog specifies an invalid source", func() {
-			BeforeEach(func() {
-				By("initializing cluster state")
-				catalog = &v1alpha1.Catalog{
-					ObjectMeta: metav1.ObjectMeta{Name: catalogKey.Name},
-					Spec: v1alpha1.CatalogSpec{
-						Source: v1alpha1.CatalogSource{
-							Type: "invalid-source",
-						},
-					},
-				}
-				Expect(cl.Create(ctx, catalog)).To(Succeed())
-			})
-
-			AfterEach(func() {
-				By("tearing down cluster state")
-				Expect(cl.Delete(ctx, catalog)).To(Succeed())
-			})
-
-			It("should set unpacking status to failed and return an error", func() {
-				res, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: catalogKey})
-				Expect(res).To(Equal(ctrl.Result{}))
-				Expect(err).To(HaveOccurred())
-
-				// get the catalog and ensure status is set properly
-				cat := &v1alpha1.Catalog{}
-				Expect(cl.Get(ctx, catalogKey, cat)).To(Succeed())
-				Expect(cat.Status.ResolvedSource).To(BeNil())
-				Expect(cat.Status.Phase).To(Equal(v1alpha1.PhaseFailing))
-				cond := meta.FindStatusCondition(cat.Status.Conditions, v1alpha1.TypeUnpacked)
-				Expect(cond).ToNot(BeNil())
-				Expect(cond.Reason).To(Equal(v1alpha1.ReasonUnpackFailed))
-				Expect(cond.Status).To(Equal(metav1.ConditionFalse))
-			})
-		})
-
 		When("the catalog specifies a valid source", func() {
 			BeforeEach(func() {
 				By("initializing cluster state")

test/e2e/cel_validation_test.go

@@ -0,0 +1,66 @@
+package e2e
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	catalogd "github.com/operator-framework/catalogd/api/core/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = FDescribe("CEL validation tests", func() {
+	var (
+		ctx     context.Context
+		catalog *catalogd.Catalog
+	)
+	When("A Catalog with a digest based image ref is created with a poll interval", func() {
+		duration45m, _ := time.ParseDuration("45m")
+		BeforeEach(func() {
+			ctx = context.Background()
+
+			catalog = &catalogd.Catalog{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: catalogName,
+				},
+				Spec: catalogd.CatalogSpec{
+					Source: catalogd.CatalogSource{
+						Type: catalogd.SourceTypeImage,
+						Image: &catalogd.ImageSource{
+							Ref:          "docker.io/test-image@sha256:asdf98234sd",
+							PollInterval: &metav1.Duration{Duration: duration45m},
+						},
+					},
+				},
+			}
+		})
+		It("throws an error", func() {
+			err = c.Create(ctx, catalog)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+	When("A Catalog with a digest based image ref is created without a poll interval", func() {
+		BeforeEach(func() {
+			ctx = context.Background()
+
+			catalog = &catalogd.Catalog{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: catalogName,
+				},
+				Spec: catalogd.CatalogSpec{
+					Source: catalogd.CatalogSource{
+						Type: catalogd.SourceTypeImage,
+						Image: &catalogd.ImageSource{
+							Ref: "docker.io/test-image@sha256:asdf98234sd",
+						},
+					},
+				},
+			}
+		})
+		It("does not throw an error", func() {
+			err = c.Create(ctx, catalog)
+			Expect(err).ToNot(HaveOccurred())
+		})
+	})
+})