Merge pull request #221 from PerseidMeteor/feat-cache-version

feat: add cache version
goharbor · Oct 26, 2023 · 0a7e1a0 · 0a7e1a0
2 parents 95fbdcd + 4b0cdae
commit 0a7e1a0
Show file tree

Hide file tree

Showing 6 changed files with 37 additions and 13 deletions.
diff --git a/misc/config/config.nydus.ref.yaml b/misc/config/config.nydus.ref.yaml
@@ -34,6 +34,8 @@ provider:
       threshold: 1000MB
   # remote cache record capacity of converted layers, default is 200.
   cache_size: 200
+  # remote cache version, cache in remote must match the specified version, or discard cache.
+  cache_version: v1 
 
 converter:
   # number of worker for executing conversion task

diff --git a/misc/config/config.nydus.yaml b/misc/config/config.nydus.yaml
@@ -34,6 +34,8 @@ provider:
       threshold: 1000MB
   # remote cache record capacity of converted layers, default is 200.
   cache_size: 200
+  # remote cache version, cache in remote must match the specified version, or discard cache.
+  cache_version: v1 
 
 converter:
   # number of worker for executing conversion task

diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go
@@ -40,6 +40,8 @@ import (
 	"github.com/pkg/errors"
 )
 
+const LayerAnnotationCacheVersion = "containerd.io/snapshot/nydus-cache-version"
+
 type cacheKey struct{}
 
 type Item struct {
@@ -69,14 +71,17 @@ type RemoteCache struct {
 	size int
 	// newAdded records the number of new caches added in one conversion.
 	newAdded uint
+
+	version string
 }
 
-func New(ctx context.Context, ref string, size int, pvd Provider) (context.Context, *RemoteCache) {
+func New(ctx context.Context, ref, version string, size int, pvd Provider) (context.Context, *RemoteCache) {
 	cache := &RemoteCache{
 		Ref:      ref,
 		provider: pvd,
 		records:  make(map[digest.Digest]*Item),
 		size:     size,
+		version:  version,
 	}
 	cxt := context.WithValue(ctx, cacheKey{}, cache)
 	return cxt, cache
@@ -246,8 +251,12 @@ func (rc *RemoteCache) Fetch(ctx context.Context, platformMC platforms.MatchComp
 			if err != nil {
 				return nil, errors.Wrap(err, "read remote cache manifest")
 			}
+			if targetManifest.Annotations[LayerAnnotationCacheVersion] != rc.version {
+				logrus.WithError(err).Warnf("ignore cache %s, unmatched version: %s, expected: %s", rc.Ref,
+					targetManifest.Annotations[LayerAnnotationCacheVersion], rc.version)
+				continue
+			}
 			targetManifests = append(targetManifests, targetManifest)
-
 		}
 		for _, manifest := range targetManifests {
 			for _, targetDesc := range manifest.Layers {
@@ -386,6 +395,11 @@ func (rc *RemoteCache) update(ctx context.Context, orgDesc, newDesc, cacheDesc *
 						return nil, errors.Wrap(err, "read cache manifest")
 					}
 					manifest.Layers = appendLayers(manifest.Layers, layers, rc.size)
+					// append LayerAnnotationCacheVersion to manifest annotations
+					if manifest.Annotations == nil {
+						manifest.Annotations = map[string]string{}
+					}
+					manifest.Annotations[LayerAnnotationCacheVersion] = rc.version
 					newManiDesc, err := utils.WriteJSON(ctx, rc.provider.ContentStore(), manifest, maniDesc, "", nil)
 					if err != nil {
 						return nil, errors.Wrap(err, "write cache manifest")
@@ -406,6 +420,9 @@ func (rc *RemoteCache) update(ctx context.Context, orgDesc, newDesc, cacheDesc *
 			MediaType: ocispec.MediaTypeImageManifest,
 			Config:    *imageConfigDesc,
 			Layers:    layers,
+			Annotations: map[string]string{
+				LayerAnnotationCacheVersion: rc.version,
+			},
 		}
 		manifestDesc, err := utils.WriteJSON(ctx, rc.provider.ContentStore(), manifest, ocispec.Descriptor{}, "", nil)
 		if err != nil {

diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -48,10 +48,11 @@ type MetricConfig struct {
 }
 
 type ProviderConfig struct {
-	Source    map[string]SourceConfig `yaml:"source"`
-	WorkDir   string                  `yaml:"work_dir"`
-	GCPolicy  GCPolicy                `yaml:"gcpolicy"`
-	CacheSize int                     `yaml:"cache_size"`
+	Source       map[string]SourceConfig `yaml:"source"`
+	WorkDir      string                  `yaml:"work_dir"`
+	GCPolicy     GCPolicy                `yaml:"gcpolicy"`
+	CacheSize    int                     `yaml:"cache_size"`
+	CacheVersion string                  `yaml:"cache_version"`
 }
 
 type GCPolicy struct {

diff --git a/pkg/content/local.go b/pkg/content/local.go
@@ -40,6 +40,7 @@ type LocalProvider struct {
 	hosts        remote.HostFunc
 	platformMC   platforms.MatchComparer
 	cacheSize    int
+	cacheVersion string
 }
 
 func NewLocalProvider(cfg *config.Config, platformMC platforms.MatchComparer) (Provider, *Content, error) {
@@ -52,11 +53,12 @@ func NewLocalProvider(cfg *config.Config, platformMC platforms.MatchComparer) (P
 		return nil, nil, errors.Wrap(err, "create local provider content")
 	}
 	return &LocalProvider{
-		content:    content,
-		images:     make(map[string]*ocispec.Descriptor),
-		hosts:      cfg.Host,
-		platformMC: platformMC,
-		cacheSize:  cfg.Provider.CacheSize,
+		content:      content,
+		images:       make(map[string]*ocispec.Descriptor),
+		hosts:        cfg.Host,
+		platformMC:   platformMC,
+		cacheSize:    cfg.Provider.CacheSize,
+		cacheVersion: cfg.Provider.CacheVersion,
 	}, content, nil
 }
 
@@ -117,7 +119,7 @@ func (pvd *LocalProvider) ContentStore() content.Store {
 
 func (pvd *LocalProvider) NewRemoteCache(ctx context.Context, cacheRef string) (context.Context, *cache.RemoteCache) {
 	if cacheRef != "" {
-		return cache.New(ctx, cacheRef, pvd.cacheSize, pvd)
+		return cache.New(ctx, cacheRef, pvd.cacheVersion, pvd.cacheSize, pvd)
 	}
 	return ctx, nil
 }

diff --git a/pkg/converter/converter.go b/pkg/converter/converter.go
@@ -117,7 +117,7 @@ func (cvt *Converter) Convert(ctx context.Context, source, target, cacheRef stri
 				if errors.Is(err, ctrErrdefs.ErrNotFound) {
 					logger.Infof("cache %s not found", cacheRef)
 				} else {
-					logger.Warnf(errors.Wrapf(err, "failed to pull cache %s", cacheRef).Error())
+					logger.Warnf(errors.Wrapf(err, "fetch cache %s", cacheRef).Error())
 				}
 			}
 		}