feat(sync): local tmp store

Signed-off-by: a <[email protected]>

pkg/cli/server/root.go

@@ -381,12 +381,6 @@ func validateConfiguration(config *config.Config, log zlog.Logger) error {
 			return zerr.ErrBadConfig
 		}
 
-		// enforce filesystem storage in case sync feature is enabled
-		if config.Extensions != nil && config.Extensions.Sync != nil {
-			log.Error().Err(zerr.ErrBadConfig).Msg("sync supports only filesystem storage")
-
-			return zerr.ErrBadConfig
-		}
 	}
 
 	// enforce s3 driver on subpaths in case of using storage driver

pkg/extensions/config/sync/config.go

@@ -15,6 +15,7 @@ type Credentials struct {
 type Config struct {
 	Enable          *bool
 	CredentialsFile string
+	TmpDir          string
 	Registries      []RegistryConfig
 }
 

pkg/extensions/extension_sync.go

@@ -6,6 +6,7 @@ package extensions
 import (
 	"net"
 	"net/url"
+	"os"
 	"strings"
 
 	zerr "zotregistry.io/zot/errors"
@@ -41,23 +42,31 @@ func EnableSyncExtension(config *config.Config, metaDB mTypes.MetaDB,
 			isPeriodical := len(registryConfig.Content) != 0 && registryConfig.PollInterval != 0
 			isOnDemand := registryConfig.OnDemand
 
-			if isPeriodical || isOnDemand {
-				service, err := sync.New(registryConfig, config.Extensions.Sync.CredentialsFile,
-					storeController, metaDB, log)
-				if err != nil {
-					return nil, err
-				}
+			if !(isPeriodical || isOnDemand) {
+				continue
+			}
 
-				if isPeriodical {
-					// add to task scheduler periodic sync
-					gen := sync.NewTaskGenerator(service, log)
-					sch.SubmitGenerator(gen, registryConfig.PollInterval, scheduler.MediumPriority)
-				}
+			tmpDir := config.Extensions.Sync.TmpDir
+			if tmpDir == "" {
+				// use an os tmpdir as tmpdir if not set
+				tmpDir = os.TempDir()
+			}
 
-				if isOnDemand {
-					// onDemand services used in routes.go
-					onDemand.Add(service)
-				}
+			service, err := sync.New(registryConfig, config.Extensions.Sync.CredentialsFile, tmpDir,
+				storeController, metaDB, log)
+			if err != nil {
+				return nil, err
+			}
+
+			if isPeriodical {
+				// add to task scheduler periodic sync
+				gen := sync.NewTaskGenerator(service, log)
+				sch.SubmitGenerator(gen, registryConfig.PollInterval, scheduler.MediumPriority)
+			}
+
+			if isOnDemand {
+				// onDemand services used in routes.go
+				onDemand.Add(service)
 			}
 		}
 

pkg/extensions/sync/local.go → pkg/extensions/sync/destination.go

@@ -29,25 +29,34 @@ import (
 	storageTypes "zotregistry.io/zot/pkg/storage/types"
 )
 
-type LocalRegistry struct {
+type DestinationRegistry struct {
 	storeController storage.StoreController
 	tempStorage     OciLayoutStorage
 	metaDB          mTypes.MetaDB
 	log             log.Logger
 }
 
-func NewLocalRegistry(storeController storage.StoreController, metaDB mTypes.MetaDB, log log.Logger) Local {
-	return &LocalRegistry{
+func NewDestinationRegistry(
+	storeController storage.StoreController,
+	tmpStorage OciLayoutStorage,
+	metaDB mTypes.MetaDB,
+	log log.Logger,
+) Destination {
+	if tmpStorage == nil {
+		// to allow passing nil we can do this, noting that it will only work for a local StoreController
+		tmpStorage = NewOciLayoutStorage(storeController)
+	}
+	return &DestinationRegistry{
 		storeController: storeController,
 		metaDB:          metaDB,
 		// first we sync from remote (using containers/image copy from docker:// to oci:) to a temp imageStore
 		// then we copy the image from tempStorage to zot's storage using ImageStore APIs
-		tempStorage: NewOciLayoutStorage(storeController),
+		tempStorage: tmpStorage,
 		log:         log,
 	}
 }
 
-func (registry *LocalRegistry) CanSkipImage(repo, tag string, imageDigest digest.Digest) (bool, error) {
+func (registry *DestinationRegistry) CanSkipImage(repo, tag string, imageDigest digest.Digest) (bool, error) {
 	// check image already synced
 	imageStore := registry.storeController.GetImageStore(repo)
 
@@ -75,16 +84,16 @@ func (registry *LocalRegistry) CanSkipImage(repo, tag string, imageDigest digest
 	return true, nil
 }
 
-func (registry *LocalRegistry) GetContext() *types.SystemContext {
+func (registry *DestinationRegistry) GetContext() *types.SystemContext {
 	return registry.tempStorage.GetContext()
 }
 
-func (registry *LocalRegistry) GetImageReference(repo, reference string) (types.ImageReference, error) {
+func (registry *DestinationRegistry) GetImageReference(repo, reference string) (types.ImageReference, error) {
 	return registry.tempStorage.GetImageReference(repo, reference)
 }
 
 // finalize a syncing image.
-func (registry *LocalRegistry) CommitImage(imageReference types.ImageReference, repo, reference string) error {
+func (registry *DestinationRegistry) CommitImage(imageReference types.ImageReference, repo, reference string) error {
 	imageStore := registry.storeController.GetImageStore(repo)
 
 	tempImageStore := getImageStoreFromImageReference(imageReference, repo, reference)
@@ -180,7 +189,7 @@ func (registry *LocalRegistry) CommitImage(imageReference types.ImageReference,
 	return nil
 }
 
-func (registry *LocalRegistry) copyManifest(repo string, manifestContent []byte, reference string,
+func (registry *DestinationRegistry) copyManifest(repo string, manifestContent []byte, reference string,
 	tempImageStore storageTypes.ImageStore,
 ) error {
 	imageStore := registry.storeController.GetImageStore(repo)
@@ -239,7 +248,7 @@ func (registry *LocalRegistry) copyManifest(repo string, manifestContent []byte,
 }
 
 // Copy a blob from one image store to another image store.
-func (registry *LocalRegistry) copyBlob(repo string, blobDigest digest.Digest, blobMediaType string,
+func (registry *DestinationRegistry) copyBlob(repo string, blobDigest digest.Digest, blobMediaType string,
 	tempImageStore storageTypes.ImageStore,
 ) error {
 	imageStore := registry.storeController.GetImageStore(repo)

pkg/extensions/sync/service.go

@@ -20,13 +20,14 @@ import (
 	"zotregistry.io/zot/pkg/log"
 	mTypes "zotregistry.io/zot/pkg/meta/types"
 	"zotregistry.io/zot/pkg/storage"
+	"zotregistry.io/zot/pkg/storage/local"
 )
 
 type BaseService struct {
 	config          syncconf.RegistryConfig
 	credentials     syncconf.CredentialsFile
 	remote          Remote
-	local           Local
+	destination     Destination
 	retryOptions    *retry.RetryOptions
 	contentManager  ContentManager
 	storeController storage.StoreController
@@ -37,8 +38,13 @@ type BaseService struct {
 	log             log.Logger
 }
 
-func New(opts syncconf.RegistryConfig, credentialsFilepath string,
-	storeController storage.StoreController, metadb mTypes.MetaDB, log log.Logger,
+func New(
+	opts syncconf.RegistryConfig,
+	credentialsFilepath string,
+	tmpDir string,
+	storeController storage.StoreController,
+	metadb mTypes.MetaDB,
+	log log.Logger,
 ) (Service, error) {
 	service := &BaseService{}
 
@@ -60,7 +66,14 @@ func New(opts syncconf.RegistryConfig, credentialsFilepath string,
 	service.credentials = credentialsFile
 
 	service.contentManager = NewContentManager(opts.Content, log)
-	service.local = NewLocalRegistry(storeController, metadb, log)
+
+	tmpImageStore := local.NewImageStore(tmpDir,
+		false, false, log, nil, nil, nil,
+	)
+
+	tmpStorage := NewOciLayoutStorage(storage.StoreController{DefaultStore: tmpImageStore})
+
+	service.destination = NewDestinationRegistry(storeController, tmpStorage, metadb, log)
 
 	retryOptions := &retry.RetryOptions{}
 
@@ -289,7 +302,7 @@ func (service *BaseService) SyncRepo(ctx context.Context, repo string) error {
 	service.log.Info().Str("repo", repo).Msgf("sync: syncing tags %v", tags)
 
 	// apply content.destination rule
-	localRepo := service.contentManager.GetRepoDestination(repo)
+	destinationRepo := service.contentManager.GetRepoDestination(repo)
 
 	for _, tag := range tags {
 		select {
@@ -305,7 +318,7 @@ func (service *BaseService) SyncRepo(ctx context.Context, repo string) error {
 		var manifestDigest digest.Digest
 
 		if err = retry.RetryIfNecessary(ctx, func() error {
-			manifestDigest, err = service.syncTag(ctx, localRepo, repo, tag)
+			manifestDigest, err = service.syncTag(ctx, destinationRepo, repo, tag)
 
 			return err
 		}, service.retryOptions); err != nil {
@@ -322,7 +335,7 @@ func (service *BaseService) SyncRepo(ctx context.Context, repo string) error {
 
 		if manifestDigest != "" {
 			if err = retry.RetryIfNecessary(ctx, func() error {
-				err = service.references.SyncAll(ctx, localRepo, repo, manifestDigest.String())
+				err = service.references.SyncAll(ctx, destinationRepo, repo, manifestDigest.String())
 				if errors.Is(err, zerr.ErrSyncReferrerNotFound) {
 					return nil
 				}
@@ -342,8 +355,8 @@ func (service *BaseService) SyncRepo(ctx context.Context, repo string) error {
 	return nil
 }
 
-func (service *BaseService) syncTag(ctx context.Context, localRepo, remoteRepo, tag string) (digest.Digest, error) {
-	copyOptions := getCopyOptions(service.remote.GetContext(), service.local.GetContext())
+func (service *BaseService) syncTag(ctx context.Context, destinationRepo, remoteRepo, tag string) (digest.Digest, error) {
+	copyOptions := getCopyOptions(service.remote.GetContext(), service.destination.GetContext())
 
 	policyContext, err := getPolicyContext(service.log)
 	if err != nil {
@@ -386,38 +399,38 @@ func (service *BaseService) syncTag(ctx context.Context, localRepo, remoteRepo,
 		}
 	}
 
-	skipImage, err := service.local.CanSkipImage(localRepo, tag, manifestDigest)
+	skipImage, err := service.destination.CanSkipImage(destinationRepo, tag, manifestDigest)
 	if err != nil {
 		service.log.Error().Err(err).Str("errortype", common.TypeOf(err)).
-			Str("repo", localRepo).Str("reference", tag).
+			Str("repo", destinationRepo).Str("reference", tag).
 			Msg("couldn't check if the local image can be skipped")
 	}
 
 	if !skipImage {
-		localImageRef, err := service.local.GetImageReference(localRepo, tag)
+		localImageRef, err := service.destination.GetImageReference(destinationRepo, tag)
 		if err != nil {
 			service.log.Error().Err(err).Str("errortype", common.TypeOf(err)).
-				Str("repo", localRepo).Str("reference", tag).Msg("couldn't get a local image reference")
+				Str("repo", destinationRepo).Str("reference", tag).Msg("couldn't get a local image reference")
 
 			return "", err
 		}
 
 		service.log.Info().Str("remote image", remoteImageRef.DockerReference().String()).
-			Str("local image", fmt.Sprintf("%s:%s", localRepo, tag)).Msg("syncing image")
+			Str("local image", fmt.Sprintf("%s:%s", destinationRepo, tag)).Msg("syncing image")
 
 		_, err = copy.Image(ctx, policyContext, localImageRef, remoteImageRef, &copyOptions)
 		if err != nil {
 			service.log.Error().Err(err).Str("errortype", common.TypeOf(err)).
 				Str("remote image", remoteImageRef.DockerReference().String()).
-				Str("local image", fmt.Sprintf("%s:%s", localRepo, tag)).Msg("coulnd't sync image")
+				Str("local image", fmt.Sprintf("%s:%s", destinationRepo, tag)).Msg("coulnd't sync image")
 
 			return "", err
 		}
 
-		err = service.local.CommitImage(localImageRef, localRepo, tag)
+		err = service.destination.CommitImage(localImageRef, destinationRepo, tag)
 		if err != nil {
 			service.log.Error().Err(err).Str("errortype", common.TypeOf(err)).
-				Str("repo", localRepo).Str("reference", tag).Msg("couldn't commit image to local image store")
+				Str("repo", destinationRepo).Str("reference", tag).Msg("couldn't commit image to local image store")
 
 			return "", err
 		}

pkg/extensions/sync/sync.go

@@ -65,7 +65,7 @@ type Remote interface {
 }
 
 // Local registry.
-type Local interface {
+type Destination interface {
 	Registry
 	// Check if an image is already synced
 	CanSkipImage(repo, tag string, imageDigest digest.Digest) (bool, error)

pkg/extensions/sync/sync_internal_test.go

@@ -162,15 +162,15 @@ func TestService(t *testing.T) {
 			URLs: []string{"http://localhost"},
 		}
 
-		service, err := New(conf, "", storage.StoreController{}, mocks.MetaDBMock{}, log.Logger{})
+		service, err := New(conf, "", os.TempDir(), storage.StoreController{}, mocks.MetaDBMock{}, log.Logger{})
 		So(err, ShouldBeNil)
 
 		err = service.SyncRepo(context.Background(), "repo")
 		So(err, ShouldNotBeNil)
 	})
 }
 
-func TestLocalRegistry(t *testing.T) {
+func TestDestinationRegistry(t *testing.T) {
 	Convey("make StoreController", t, func() {
 		dir := t.TempDir()
 
@@ -185,7 +185,7 @@ func TestLocalRegistry(t *testing.T) {
 		syncImgStore := local.NewImageStore(dir, true, true, log, metrics, nil, cacheDriver)
 		repoName := "repo"
 
-		registry := NewLocalRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, log)
+		registry := NewDestinationRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, nil, log)
 		imageReference, err := registry.GetImageReference(repoName, "1.0")
 		So(err, ShouldBeNil)
 		So(imageReference, ShouldNotBeNil)
@@ -302,7 +302,7 @@ func TestLocalRegistry(t *testing.T) {
 			syncImgStore := local.NewImageStore(dir, true, true, log, metrics, linter, cacheDriver)
 			repoName := "repo"
 
-			registry := NewLocalRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, log)
+			registry := NewDestinationRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, nil, log)
 
 			err = registry.CommitImage(imageReference, repoName, "1.0")
 			So(err, ShouldBeNil)
@@ -336,7 +336,7 @@ func TestLocalRegistry(t *testing.T) {
 		})
 
 		Convey("trigger metaDB error on index manifest in CommitImage()", func() {
-			registry := NewLocalRegistry(storage.StoreController{DefaultStore: syncImgStore}, mocks.MetaDBMock{
+			registry := NewDestinationRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, mocks.MetaDBMock{
 				SetRepoReferenceFn: func(ctx context.Context, repo string, reference string, imageMeta mTypes.ImageMeta) error {
 					if reference == "1.0" {
 						return zerr.ErrRepoMetaNotFound
@@ -351,7 +351,7 @@ func TestLocalRegistry(t *testing.T) {
 		})
 
 		Convey("trigger metaDB error on image manifest in CommitImage()", func() {
-			registry := NewLocalRegistry(storage.StoreController{DefaultStore: syncImgStore}, mocks.MetaDBMock{
+			registry := NewDestinationRegistry(storage.StoreController{DefaultStore: syncImgStore}, nil, mocks.MetaDBMock{
 				SetRepoReferenceFn: func(ctx context.Context, repo, reference string, imageMeta mTypes.ImageMeta) error {
 					return zerr.ErrRepoMetaNotFound
 				},

pkg/storage/imagestore/imagestore.go

@@ -392,6 +392,14 @@ func (is *ImageStore) GetNextRepository(repo string) (string, error) {
 
 	driverErr := &driver.Error{}
 
+	// some s3 implementations (eg, digitalocean spaces) will return pathnotfounderror for walk but not list
+	// therefore, we must also catch that error here.
+	if errors.As(err, &driver.PathNotFoundError{}) {
+		is.log.Debug().Msg("empty rootDir")
+
+		return "", nil
+	}
+
 	if errors.Is(err, io.EOF) ||
 		(errors.As(err, driverErr) && errors.Is(driverErr.Enclosed, io.EOF)) {
 		return store, nil