Make parallel repository listing configurable + Porch server doesn't respond to PackageRevision queries if one of the registered git repositories fails (#282)

* Porch server doesn't respond to PackageRevision queries if one of the registered git repositories fails

* Changes to pr listing

* Make parallel repository listing configurable

* Fixes to way context cancel is handled

* Generate mocks for PackageRevision and PackageRevisionDraft and refactor tests to use it

* Update tests to align with main

* Refactor of test cases

* Fix tests after fixing merge conflicts

* Fix hanging repository test

* Fixes from commits

* Refactor of test cases

* Merged tests together and refactored

* Add missing envvars

* Refactor envvar for kpt_function_test

* Missed default branch for kpt_function_test

* Fix fake pkgrev after rebase

* rewrite mutex logic of repository caching

* fix lint

* Fix failing e2e tests

* Added testing to background

* add main mutex to map operations in dbcache

* Add missed test

---------

Co-authored-by: Kosimov Javlonbek <javlonbek.kosimov@nokia.com>
Co-authored-by: Mozes László <laszlo.mozes@nokia.com>

Author: 3 people

.env.template

@@ -14,10 +14,12 @@
 #PORCH_KPT_REPO_USER=<username>
 #PORCH_KPT_REPO_PASSWORD=<password/access_token>
 
+#PORCH_GCP_BUCKET_REF=<branch_name>
+#PORCH_GCP_REDIS_BUCKET_REF=<branch_name>
+#PORCH_GCP_HIERARCHY_REF=<branch_name>
+#PORCH_KPT_FUNCTION_REF=<branch_name>
+
 # GCR prefix should not contain trailing "/": TestPodFunctionEvaluatorWithDistrolessImage uses kpt-fn-demo directory.
 # If GRC prefix environment variable is set, make run-in-kind should be run after for changes to take effect.
 # Format: your-gcr-mirror/kpt-fn
 #PORCH_GCR_PREFIX_URL=<your_gcr_registry_mirror_url>
-
-# This environment variable is used in TestPodEvaluator and should only be set if the PORCH_GCR_PREFIX_URL is set.
-#PORCH_POD_EVAL_REF=<branch_name>

.mockery.yaml

@@ -10,6 +10,14 @@ packages:
         config:
           dir: "test/mockery/mocks/external/{{.InterfaceDirRelative}}"
           outpkg: "{{.PackageName}}"
+      SubResourceWriter:
+        config:
+          dir: "test/mockery/mocks/external/{{.InterfaceDirRelative}}"
+          outpkg: "{{.PackageName}}"
+      WithWatch:
+        config:
+          dir: "test/mockery/mocks/external/{{.InterfaceDirRelative}}"
+          outpkg: "{{.PackageName}}"
   github.com/nephio-project/porch/pkg/cache/crcache/meta:
     interfaces:
       MetadataStore:

pkg/apiserver/apiserver.go

@@ -74,9 +74,11 @@ func init() {
 
 // ExtraConfig holds custom apiserver config
 type ExtraConfig struct {
-	CoreAPIKubeconfigPath string
-	GRPCRuntimeOptions    engine.GRPCRuntimeOptions
-	CacheOptions          cachetypes.CacheOptions
+	CoreAPIKubeconfigPath    string
+	GRPCRuntimeOptions       engine.GRPCRuntimeOptions
+	CacheOptions             cachetypes.CacheOptions
+	ListTimeoutPerRepository time.Duration
+	MaxConcurrentLists       int
 }
 
 // Config defines the config for the apiserver
@@ -91,6 +93,7 @@ type PorchServer struct {
 	coreClient                client.WithWatch
 	cache                     cachetypes.Cache
 	PeriodicRepoSyncFrequency time.Duration
+	ListTimeoutPerRepository  time.Duration
 }
 
 type completedConfig struct {
@@ -260,7 +263,15 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 		return nil, err
 	}
 
-	porchGroup, err := porch.NewRESTStorage(Scheme, Codecs, cad, coreClient)
+	restStorageOptions := porch.RESTStorageOptions{
+		Scheme:               Scheme,
+		Codecs:               Codecs,
+		CaD:                  cad,
+		CoreClient:           coreClient,
+		TimeoutPerRepository: c.ExtraConfig.ListTimeoutPerRepository,
+		MaxConcurrentLists:   c.ExtraConfig.MaxConcurrentLists,
+	}
+	porchGroup, err := restStorageOptions.NewRESTStorage()
 	if err != nil {
 		return nil, err
 	}
@@ -271,6 +282,7 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 		cache:            cacheImpl,
 		// Set background job periodic frequency the same as repo sync frequency.
 		PeriodicRepoSyncFrequency: c.ExtraConfig.CacheOptions.RepoSyncFrequency,
+		ListTimeoutPerRepository:  c.ExtraConfig.ListTimeoutPerRepository,
 	}
 
 	// Install the groups.
@@ -282,7 +294,10 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) {
 }
 
 func (s *PorchServer) Run(ctx context.Context) error {
-	porch.RunBackground(ctx, s.coreClient, s.cache, s.PeriodicRepoSyncFrequency)
+	porch.RunBackground(ctx, s.coreClient, s.cache,
+		porch.WithPeriodicRepoSyncFrequency(s.PeriodicRepoSyncFrequency),
+		porch.WithListTimeoutPerRepo(s.ListTimeoutPerRepository),
+	)
 
 	// TODO: Reconsider if the existence of CERT_STORAGE_DIR was a good inidcator for webhook setup,
 	// but for now we keep backward compatiblity

pkg/cache/crcache/cache.go

@@ -17,6 +17,7 @@ package crcache
 import (
 	"context"
 	"sync"
+	"time"
 
 	configapi "github.com/nephio-project/porch/api/porchconfig/v1alpha1"
 	"github.com/nephio-project/porch/pkg/cache/crcache/meta"
@@ -25,13 +26,15 @@ import (
 	"github.com/nephio-project/porch/pkg/repository"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
+	"k8s.io/klog/v2"
 )
 
 var tracer = otel.Tracer("crcache")
 
 type Cache struct {
-	mutex         sync.Mutex
 	repositories  map[repository.RepositoryKey]*cachedRepository
+	mainLock      *sync.RWMutex
+	locks         map[repository.RepositoryKey]*sync.Mutex
 	metadataStore meta.MetadataStore
 	options       cachetypes.CacheOptions
 }
@@ -41,40 +44,48 @@ var _ cachetypes.Cache = &Cache{}
 func (c *Cache) OpenRepository(ctx context.Context, repositorySpec *configapi.Repository) (repository.Repository, error) {
 	ctx, span := tracer.Start(ctx, "Cache::OpenRepository", trace.WithAttributes())
 	defer span.End()
+	start := time.Now()
+	defer func() { klog.V(4).Infof("Cache::OpenRepository (%s) took %s", repositorySpec.Name, time.Since(start)) }()
 
 	key, err := externalrepo.RepositoryKey(repositorySpec)
 	if err != nil {
 		return nil, err
 	}
 
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	lock := c.getOrInsertLock(key)
+	lock.Lock()
+	defer lock.Unlock()
 
-	if cachedRepo := c.repositories[key]; cachedRepo != nil {
+	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 {
-			return cachedRepo, err
-		} else {
-			// If there is an error from the background refresh goroutine, return it.
-			if err := cachedRepo.getRefreshError(); err == nil {
-				return cachedRepo, nil
-			}
+		if _, err := externalrepo.CreateRepositoryImpl(ctx, repositorySpec, c.options.ExternalRepoOptions); err != nil {
+			return nil, err
 		}
-		return nil, err
+		// If there is an error from the background refresh goroutine, return it.
+		if err := repo.getRefreshError(); err != nil {
+			return nil, err
+		}
+		return repo, nil
 	}
+	c.mainLock.RUnlock()
 
 	externalRepo, err := externalrepo.CreateRepositoryImpl(ctx, repositorySpec, c.options.ExternalRepoOptions)
 	if err != nil {
 		return nil, err
 	}
 
 	cachedRepo := newRepository(key, repositorySpec, externalRepo, c.metadataStore, c.options)
+
+	c.mainLock.Lock()
 	c.repositories[key] = cachedRepo
+	c.mainLock.Unlock()
 
 	return cachedRepo, nil
 }
 
-func (c *Cache) UpdateRepository(ctx context.Context, repositorySpec *configapi.Repository) error {
+func (c *Cache) UpdateRepository(context.Context, *configapi.Repository) error {
 	panic("Update on CR cached repositories is not applicable")
 }
 
@@ -98,36 +109,69 @@ func (c *Cache) CloseRepository(ctx context.Context, repositorySpec *configapi.R
 		}
 		if otherKey == key {
 			// do not close cached repo if it is shared
+			klog.Infof("Not closing cached repository %q because it is shared", key)
 			return nil
 		}
 	}
 
-	var repository *cachedRepository
-	{
-		c.mutex.Lock()
-		if r, ok := c.repositories[key]; ok {
-			delete(c.repositories, key)
-			repository = r
-		}
-		c.mutex.Unlock()
-	}
+	lock := c.getOrInsertLock(key)
+	lock.Lock()
+	defer lock.Unlock()
+
+	c.mainLock.RLock()
+	repo, ok := c.repositories[key]
+	c.mainLock.RUnlock()
 
-	if repository != nil {
-		return repository.Close(ctx)
+	if ok {
+		c.mainLock.Lock()
+		delete(c.locks, key)
+		delete(c.repositories, key)
+		c.mainLock.Unlock()
+
+		if repo != nil {
+			return repo.Close(ctx)
+		} else {
+			klog.Warningf("cached repository with key %q had stored value nil", key)
+		}
 	} else {
-		return nil
+		c.mainLock.Lock()
+		delete(c.locks, key)
+		c.mainLock.Unlock()
 	}
+
+	return nil
 }
 
 func (c *Cache) GetRepositories() []*configapi.Repository {
 	repoSlice := []*configapi.Repository{}
 
+	c.mainLock.RLock()
+	defer c.mainLock.RUnlock()
 	for _, repo := range c.repositories {
 		repoSlice = append(repoSlice, repo.repoSpec)
 	}
+
 	return repoSlice
 }
 
 func (c *Cache) GetRepository(repoKey repository.RepositoryKey) repository.Repository {
+	c.mainLock.RLock()
+	defer c.mainLock.RUnlock()
 	return c.repositories[repoKey]
 }
+
+func (c *Cache) getOrInsertLock(key repository.RepositoryKey) *sync.Mutex {
+	c.mainLock.RLock()
+	if lock, exists := c.locks[key]; exists {
+		c.mainLock.RUnlock()
+		return lock
+	}
+	c.mainLock.RUnlock()
+
+	c.mainLock.Lock()
+	lock := &sync.Mutex{}
+	c.locks[key] = lock
+	c.mainLock.Unlock()
+
+	return lock
+}

pkg/cache/crcache/cache_test.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -246,7 +247,9 @@ func openRepositoryFromArchive(t *testing.T, ctx context.Context, testPath, name
 	metadataStore := createMetadataStoreFromArchive(t, fmt.Sprintf("%s-metadata.yaml", name), name)
 
 	cache := &Cache{
-		repositories:  make(map[repository.RepositoryKey]*cachedRepository),
+		repositories:  map[repository.RepositoryKey]*cachedRepository{},
+		locks:         map[repository.RepositoryKey]*sync.Mutex{},
+		mainLock:      &sync.RWMutex{},
 		metadataStore: metadataStore,
 		options: cachetypes.CacheOptions{
 			ExternalRepoOptions: externalrepotypes.ExternalRepoOptions{

pkg/cache/crcache/crcachefactory.go

@@ -16,6 +16,7 @@ package crcache
 
 import (
 	"context"
+	"sync"
 
 	"github.com/nephio-project/porch/pkg/cache/crcache/meta"
 	cachetypes "github.com/nephio-project/porch/pkg/cache/types"
@@ -29,7 +30,9 @@ type CrCacheFactory struct {
 
 func (f *CrCacheFactory) NewCache(_ context.Context, options cachetypes.CacheOptions) (cachetypes.Cache, error) {
 	return &Cache{
-		repositories:  make(map[repository.RepositoryKey]*cachedRepository),
+		repositories:  map[repository.RepositoryKey]*cachedRepository{},
+		locks:         map[repository.RepositoryKey]*sync.Mutex{},
+		mainLock:      &sync.RWMutex{},
 		metadataStore: meta.NewCrdMetadataStore(options.CoreClient),
 		options:       options,
 	}, nil

pkg/cache/crcache/repository.go

@@ -144,7 +144,7 @@ func (r *cachedRepository) getRefreshError() error {
 }
 
 func (r *cachedRepository) getPackageRevisions(ctx context.Context, filter repository.ListPackageRevisionFilter, forceRefresh bool) ([]repository.PackageRevision, error) {
-
+	klog.Infof("Cache::OpenRepository(%s) fetching packages", r.Key())
 	_, packageRevisions, err := r.getCachedPackages(ctx, forceRefresh)
 	if err != nil {
 		return nil, err
@@ -490,22 +490,24 @@ func (r *cachedRepository) Close(ctx context.Context) error {
 
 // pollForever will continue polling until the signal channel is closed or the ctx is done.
 func (r *cachedRepository) pollForever(ctx context.Context, repoSyncFrequency time.Duration) {
+	ticker := time.NewTicker(repoSyncFrequency)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
-			klog.V(2).Infof("repo %+v: exiting repository poller, because context is done: %v", r.Key(), ctx.Err())
+			klog.Infof("repo %+v: exiting repository poller, because context is done: %v", r.Key(), ctx.Err())
 			return
-		default:
+		case <-ticker.C:
 			r.pollOnce(ctx)
-			time.Sleep(repoSyncFrequency)
+			ticker.Reset(repoSyncFrequency)
 		}
 	}
 }
 
 func (r *cachedRepository) pollOnce(ctx context.Context) {
 	start := time.Now()
 	klog.Infof("repo %+v: poll started", r.Key())
-	defer func() { klog.Infof("repo %+v: poll finished in %f secs", r.Key(), time.Since(start).Seconds()) }()
+	defer func() { klog.Infof("repo %+v: poll finished in %s", r.Key(), time.Since(start)) }()
 	ctx, span := tracer.Start(ctx, "[START]::Repository::pollOnce", trace.WithAttributes())
 	defer span.End()