Improve repository reconciliation to use cron based periodic syncs and introduce one time syncs (#328)

* Introduce cronbased and runOnceAt reconciliation for db-cache

* Introduce new command to allow user to run repository sync once

- updated repo reg and added tests to cover sync schedule
- added unit tests for runOnce feature

* Update cr cache to run repo sync using cron + runOnceAt mechanism

* Add a feature to change repo cr status when reconciliation is in progress

* fix comment

* Fix cache tests and reduce code complexity

* Align logs for handleRunOnceAt function

* Add unit tests for sync and git connectivity check

* Add more test coverage and refactor background cache updates

* Add retry mechanism for repo status update for api conflicts

* Align logs for invalid spec and add test coverage for external repo connection check

* Introduce next sync time in Repo CR status message

* Refactor cr and db cache to use a common sync interface

* Parameterize repo retry attempts and fix merge conflict

* Make fakeclient common to all tests and improve code

Author: kushnaidu

api/porchconfig/v1alpha1/config.porch.kpt.dev_repositories.yaml

@@ -34,6 +34,9 @@ spec:
     - jsonPath: .spec.content
       name: Content
       type: string
+    - jsonPath: .spec.sync.schedule
+      name: Sync schedule
+      type: string
     - jsonPath: .spec.deployment
       name: Deployment
       type: boolean
@@ -155,6 +158,21 @@ spec:
                 required:
                 - registry
                 type: object
+              sync:
+                description: Repository sync/reconcile details
+                properties:
+                  runOnceAt:
+                    description: Value in metav1.Time format to indicate when the
+                      repository should be synced once outside the periodic cron based
+                      reconcile loop.
+                    format: date-time
+                    type: string
+                  schedule:
+                    description: 'Cron value to indicate when the repository should
+                      be synced periodically. Example: `*/10 * * * *` to sync every
+                      10 minutes.'
+                    type: string
+                type: object
               type:
                 description: Type of the repository (i.e. git, OCI)
                 type: string

api/porchconfig/v1alpha1/types.go

@@ -23,6 +23,7 @@ import (
 //+kubebuilder:resource:path=repositories,singular=repository
 //+kubebuilder:printcolumn:name="Type",type=string,JSONPath=`.spec.type`
 //+kubebuilder:printcolumn:name="Content",type=string,JSONPath=`.spec.content`
+// +kubebuilder:printcolumn:name="Sync schedule",type=string,JSONPath=`.spec.sync.schedule`
 //+kubebuilder:printcolumn:name="Deployment",type=boolean,JSONPath=`.spec.deployment`
 //+kubebuilder:printcolumn:name="Ready",type=string,JSONPath=`.status.conditions[?(@.type=='Ready')].status`
 //+kubebuilder:printcolumn:name="Address",type=string,JSONPath=`.spec['git','oci']['repo','registry']`
@@ -67,13 +68,21 @@ type RepositorySpec struct {
 	// +kubebuilder:validation:XValidation:message="The 'content' field is deprecated, its only valid value is 'Package'",rule="self == '' || self == 'Package'"
 	// +kubebuilder:default="Package"
 	Content *RepositoryContent `json:"content,omitempty"`
-
+	// Repository sync/reconcile details
+	Sync *RepositorySync `json:"sync,omitempty"`
 	// Git repository details. Required if `type` is `git`. Ignored if `type` is not `git`.
 	Git *GitRepository `json:"git,omitempty"`
 	// OCI repository details. Required if `type` is `oci`. Ignored if `type` is not `oci`.
 	Oci *OciRepository `json:"oci,omitempty"`
 }
 
+type RepositorySync struct {
+	// Value in metav1.Time format to indicate when the repository should be synced once outside the periodic cron based reconcile loop.
+	RunOnceAt *metav1.Time `json:"runOnceAt,omitempty"`
+	// Cron value to indicate when the repository should be synced periodically. Example: `*/10 * * * *` to sync every 10 minutes.
+	Schedule string `json:"schedule,omitempty"`
+}
+
 // GitRepository describes a Git repository.
 // TODO: authentication methods
 type GitRepository struct {
@@ -139,11 +148,12 @@ type FunctionEval struct {
 const (
 	// Type of the Repository condition.
 	RepositoryReady = "Ready"
-
 	// Reason for the condition is error.
 	ReasonError = "Error"
 	// Reason for the condition is the repository is ready.
 	ReasonReady = "Ready"
+	// Reason for the condition is repository reconciliation is in progress.
+	ReasonReconciling = "Reconciling"
 )
 
 // RepositoryStatus defines the observed state of Repository

api/porchconfig/v1alpha1/zz_generated.deepcopy.go

@@ -78,6 +78,7 @@ spec:
             - --cert-dir=/tmp/certs
             - --secure-port=4443
             - --repo-sync-frequency=3m
+            - --repo-operation-retry-attempts=3
             - --disable-validating-admissions-policy=true
             - --max-request-body-size=6291456 # Keep this in sync with function-runner's corresponding argument
           #adding livenessProbes and readinessProbes for porch server

deployments/porch/3-porch-server.yaml

@@ -164,6 +164,7 @@ require (
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.2 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/sergi/go-diff v1.4.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect

go.mod

@@ -340,6 +340,8 @@ github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4
 github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
 github.com/qri-io/starlib v0.5.0 h1:NlveoBAhO6mNgM7+JpM9QlHh3/3pOtOiH6iXaqSdVK0=
 github.com/qri-io/starlib v0.5.0/go.mod h1:FpVumyB2CMrKIrjf39fAi4uydYWVvnWEvXEOwfzZRHY=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=

go.sum

@@ -92,8 +92,9 @@ type PorchServer struct {
 	GenericAPIServer          *genericapiserver.GenericAPIServer
 	coreClient                client.WithWatch
 	cache                     cachetypes.Cache
-	PeriodicRepoSyncFrequency time.Duration
-	ListTimeoutPerRepository  time.Duration
+	periodicRepoSyncFrequency    time.Duration
+	ListTimeoutPerRepository     time.Duration
+	repoOperationRetryAttempts   int
 }
 
 type completedConfig struct {
@@ -230,6 +231,7 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 	c.ExtraConfig.CacheOptions.ExternalRepoOptions.CredentialResolver = credentialResolver
 	c.ExtraConfig.CacheOptions.ExternalRepoOptions.CaBundleResolver = caBundleResolver
 	c.ExtraConfig.CacheOptions.ExternalRepoOptions.UserInfoProvider = userInfoProvider
+	c.ExtraConfig.CacheOptions.ExternalRepoOptions.RepoOperationRetryAttempts = c.ExtraConfig.CacheOptions.RepoOperationRetryAttempts
 
 	cacheImpl, err := cache.GetCacheImpl(ctx, c.ExtraConfig.CacheOptions)
 
@@ -255,6 +257,7 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 		engine.WithReferenceResolver(referenceResolver),
 		engine.WithUserInfoProvider(userInfoProvider),
 		engine.WithWatcherManager(watcherMgr),
+		engine.WithRepoOperationRetryAttempts(c.ExtraConfig.CacheOptions.RepoOperationRetryAttempts),
 	)
 	if err != nil {
 		return nil, err
@@ -278,8 +281,9 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 		coreClient:       coreClient,
 		cache:            cacheImpl,
 		// Set background job periodic frequency the same as repo sync frequency.
-		PeriodicRepoSyncFrequency: c.ExtraConfig.CacheOptions.RepoSyncFrequency,
-		ListTimeoutPerRepository:  c.ExtraConfig.ListTimeoutPerRepository,
+		periodicRepoSyncFrequency: c.ExtraConfig.CacheOptions.RepoSyncFrequency,
+		ListTimeoutPerRepository:    c.ExtraConfig.ListTimeoutPerRepository,
+		repoOperationRetryAttempts:  c.ExtraConfig.CacheOptions.RepoOperationRetryAttempts,
 	}
 
 	// Install the groups.
@@ -292,8 +296,9 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 
 func (s *PorchServer) Run(ctx context.Context) error {
 	porch.RunBackground(ctx, s.coreClient, s.cache,
-		porch.WithPeriodicRepoSyncFrequency(s.PeriodicRepoSyncFrequency),
+		porch.WithPeriodicRepoSyncFrequency(s.periodicRepoSyncFrequency),
 		porch.WithListTimeoutPerRepo(s.ListTimeoutPerRepository),
+		porch.WithRepoOperationRetryAttempts(s.repoOperationRetryAttempts),
 	)
 
 	// TODO: Reconsider if the existence of CERT_STORAGE_DIR was a good inidcator for webhook setup,

pkg/apiserver/apiserver.go

@@ -59,10 +59,14 @@ func (c *Cache) OpenRepository(ctx context.Context, repositorySpec *configapi.Re
 	c.mainLock.RLock()
 	if repo, ok := c.repositories[key]; ok && repo != nil {
 		c.mainLock.RUnlock()
-		// Test if credentials are okay for the cached repo and update the status accordingly
-		if _, err := externalrepo.CreateRepositoryImpl(ctx, repositorySpec, c.options.ExternalRepoOptions); err != nil {
+		// Keep the spec updated in the cache.
+		repo.repoSpec = repositorySpec
+		// Check external repo connectivity
+		if err := externalrepo.CheckRepositoryConnection(ctx, repositorySpec, c.options.ExternalRepoOptions); err != nil {
+			klog.Warningf("Cache:OpenRepository: repo %+v connectivity check failed with error %q", key, err)
 			return nil, err
 		}
+		klog.V(2).Infof("Cache::OpenRepository: verified repo connectivity %+v", key)
 		// If there is an error from the background refresh goroutine, return it.
 		if err := repo.getRefreshError(); err != nil {
 			return nil, err

pkg/cache/crcache/cache.go

@@ -26,7 +26,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	api "github.com/nephio-project/porch/api/porch/v1alpha1"
 	"github.com/nephio-project/porch/api/porchconfig/v1alpha1"
-
 	"github.com/nephio-project/porch/pkg/cache/crcache/meta"
 	fakemeta "github.com/nephio-project/porch/pkg/cache/crcache/meta/fake"
 	fakecache "github.com/nephio-project/porch/pkg/cache/fake"
@@ -35,6 +34,8 @@ import (
 	externalrepotypes "github.com/nephio-project/porch/pkg/externalrepo/types"
 	"github.com/nephio-project/porch/pkg/repository"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	k8sfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/yaml"
 )
 
@@ -246,21 +247,6 @@ func openRepositoryFromArchive(t *testing.T, ctx context.Context, testPath, name
 	_, address := git.ServeGitRepository(t, tarfile, tempdir)
 	metadataStore := createMetadataStoreFromArchive(t, fmt.Sprintf("%s-metadata.yaml", name), name)
 
-	cache := &Cache{
-		repositories:  map[repository.RepositoryKey]*cachedRepository{},
-		locks:         map[repository.RepositoryKey]*sync.Mutex{},
-		mainLock:      &sync.RWMutex{},
-		metadataStore: metadataStore,
-		options: cachetypes.CacheOptions{
-			ExternalRepoOptions: externalrepotypes.ExternalRepoOptions{
-				LocalDirectory:         t.TempDir(),
-				UseUserDefinedCaBundle: true,
-				CredentialResolver:     &fakecache.CredentialResolver{},
-			},
-			RepoSyncFrequency:    60 * time.Second,
-			RepoPRChangeNotifier: &fakecache.ObjectNotifier{},
-		}}
-
 	apiRepo := &v1alpha1.Repository{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       v1alpha1.TypeRepository.Kind,
@@ -278,6 +264,27 @@ func openRepositoryFromArchive(t *testing.T, ctx context.Context, testPath, name
 			},
 		},
 	}
+
+	scheme := runtime.NewScheme()
+	_ = v1alpha1.AddToScheme(scheme)
+
+	fakeClient := k8sfake.NewClientBuilder().WithScheme(scheme).WithObjects(apiRepo).Build()
+	cache := &Cache{
+		repositories:  map[repository.RepositoryKey]*cachedRepository{},
+		locks:         map[repository.RepositoryKey]*sync.Mutex{},
+		mainLock:      &sync.RWMutex{},
+		metadataStore: metadataStore,
+		options: cachetypes.CacheOptions{
+			ExternalRepoOptions: externalrepotypes.ExternalRepoOptions{
+				LocalDirectory:         t.TempDir(),
+				UseUserDefinedCaBundle: true,
+				CredentialResolver:     &fakecache.CredentialResolver{},
+				RepoOperationRetryAttempts: 3,
+			},
+			CoreClient:           fakeClient,
+			RepoSyncFrequency:    60 * time.Second,
+			RepoPRChangeNotifier: &fakecache.ObjectNotifier{},
+		}}
 	cachedRepo, err := cache.OpenRepository(ctx, apiRepo)
 	if err != nil {
 		t.Fatalf("OpenRepository(%q) of %q failed; %v", address, tarfile, err)