Add Veeam storage class override (#19748)

Recent Veeam is very picky about storage class names. Add `_MINIO_VEEAM_FORCE_SC` env var.

It will override the storage class returned by the storage backend if it is non-standard
and we detect a Veeam client by checking the User Agent.

Applies to HeadObject/GetObject/ListObject*

cmd/api-headers.go

@@ -19,6 +19,7 @@ package cmd
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"encoding/xml"
 	"fmt"
@@ -107,7 +108,7 @@ func setPartsCountHeaders(w http.ResponseWriter, objInfo ObjectInfo) {
 }
 
 // Write object header
-func setObjectHeaders(w http.ResponseWriter, objInfo ObjectInfo, rs *HTTPRangeSpec, opts ObjectOptions) (err error) {
+func setObjectHeaders(ctx context.Context, w http.ResponseWriter, objInfo ObjectInfo, rs *HTTPRangeSpec, opts ObjectOptions) (err error) {
 	// set common headers
 	setCommonHeaders(w)
 
@@ -212,7 +213,7 @@ func setObjectHeaders(w http.ResponseWriter, objInfo ObjectInfo, rs *HTTPRangeSp
 	if objInfo.IsRemote() {
 		// Check if object is being restored. For more information on x-amz-restore header see
 		// https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_ResponseSyntax
-		w.Header()[xhttp.AmzStorageClass] = []string{objInfo.TransitionedObject.Tier}
+		w.Header()[xhttp.AmzStorageClass] = []string{filterStorageClass(ctx, objInfo.TransitionedObject.Tier)}
 	}
 
 	if lc, err := globalLifecycleSys.Get(objInfo.Bucket); err == nil {

cmd/api-response.go

@@ -544,7 +544,7 @@ func cleanReservedKeys(metadata map[string]string) map[string]string {
 }
 
 // generates an ListBucketVersions response for the said bucket with other enumerated options.
-func generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delimiter, encodingType string, maxKeys int, resp ListObjectVersionsInfo, metadata metaCheckFn) ListVersionsResponse {
+func generateListVersionsResponse(ctx context.Context, bucket, prefix, marker, versionIDMarker, delimiter, encodingType string, maxKeys int, resp ListObjectVersionsInfo, metadata metaCheckFn) ListVersionsResponse {
 	versions := make([]ObjectVersion, 0, len(resp.Objects))
 
 	owner := &Owner{
@@ -573,7 +573,7 @@ func generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delim
 		}
 		content.Size = object.Size
 		if object.StorageClass != "" {
-			content.StorageClass = object.StorageClass
+			content.StorageClass = filterStorageClass(ctx, object.StorageClass)
 		} else {
 			content.StorageClass = globalMinioDefaultStorageClass
 		}
@@ -634,7 +634,7 @@ func generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delim
 }
 
 // generates an ListObjectsV1 response for the said bucket with other enumerated options.
-func generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType string, maxKeys int, resp ListObjectsInfo) ListObjectsResponse {
+func generateListObjectsV1Response(ctx context.Context, bucket, prefix, marker, delimiter, encodingType string, maxKeys int, resp ListObjectsInfo) ListObjectsResponse {
 	contents := make([]Object, 0, len(resp.Objects))
 	owner := &Owner{
 		ID:          globalMinioDefaultOwnerID,
@@ -654,7 +654,7 @@ func generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingTy
 		}
 		content.Size = object.Size
 		if object.StorageClass != "" {
-			content.StorageClass = object.StorageClass
+			content.StorageClass = filterStorageClass(ctx, object.StorageClass)
 		} else {
 			content.StorageClass = globalMinioDefaultStorageClass
 		}
@@ -683,7 +683,7 @@ func generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingTy
 }
 
 // generates an ListObjectsV2 response for the said bucket with other enumerated options.
-func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter, delimiter, encodingType string, fetchOwner, isTruncated bool, maxKeys int, objects []ObjectInfo, prefixes []string, metadata metaCheckFn) ListObjectsV2Response {
+func generateListObjectsV2Response(ctx context.Context, bucket, prefix, token, nextToken, startAfter, delimiter, encodingType string, fetchOwner, isTruncated bool, maxKeys int, objects []ObjectInfo, prefixes []string, metadata metaCheckFn) ListObjectsV2Response {
 	contents := make([]Object, 0, len(objects))
 	var owner *Owner
 	if fetchOwner {
@@ -707,7 +707,7 @@ func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter,
 		}
 		content.Size = object.Size
 		if object.StorageClass != "" {
-			content.StorageClass = object.StorageClass
+			content.StorageClass = filterStorageClass(ctx, object.StorageClass)
 		} else {
 			content.StorageClass = globalMinioDefaultStorageClass
 		}

cmd/bucket-listobjects-handlers.go

@@ -124,7 +124,7 @@ func (api objectAPIHandlers) listObjectVersionsHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
-	response := generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delimiter, encodingType, maxkeys, listObjectVersionsInfo, checkObjMeta)
+	response := generateListVersionsResponse(ctx, bucket, prefix, marker, versionIDMarker, delimiter, encodingType, maxkeys, listObjectVersionsInfo, checkObjMeta)
 
 	// Write success response.
 	writeSuccessResponseXML(w, encodeResponseList(response))
@@ -219,7 +219,7 @@ func (api objectAPIHandlers) listObjectsV2Handler(ctx context.Context, w http.Re
 		return
 	}
 
-	response := generateListObjectsV2Response(bucket, prefix, token, listObjectsV2Info.NextContinuationToken, startAfter,
+	response := generateListObjectsV2Response(ctx, bucket, prefix, token, listObjectsV2Info.NextContinuationToken, startAfter,
 		delimiter, encodingType, fetchOwner, listObjectsV2Info.IsTruncated,
 		maxKeys, listObjectsV2Info.Objects, listObjectsV2Info.Prefixes, checkObjMeta)
 
@@ -318,7 +318,7 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 		return
 	}
 
-	response := generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType, maxKeys, listObjectsInfo)
+	response := generateListObjectsV1Response(ctx, bucket, prefix, marker, delimiter, encodingType, maxKeys, listObjectsInfo)
 
 	// Write success response.
 	writeSuccessResponseXML(w, encodeResponseList(response))

cmd/erasure-server-pool.go

@@ -1387,8 +1387,7 @@ func (z *erasureServerPools) ListObjectVersions(ctx context.Context, bucket, pre
 	// 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 isVeeamClient(ctx) && strings.HasSuffix(prefix, ".blk") {
-	if ri != nil && strings.Contains(ri.UserAgent, `1.0 Veeam/1.0 Backup`) && strings.HasSuffix(prefix, ".blk") {
 		opts.BaseDir = prefix
 		opts.Transient = true
 	}

cmd/object-handlers.go

@@ -647,7 +647,7 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 		}()
 	}
 
-	if err = setObjectHeaders(w, objInfo, rs, opts); err != nil {
+	if err = setObjectHeaders(ctx, w, objInfo, rs, opts); err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
@@ -786,7 +786,7 @@ func (api objectAPIHandlers) getObjectAttributesHandler(ctx context.Context, obj
 	}
 
 	if _, ok := opts.ObjectAttributes[xhttp.StorageClass]; ok {
-		OA.StorageClass = objInfo.StorageClass
+		OA.StorageClass = filterStorageClass(ctx, objInfo.StorageClass)
 	}
 
 	objInfo.decryptPartsChecksums()
@@ -1173,7 +1173,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 	}
 
 	// Set standard object headers.
-	if err = setObjectHeaders(w, objInfo, rs, opts); err != nil {
+	if err = setObjectHeaders(ctx, w, objInfo, rs, opts); err != nil {
 		writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
 		return
 	}

cmd/s3-zip-handlers.go

@@ -202,7 +202,7 @@ func (api objectAPIHandlers) getObjectInArchiveFileHandler(ctx context.Context,
 
 	defer rc.Close()
 
-	if err = setObjectHeaders(w, fileObjInfo, nil, opts); err != nil {
+	if err = setObjectHeaders(ctx, w, fileObjInfo, nil, opts); err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
@@ -470,7 +470,7 @@ func (api objectAPIHandlers) headObjectInArchiveFileHandler(ctx context.Context,
 	}
 
 	// Set standard object headers.
-	if err = setObjectHeaders(w, objInfo, nil, opts); err != nil {
+	if err = setObjectHeaders(ctx, w, objInfo, nil, opts); err != nil {
 		writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
 		return
 	}

cmd/utils.go

@@ -47,6 +47,7 @@ import (
 	"github.com/minio/minio/internal/config"
 	"github.com/minio/minio/internal/config/api"
 	xtls "github.com/minio/minio/internal/config/identity/tls"
+	"github.com/minio/minio/internal/config/storageclass"
 	"github.com/minio/minio/internal/fips"
 	"github.com/minio/minio/internal/handlers"
 	"github.com/minio/minio/internal/hash"
@@ -1142,3 +1143,11 @@ type itemOrErr[V any] struct {
 	Item V
 	Err  error
 }
+
+func filterStorageClass(ctx context.Context, s string) string {
+	// Veeam 14.0 and later clients are not compatible with custom storage classes.
+	if globalVeeamForceSC != "" && s != storageclass.STANDARD && s != storageclass.RRS && isVeeamClient(ctx) {
+		return globalVeeamForceSC
+	}
+	return s
+}

cmd/veeam-sos-api.go

@@ -22,8 +22,11 @@ import (
 	"context"
 	"encoding/xml"
 	"io"
+	"os"
+	"strings"
 
 	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/logger"
 )
 
 // From Veeam-SOSAPI_1.0_Document_v1.02d.pdf
@@ -83,6 +86,11 @@ type apiEndpoints struct {
 	STSEndpoint string `xml:"STSEndpoint"`
 }
 
+// globalVeeamForceSC is set by the environment variable _MINIO_VEEAM_FORCE_SC
+// This will override the storage class returned by the storage backend if it is non-standard
+// and we detect a Veeam client by checking the User Agent.
+var globalVeeamForceSC = os.Getenv("_MINIO_VEEAM_FORCE_SC")
+
 type systemInfo struct {
 	XMLName              xml.Name `xml:"SystemInfo" json:"-"`
 	ProtocolVersion      string   `xml:"ProtocolVersion"`
@@ -115,6 +123,7 @@ type capacityInfo struct {
 const (
 	systemXMLObject   = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/system.xml"
 	capacityXMLObject = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/capacity.xml"
+	veeamAgentSubstr  = "1.0 Veeam/1.0 Backup"
 )
 
 func isVeeamSOSAPIObject(object string) bool {
@@ -126,6 +135,12 @@ func isVeeamSOSAPIObject(object string) bool {
 	}
 }
 
+// isVeeamClient - returns true if the request is from Veeam client.
+func isVeeamClient(ctx context.Context) bool {
+	ri := logger.GetReqInfo(ctx)
+	return ri != nil && strings.Contains(ri.UserAgent, veeamAgentSubstr)
+}
+
 func veeamSOSAPIHeadObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
 	gr, err := veeamSOSAPIGetObject(ctx, bucket, object, nil, opts)
 	if gr != nil {