From b5a3d79bce9fa20f56cb22ed6808914b34daff35 Mon Sep 17 00:00:00 2001 From: Klaus Post Date: Fri, 13 Nov 2020 16:58:20 -0800 Subject: [PATCH] listobjectversions: Add shortcut for Veeam blocks (#10893) Add shortcut for `APN/1.0 Veeam/1.0 Backup/10.0` It requests unique blocks with a specific prefix. We skip scanning the parent directory for more objects matching the prefix. --- cmd/erasure-server-sets.go | 18 ++++++++++++++++-- cmd/metacache-server-sets.go | 4 ++++ cmd/metacache-set.go | 9 ++++++++- cmd/metacache-walk.go | 10 ++++++---- 4 files changed, 34 insertions(+), 7 deletions(-) diff --git a/cmd/erasure-server-sets.go b/cmd/erasure-server-sets.go index 21386f8ae..bd7bb0eb5 100644 --- a/cmd/erasure-server-sets.go +++ b/cmd/erasure-server-sets.go @@ -24,6 +24,7 @@ import ( "math/rand" "net/http" "strconv" + "strings" "sync" "time" @@ -662,7 +663,8 @@ func (z *erasureServerSets) ListObjectVersions(ctx context.Context, bucket, pref if marker == "" && versionMarker != "" { return loi, NotImplemented{} } - merged, err := z.listPath(ctx, listPathOptions{ + + opts := listPathOptions{ Bucket: bucket, Prefix: prefix, Separator: delimiter, @@ -670,7 +672,19 @@ func (z *erasureServerSets) ListObjectVersions(ctx context.Context, bucket, pref Marker: marker, InclDeleted: true, AskDisks: globalAPIConfig.getListQuorum(), - }) + } + + // Shortcut for APN/1.0 Veeam/1.0 Backup/10.0 + // It requests unique blocks with a specific prefix. + // We skip scanning the parent directory for + // more objects matching the prefix. + ri := logger.GetReqInfo(ctx) + if ri != nil && strings.Contains(ri.UserAgent, `1.0 Veeam/1.0 Backup`) && strings.HasSuffix(prefix, ".blk") { + opts.singleObject = true + opts.Transient = true + } + + merged, err := z.listPath(ctx, opts) if err != nil && err != io.EOF { return loi, err } diff --git a/cmd/metacache-server-sets.go b/cmd/metacache-server-sets.go index 468ba49b0..a779865af 100644 --- a/cmd/metacache-server-sets.go +++ b/cmd/metacache-server-sets.go @@ -83,6 +83,10 @@ func (z *erasureServerSets) listPath(ctx context.Context, o listPathOptions) (en o.ID = mustGetUUID() } o.BaseDir = baseDirFromPrefix(o.Prefix) + if o.singleObject { + // Override for single object. + o.BaseDir = o.Prefix + } var cache metacache // If we don't have a list id we must ask the server if it has a cache or create a new. diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go index 33807e3fd..2c237b7ed 100644 --- a/cmd/metacache-set.go +++ b/cmd/metacache-set.go @@ -87,6 +87,9 @@ type listPathOptions struct { // This means the cache metadata will not be persisted on disk. // A transient result will never be returned from the cache so knowing the list id is required. Transient bool + + // singleObject will assume that prefix refers to an exact single object. + singleObject bool } func init() { @@ -542,7 +545,7 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions) (entr console.Printf("listPath with options: %#v\n", o) } // See if we have the listing stored. - if !o.Create { + if !o.Create && !o.singleObject { entries, err := er.streamMetadataParts(ctx, o) switch err { case nil, io.EOF, context.Canceled, context.DeadlineExceeded: @@ -662,6 +665,10 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions) (entr // Write results to disk. bw := newMetacacheBlockWriter(cacheCh, func(b *metacacheBlock) error { + if o.singleObject { + // Don't save single object listings. + return nil + } if debugPrint { console.Println("listPath: saving block", b.n, "to", o.objectPath(b.n)) } diff --git a/cmd/metacache-walk.go b/cmd/metacache-walk.go index aad7b957c..6d2a755f7 100644 --- a/cmd/metacache-walk.go +++ b/cmd/metacache-walk.go @@ -116,13 +116,14 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ // If root was an object return it as such. if HasSuffix(entry, xlStorageFormatFile) { var meta metaCacheEntry - meta.metadata, err = ioutil.ReadFile(pathJoin(volumeDir, meta.name, xlStorageFormatFile)) + meta.metadata, err = ioutil.ReadFile(pathJoin(volumeDir, current, entry)) if err != nil { logger.LogIf(ctx, err) continue } - meta.name = strings.TrimSuffix(meta.name, xlStorageFormatFile) + meta.name = strings.TrimSuffix(entry, xlStorageFormatFile) meta.name = strings.TrimSuffix(meta.name, SlashSeparator) + meta.name = pathJoin(current, meta.name) meta.name = decodeDirObject(meta.name) out <- meta return nil @@ -130,13 +131,14 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ // Check legacy. if HasSuffix(entry, xlStorageFormatFileV1) { var meta metaCacheEntry - meta.metadata, err = ioutil.ReadFile(pathJoin(volumeDir, meta.name, xlStorageFormatFileV1)) + meta.metadata, err = ioutil.ReadFile(pathJoin(volumeDir, current, entry)) if err != nil { logger.LogIf(ctx, err) continue } - meta.name = strings.TrimSuffix(meta.name, xlStorageFormatFileV1) + meta.name = strings.TrimSuffix(entry, xlStorageFormatFileV1) meta.name = strings.TrimSuffix(meta.name, SlashSeparator) + meta.name = pathJoin(current, meta.name) out <- meta return nil }