Revert "fix: remove persistence layer for metacache store in memory (#11538)"

This reverts commit b23659927c.

cmd/metacache-bucket.go

@@ -17,6 +17,11 @@
 package cmd
 
 import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
 	"path"
 	"runtime/debug"
 	"sort"
@@ -24,8 +29,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/klauspost/compress/s2"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/console"
+	"github.com/minio/minio/pkg/hash"
+	"github.com/tinylib/msgp/msgp"
 )
 
 //go:generate msgp -file $GOFILE -unexported
@@ -49,7 +57,16 @@ type bucketMetacache struct {
 
 // newBucketMetacache creates a new bucketMetacache.
 // Optionally remove all existing caches.
-func newBucketMetacache(bucket string) *bucketMetacache {
+func newBucketMetacache(bucket string, cleanup bool) *bucketMetacache {
+	if cleanup {
+		// Recursively delete all caches.
+		objAPI := newObjectLayerFn()
+		ez, ok := objAPI.(*erasureServerPools)
+		if ok {
+			ctx := context.Background()
+			ez.renameAll(ctx, minioMetaBucket, metacachePrefixForID(bucket, slashSeparator))
+		}
+	}
 	return &bucketMetacache{
 		bucket:     bucket,
 		caches:     make(map[string]metacache, 10),
@@ -63,6 +80,111 @@ func (b *bucketMetacache) debugf(format string, data ...interface{}) {
 	}
 }
 
+// loadBucketMetaCache will load the cache from the object layer.
+// If the cache cannot be found a new one is created.
+func loadBucketMetaCache(ctx context.Context, bucket string) (*bucketMetacache, error) {
+	objAPI := newObjectLayerFn()
+	for objAPI == nil {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			time.Sleep(250 * time.Millisecond)
+		}
+		objAPI = newObjectLayerFn()
+		if objAPI == nil {
+			logger.LogIf(ctx, fmt.Errorf("loadBucketMetaCache: object layer not ready. bucket: %q", bucket))
+		}
+	}
+
+	var meta bucketMetacache
+	var decErr error
+	// Use global context for this.
+	r, err := objAPI.GetObjectNInfo(GlobalContext, minioMetaBucket, pathJoin("buckets", bucket, ".metacache", "index.s2"), nil, http.Header{}, readLock, ObjectOptions{})
+	if err == nil {
+		dec := s2DecPool.Get().(*s2.Reader)
+		dec.Reset(r)
+		decErr = meta.DecodeMsg(msgp.NewReader(dec))
+		dec.Reset(nil)
+		r.Close()
+		s2DecPool.Put(dec)
+	}
+	if err != nil {
+		switch err.(type) {
+		case ObjectNotFound:
+			err = nil
+		case InsufficientReadQuorum:
+			// Cache is likely lost. Clean up and return new.
+			return newBucketMetacache(bucket, true), nil
+		default:
+			logger.LogIf(ctx, err)
+		}
+		return newBucketMetacache(bucket, false), err
+	}
+	if decErr != nil {
+		if errors.Is(err, context.Canceled) {
+			return newBucketMetacache(bucket, false), err
+		}
+		// Log the error, but assume the data is lost and return a fresh bucket.
+		// Otherwise a broken cache will never recover.
+		logger.LogIf(ctx, decErr)
+		return newBucketMetacache(bucket, true), nil
+	}
+	// Sanity check...
+	if meta.bucket != bucket {
+		logger.Info("loadBucketMetaCache: loaded cache name mismatch, want %s, got %s. Discarding.", bucket, meta.bucket)
+		return newBucketMetacache(bucket, true), nil
+	}
+	meta.cachesRoot = make(map[string][]string, len(meta.caches)/10)
+	// Index roots
+	for id, cache := range meta.caches {
+		meta.cachesRoot[cache.root] = append(meta.cachesRoot[cache.root], id)
+	}
+	return &meta, nil
+}
+
+// save the bucket cache to the object storage.
+func (b *bucketMetacache) save(ctx context.Context) error {
+	if b.transient {
+		return nil
+	}
+	objAPI := newObjectLayerFn()
+	if objAPI == nil {
+		return errServerNotInitialized
+	}
+
+	// Keep lock while we marshal.
+	// We need a write lock since we update 'updated'
+	b.mu.Lock()
+	if !b.updated {
+		b.mu.Unlock()
+		return nil
+	}
+	// Save as s2 compressed msgpack
+	tmp := bytes.NewBuffer(make([]byte, 0, b.Msgsize()))
+	enc := s2.NewWriter(tmp)
+	err := msgp.Encode(enc, b)
+	if err != nil {
+		b.mu.Unlock()
+		return err
+	}
+	err = enc.Close()
+	if err != nil {
+		b.mu.Unlock()
+		return err
+	}
+	b.updated = false
+	b.mu.Unlock()
+
+	hr, err := hash.NewReader(tmp, int64(tmp.Len()), "", "", int64(tmp.Len()))
+	if err != nil {
+		return err
+	}
+	_, err = objAPI.PutObject(ctx, minioMetaBucket, pathJoin("buckets", b.bucket, ".metacache", "index.s2"), NewPutObjReader(hr), ObjectOptions{})
+	logger.LogIf(ctx, err)
+	return err
+}
+
 // findCache will attempt to find a matching cache for the provided options.
 // If a cache with the same ID exists already it will be returned.
 // If none can be found a new is created with the provided ID.
@@ -326,6 +448,41 @@ func (b *bucketMetacache) getCache(id string) *metacache {
 	return &c
 }
 
+// deleteAll will delete all on disk data for ALL caches.
+// Deletes are performed concurrently.
+func (b *bucketMetacache) deleteAll() {
+	ctx := context.Background()
+	ez, ok := newObjectLayerFn().(*erasureServerPools)
+	if !ok {
+		logger.LogIf(ctx, errors.New("bucketMetacache: expected objAPI to be *erasurePools"))
+		return
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.updated = true
+	if !b.transient {
+		// Delete all.
+		ez.renameAll(ctx, minioMetaBucket, metacachePrefixForID(b.bucket, slashSeparator))
+		b.caches = make(map[string]metacache, 10)
+		b.cachesRoot = make(map[string][]string, 10)
+		return
+	}
+
+	// Transient are in different buckets.
+	var wg sync.WaitGroup
+	for id := range b.caches {
+		wg.Add(1)
+		go func(cache metacache) {
+			defer wg.Done()
+			ez.renameAll(ctx, minioMetaBucket, metacachePrefixForID(cache.bucket, cache.id))
+		}(b.caches[id])
+	}
+	wg.Wait()
+	b.caches = make(map[string]metacache, 10)
+}
+
 // deleteCache will delete a specific cache and all files related to it across the cluster.
 func (b *bucketMetacache) deleteCache(id string) {
 	b.mu.Lock()
@@ -344,4 +501,7 @@ func (b *bucketMetacache) deleteCache(id string) {
 		b.updated = true
 	}
 	b.mu.Unlock()
+	if ok {
+		c.delete(context.Background())
+	}
 }