Enhance sync logic - stop blob redownloads and re-pushes (#479 #480)

Changed imagesToCopyFromUpstream to return a map[string][]types.ImageReference from just an array of refs
Rewrote some logic in sync.go to use the new signature of imagesToCopyFromUpstream
Split getLocalImageRef by adding function getLocalCachePath
Adapted tests for new changes, added some tests
Merged #481

Signed-off-by: Catalin Hofnar <catalin.hofnar@gmail.com>

pkg/extensions/sync/on_demand.go

@@ -309,7 +309,12 @@ func syncRun(regCfg RegistryConfig, localRepo, remoteRepo, tag string, utils syn
 		}
 	}
 
-	localImageRef, localCachePath, err := getLocalImageRef(utils.imageStore, localRepo, tag)
+	localCachePath, err := getLocalCachePath(utils.imageStore, localRepo)
+	if err != nil {
+		log.Error().Err(err).Msgf("couldn't get localCachePath for %s", localRepo)
+	}
+
+	localImageRef, err := getLocalImageRef(localCachePath, localRepo, tag)
 	if err != nil {
 		log.Error().Err(err).Msgf("couldn't obtain a valid image reference for reference %s/%s:%s",
 			localCachePath, localRepo, tag)

pkg/extensions/sync/sync.go

@@ -190,10 +190,12 @@ func filterImagesBySemver(upstreamReferences *[]types.ImageReference, content Co
 // imagesToCopyFromRepos lists all images given a registry name and its repos.
 func imagesToCopyFromUpstream(ctx context.Context, registryName string, repos []string,
 	upstreamCtx *types.SystemContext, content Content, log log.Logger,
-) ([]types.ImageReference, error) {
-	var upstreamReferences []types.ImageReference
+) (map[string][]types.ImageReference, error) {
+	upstreamReferences := make(map[string][]types.ImageReference)
 
 	for _, repoName := range repos {
+		repoUpstreamReferences := make([]types.ImageReference, 0)
+
 		repoRef, err := parseRepositoryReference(fmt.Sprintf("%s/%s", registryName, repoName))
 		if err != nil {
 			log.Error().Err(err).Msgf("couldn't parse repository reference: %s", repoRef)
@@ -230,22 +232,26 @@ func imagesToCopyFromUpstream(ctx context.Context, registryName string, repos []
 				return nil, err
 			}
 
-			upstreamReferences = append(upstreamReferences, ref)
-		}
+			repoUpstreamReferences = append(repoUpstreamReferences, ref)
 		}
 
-	log.Debug().Msgf("upstream refs to be copied: %v", upstreamReferences)
+		upstreamReferences[repoName] = repoUpstreamReferences
 
-	err := filterImagesByTagRegex(&upstreamReferences, content, log)
+		log.Debug().Msgf("repo: %s - upstream refs to be copied: %v", repoName, upstreamReferences)
+
+		err = filterImagesByTagRegex(&repoUpstreamReferences, content, log)
 		if err != nil {
-		return []types.ImageReference{}, err
+			return map[string][]types.ImageReference{}, err
 		}
 
-	log.Debug().Msgf("remaining upstream refs to be copied: %v", upstreamReferences)
+		log.Debug().Msgf("repo: %s - remaining upstream refs to be copied: %v", repoName, repoUpstreamReferences)
 
-	filterImagesBySemver(&upstreamReferences, content, log)
+		filterImagesBySemver(&repoUpstreamReferences, content, log)
 
-	log.Debug().Msgf("remaining upstream refs to be copied: %v", upstreamReferences)
+		log.Debug().Msgf("repo: %s - remaining upstream refs to be copied: %v", repoName, repoUpstreamReferences)
+
+		upstreamReferences[repoName] = repoUpstreamReferences
+	}
 
 	return upstreamReferences, nil
 }
@@ -284,6 +290,7 @@ func getUpstreamContext(regCfg *RegistryConfig, credentials Credentials) *types.
 	return upstreamCtx
 }
 
+// nolint:gocyclo  // offloading some of the functionalities from here would make the code harder to follow
 func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string,
 	storeController storage.StoreController, localCtx *types.SystemContext,
 	policyCtx *signature.PolicyContext, credentials Credentials,
@@ -321,21 +328,26 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 
 	log.Info().Msgf("got repos: %v", repos)
 
-	var images []struct {
+	upstreamAddr := StripRegistryTransport(upstreamURL)
+
+	reposWithContentID := make(map[string][]struct {
 		ref     types.ImageReference
 		content Content
-	}
-
-	upstreamAddr := StripRegistryTransport(upstreamURL)
+	})
 
 	for contentID, repos := range repos {
 		r := repos
 		contentID := contentID
 
 		if err = retry.RetryIfNecessary(ctx, func() error {
+			for _, repo := range r {
 				refs, err := imagesToCopyFromUpstream(ctx, upstreamAddr, r, upstreamCtx, regCfg.Content[contentID], log)
-			for _, ref := range refs {
-				images = append(images, struct {
+				if err != nil {
+					return err
+				}
+
+				for _, ref := range refs[repo] {
+					reposWithContentID[repo] = append(reposWithContentID[repo], struct {
 						ref     types.ImageReference
 						content Content
 					}{
@@ -343,8 +355,9 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 						content: regCfg.Content[contentID],
 					})
 				}
+			}
 
-			return err
+			return nil
 		}, retryOptions); err != nil {
 			log.Error().Err(err).Msg("error while getting images references from upstream, retrying...")
 
@@ -352,7 +365,7 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 		}
 	}
 
-	for _, image := range images {
+	for remoteRepo, imageList := range reposWithContentID {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -360,10 +373,23 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 			break
 		}
 
-		upstreamImageRef := image.ref
+		remoteRepoCopy := remoteRepo
+		imageStore := storeController.GetImageStore(remoteRepoCopy)
 
-		remoteRepo := getRepoFromRef(upstreamImageRef, upstreamAddr)
-		localRepo := getRepoDestination(remoteRepo, image.content)
+		localCachePath, err := getLocalCachePath(imageStore, remoteRepoCopy)
+		if err != nil {
+			log.Error().Err(err).Msgf("couldn't get localCachePath for %s", remoteRepoCopy)
+
+			return err
+		}
+
+		if localCachePath != "" {
+			defer os.RemoveAll(localCachePath)
+		}
+
+		for _, image := range imageList {
+			localRepo := remoteRepoCopy
+			upstreamImageRef := image.ref
 
 			upstreamImageDigest, err := docker.GetDigest(ctx, upstreamCtx, upstreamImageRef)
 			if err != nil {
@@ -373,11 +399,8 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 			}
 
 			tag := getTagFromRef(upstreamImageRef, log).Tag()
-
-		imageStore := storeController.GetImageStore(localRepo)
-
 			// get upstream signatures
-		cosignManifest, err := getCosignManifest(httpClient, *registryURL, remoteRepo,
+			cosignManifest, err := getCosignManifest(httpClient, *registryURL, remoteRepoCopy,
 				upstreamImageDigest.String(), log)
 			if err != nil && !errors.Is(err, zerr.ErrSyncSignatureNotFound) {
 				log.Error().Err(err).Msgf("couldn't get upstream image %s cosign manifest", upstreamImageRef.DockerReference())
@@ -385,7 +408,7 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 				return err
 			}
 
-		refs, err := getNotaryRefs(httpClient, *registryURL, remoteRepo, upstreamImageDigest.String(), log)
+			refs, err := getNotaryRefs(httpClient, *registryURL, remoteRepoCopy, upstreamImageDigest.String(), log)
 			if err != nil && !errors.Is(err, zerr.ErrSyncSignatureNotFound) {
 				log.Error().Err(err).Msgf("couldn't get upstream image %s notary references", upstreamImageRef.DockerReference())
 
@@ -424,7 +447,7 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 
 				if !skipNotarySig {
 					if err = retry.RetryIfNecessary(ctx, func() error {
-					err = syncNotarySignature(httpClient, imageStore, *registryURL, localRepo, remoteRepo,
+						err = syncNotarySignature(httpClient, imageStore, *registryURL, localRepo, remoteRepoCopy,
 							upstreamImageDigest.String(), refs, log)
 
 						return err
@@ -442,7 +465,7 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 
 				if !skipCosignSig {
 					if err = retry.RetryIfNecessary(ctx, func() error {
-					err = syncCosignSignature(httpClient, imageStore, *registryURL, localRepo, remoteRepo,
+						err = syncCosignSignature(httpClient, imageStore, *registryURL, localRepo, remoteRepoCopy,
 							upstreamImageDigest.String(), cosignManifest, log)
 
 						return err
@@ -454,7 +477,7 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 				continue
 			}
 
-		localImageRef, localCachePath, err := getLocalImageRef(imageStore, localRepo, tag)
+			localImageRef, err := getLocalImageRef(localCachePath, localRepo, tag)
 			if err != nil {
 				log.Error().Err(err).Msgf("couldn't obtain a valid image reference for reference %s/%s:%s",
 					localCachePath, localRepo, tag)
@@ -462,8 +485,6 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 				return err
 			}
 
-		defer os.RemoveAll(localCachePath)
-
 			log.Info().Msgf("copying image %s to %s", upstreamImageRef.DockerReference(), localCachePath)
 
 			if err = retry.RetryIfNecessary(ctx, func() error {
@@ -476,8 +497,9 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 
 				return err
 			}
-
+			// push from cache to repo
 			err = pushSyncedLocalImage(localRepo, tag, localCachePath, imageStore, log)
+
 			if err != nil {
 				log.Error().Err(err).Msgf("error while pushing synced cached image %s",
 					fmt.Sprintf("%s/%s:%s", localCachePath, localRepo, tag))
@@ -485,24 +507,28 @@ func syncRegistry(ctx context.Context, regCfg RegistryConfig, upstreamURL string
 				return err
 			}
 
+			refs, err = getNotaryRefs(httpClient, *registryURL, remoteRepoCopy, upstreamImageDigest.String(), log)
 			if err = retry.RetryIfNecessary(ctx, func() error {
-			err = syncNotarySignature(httpClient, imageStore, *registryURL, localRepo, remoteRepo, upstreamImageDigest.String(),
-				refs, log)
+				err = syncNotarySignature(httpClient, imageStore, *registryURL, localRepo,
+					remoteRepoCopy, upstreamImageDigest.String(), refs, log)
 
 				return err
 			}, retryOptions); err != nil {
 				log.Error().Err(err).Msgf("couldn't copy notary signature for %s", upstreamImageRef.DockerReference())
 			}
 
+			cosignManifest, err = getCosignManifest(httpClient, *registryURL, remoteRepoCopy,
+				upstreamImageDigest.String(), log)
 			if err = retry.RetryIfNecessary(ctx, func() error {
-			err = syncCosignSignature(httpClient, imageStore, *registryURL, localRepo, remoteRepo, upstreamImageDigest.String(),
-				cosignManifest, log)
+				err = syncCosignSignature(httpClient, imageStore, *registryURL, localRepo,
+					remoteRepoCopy, upstreamImageDigest.String(), cosignManifest, log)
 
 				return err
 			}, retryOptions); err != nil {
 				log.Error().Err(err).Msgf("couldn't copy cosign signature for %s", upstreamImageRef.DockerReference())
 			}
 		}
+	}
 
 	log.Info().Msgf("finished syncing %s", upstreamAddr)
 

pkg/extensions/sync/sync_internal_test.go

@@ -61,11 +61,10 @@ func TestInjectSyncUtils(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-
 		imageStore := storage.NewImageStore(t.TempDir(), false, storage.DefaultGCDelay, false, false, log, metrics)
-
 		injected = test.InjectFailure(0)
-		_, _, err = getLocalImageRef(imageStore, testImage, testImageTag)
+
+		_, err = getLocalCachePath(imageStore, testImage)
 		if injected {
 			So(err, ShouldNotBeNil)
 		} else {
@@ -154,7 +153,7 @@ func TestSyncInternal(t *testing.T) {
 		So(err, ShouldNotBeNil)
 	})
 
-	Convey("Verify getLocalImageRef()", t, func() {
+	Convey("Verify getLocalImageRef() and getLocalCachePath()", t, func() {
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
 
@@ -163,13 +162,36 @@ func TestSyncInternal(t *testing.T) {
 		err := os.Chmod(imageStore.RootDir(), 0o000)
 		So(err, ShouldBeNil)
 
-		_, _, err = getLocalImageRef(imageStore, testImage, testImageTag)
+		localCachePath, err := getLocalCachePath(imageStore, testImage)
+		So(err, ShouldNotBeNil)
+
+		_, err = getLocalImageRef(localCachePath, testImage, testImageTag)
+		So(err, ShouldNotBeNil)
+
+		err = os.Chmod(imageStore.RootDir(), 0o544)
+		So(err, ShouldBeNil)
+
+		_, err = getLocalCachePath(imageStore, testImage)
 		So(err, ShouldNotBeNil)
 
 		err = os.Chmod(imageStore.RootDir(), 0o755)
 		So(err, ShouldBeNil)
 
-		_, _, err = getLocalImageRef(imageStore, "zot][]321", "tag_tag][]")
+		localCachePath, err = getLocalCachePath(imageStore, testImage)
+		So(err, ShouldBeNil)
+
+		testPath, _ := path.Split(localCachePath)
+
+		err = os.Chmod(testPath, 0o544)
+		So(err, ShouldBeNil)
+
+		_, err = getLocalCachePath(imageStore, testImage)
+		So(err, ShouldNotBeNil)
+
+		err = os.Chmod(testPath, 0o755)
+		So(err, ShouldBeNil)
+
+		_, err = getLocalImageRef(localCachePath, "zot][]321", "tag_tag][]")
 		So(err, ShouldNotBeNil)
 	})
 
@@ -471,15 +493,41 @@ func TestSyncInternal(t *testing.T) {
 			panic(err)
 		}
 
-		manifestConfigPath := path.Join(imageStore.RootDir(), testImage, "blobs", "sha256", manifest.Config.Digest.Hex())
-		if err := os.MkdirAll(manifestConfigPath, 0o000); err != nil {
+		cachedManifestBackup, err := os.ReadFile(cachedManifestConfigPath)
+		if err != nil {
+			panic(err)
+		}
+
+		configDigestBackup := manifest.Config.Digest
+		manifest.Config.Digest = "not what it needs to be"
+		manifestBuf, err := json.Marshal(manifest)
+		if err != nil {
+			panic(err)
+		}
+
+		if err = os.WriteFile(cachedManifestConfigPath, manifestBuf, 0o600); err != nil {
+			panic(err)
+		}
+
+		if err = os.Chmod(cachedManifestConfigPath, 0o755); err != nil {
 			panic(err)
 		}
 
 		err = pushSyncedLocalImage(testImage, testImageTag, testRootDir, imageStore, log)
 		So(err, ShouldNotBeNil)
 
-		if err := os.Remove(manifestConfigPath); err != nil {
+		manifest.Config.Digest = configDigestBackup
+		manifestBuf = cachedManifestBackup
+
+		if err := os.Remove(cachedManifestConfigPath); err != nil {
+			panic(err)
+		}
+
+		if err = os.WriteFile(cachedManifestConfigPath, manifestBuf, 0o600); err != nil {
+			panic(err)
+		}
+
+		if err = os.Chmod(cachedManifestConfigPath, 0o755); err != nil {
 			panic(err)
 		}
 

pkg/extensions/sync/sync_test.go

@@ -3295,8 +3295,8 @@ func TestSyncOnlyDiff(t *testing.T) {
 				case <-done:
 					return
 				default:
-					_, err := os.ReadDir(path.Join(destDir, testImage, ".sync"))
-					if err == nil {
+					fileList, _ := os.ReadDir(path.Join(destDir, testImage, ".sync"))
+					if len(fileList) > 0 {
 						isPopulated = true
 					}
 					time.Sleep(200 * time.Millisecond)

pkg/extensions/sync/utils.go

@@ -44,14 +44,6 @@ func getTagFromRef(ref types.ImageReference, log log.Logger) reference.Tagged {
 	return tagged
 }
 
-// getRepoFromRef returns repo name from a registry ImageReference.
-func getRepoFromRef(ref types.ImageReference, registryDomain string) string {
-	imageName := strings.Replace(ref.DockerReference().Name(), registryDomain, "", 1)
-	imageName = strings.TrimPrefix(imageName, "/")
-
-	return imageName
-}
-
 // parseRepositoryReference parses input into a reference.Named, and verifies that it names a repository, not an image.
 func parseRepositoryReference(input string) (reference.Named, error) {
 	ref, err := reference.ParseNormalizedNamed(input)
@@ -296,6 +288,7 @@ func pushSyncedLocalImage(localRepo, tag, localCachePath string,
 			return err
 		}
 
+		if found, _, _ := imageStore.CheckBlob(localRepo, blob.Digest.String()); !found {
 			_, _, err = imageStore.FullBlobUpload(localRepo, blobReader, blob.Digest.String())
 			if err != nil {
 				log.Error().Err(err).Str("blob digest", blob.Digest.String()).Msg("couldn't upload blob")
@@ -303,6 +296,7 @@ func pushSyncedLocalImage(localRepo, tag, localCachePath string,
 				return err
 			}
 		}
+	}
 
 	blobReader, _, err := cacheImageStore.GetBlob(localRepo, manifest.Config.Digest.String(), manifest.Config.MediaType)
 	if err != nil {
@@ -312,12 +306,14 @@ func pushSyncedLocalImage(localRepo, tag, localCachePath string,
 		return err
 	}
 
+	if found, _, _ := imageStore.CheckBlob(localRepo, manifest.Config.Digest.String()); !found {
 		_, _, err = imageStore.FullBlobUpload(localRepo, blobReader, manifest.Config.Digest.String())
 		if err != nil {
 			log.Error().Err(err).Str("blob digest", manifest.Config.Digest.String()).Msg("couldn't upload config blob")
 
 			return err
 		}
+	}
 
 	_, err = imageStore.PutImageManifest(localRepo, tag, ispec.MediaTypeImageManifest, manifestContent)
 	if err != nil {
@@ -326,14 +322,6 @@ func pushSyncedLocalImage(localRepo, tag, localCachePath string,
 		return err
 	}
 
-	log.Info().Msgf("removing temporary cached synced repo %s", path.Join(cacheImageStore.RootDir(), localRepo))
-
-	if err := os.RemoveAll(cacheImageStore.RootDir()); err != nil {
-		log.Error().Err(err).Msg("couldn't remove locally cached sync repo")
-
-		return err
-	}
-
 	return nil
 }
 
@@ -364,17 +352,9 @@ func getImageRef(registryDomain, repo, tag string) (types.ImageReference, error)
 }
 
 // get a local ImageReference used to temporary store one synced image.
-func getLocalImageRef(imageStore storage.ImageStore, repo, tag string) (types.ImageReference, string, error) {
-	uuid, err := guuid.NewV4()
-	// hard to reach test case, injected error, see pkg/test/dev.go
-	if err := test.Error(err); err != nil {
-		return nil, "", err
-	}
-
-	localCachePath := path.Join(imageStore.RootDir(), repo, SyncBlobUploadDir, uuid.String())
-
-	if err = os.MkdirAll(path.Join(localCachePath, repo), storage.DefaultDirPerms); err != nil {
-		return nil, "", err
+func getLocalImageRef(localCachePath, repo, tag string) (types.ImageReference, error) {
+	if _, err := os.ReadDir(localCachePath); err != nil {
+		return nil, err
 	}
 
 	localRepo := path.Join(localCachePath, repo)
@@ -382,10 +362,42 @@ func getLocalImageRef(imageStore storage.ImageStore, repo, tag string) (types.Im
 
 	localImageRef, err := layout.ParseReference(localTaggedRepo)
 	if err != nil {
-		return nil, "", err
+		return nil, err
 	}
 
-	return localImageRef, localCachePath, nil
+	return localImageRef, nil
+}
+
+// Returns the localCachePath with an UUID at the end. Only to be called once per repo.
+func getLocalCachePath(imageStore storage.ImageStore, repo string) (string, error) {
+	localRepoPath := path.Join(imageStore.RootDir(), repo, SyncBlobUploadDir)
+	// check if SyncBlobUploadDir exists, create if not
+	var err error
+	if _, err = os.ReadDir(localRepoPath); os.IsNotExist(err) {
+		if err = os.MkdirAll(localRepoPath, storage.DefaultDirPerms); err != nil {
+			return "", err
+		}
+	}
+
+	if err != nil {
+		return "", err
+	}
+
+	// create uuid folder
+	uuid, err := guuid.NewV4()
+	// hard to reach test case, injected error, see pkg/test/dev.go
+	if err := test.Error(err); err != nil {
+		return "", err
+	}
+
+	localCachePath := path.Join(localRepoPath, uuid.String())
+
+	cachedRepoPath := path.Join(localCachePath, repo)
+	if err = os.MkdirAll(cachedRepoPath, storage.DefaultDirPerms); err != nil {
+		return "", err
+	}
+
+	return localCachePath, nil
 }
 
 // canSkipImage returns whether or not we already synced this image.