Edward Klesel
e4c7aca6bd
When using Microsoft Entra ID credentials, Azure Blob server-side copy uses a user delegation SAS URL for the private copy source. The SAS start time was set to the current local time. Azure Storage validates the copy source from the service side, and small clock differences can make that SAS appear not yet valid. The service then returns 403 CannotVerifyCopySource with AuthenticationFailed. Start the copy-source SAS 15 minutes in the past, matching Microsoft SAS guidance for clock skew.
3370 lines
106 KiB
Go
3370 lines
106 KiB
Go
//go:build !plan9 && !js
|
|
|
|
// Package azureblob provides an interface to the Microsoft Azure blob object storage system
|
|
package azureblob
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/md5"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"maps"
|
|
"net/http"
|
|
"path"
|
|
"slices"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/Azure/azure-sdk-for-go/sdk/azcore"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas"
|
|
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
|
|
"github.com/rclone/rclone/backend/azureblob/arrowlist"
|
|
"github.com/rclone/rclone/backend/azureblob/auth"
|
|
"github.com/rclone/rclone/fs"
|
|
"github.com/rclone/rclone/fs/chunksize"
|
|
"github.com/rclone/rclone/fs/config"
|
|
"github.com/rclone/rclone/fs/config/configmap"
|
|
"github.com/rclone/rclone/fs/config/configstruct"
|
|
"github.com/rclone/rclone/fs/fserrors"
|
|
"github.com/rclone/rclone/fs/hash"
|
|
"github.com/rclone/rclone/fs/list"
|
|
"github.com/rclone/rclone/lib/atexit"
|
|
"github.com/rclone/rclone/lib/bucket"
|
|
"github.com/rclone/rclone/lib/encoder"
|
|
"github.com/rclone/rclone/lib/multipart"
|
|
"github.com/rclone/rclone/lib/pacer"
|
|
"github.com/rclone/rclone/lib/pool"
|
|
"github.com/rclone/rclone/lib/readers"
|
|
"github.com/rclone/rclone/lib/transferaccounter"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
const (
|
|
minSleep = 10 * time.Millisecond
|
|
maxSleep = 10 * time.Second
|
|
decayConstant = 1 // bigger for slower decay, exponential
|
|
maxListChunkSize = 5000 // number of items to read at once
|
|
modTimeKey = "mtime"
|
|
dirMetaKey = "hdi_isfolder"
|
|
dirMetaValue = "true"
|
|
timeFormatIn = time.RFC3339
|
|
timeFormatOut = "2006-01-02T15:04:05.000000000Z07:00"
|
|
storageDefaultBaseURL = "blob.core.windows.net"
|
|
defaultChunkSize = 4 * fs.Mebi
|
|
defaultAccessTier = blob.AccessTier("") // FIXME AccessTierNone
|
|
sasCopyValidity = time.Hour // how long SAS should last when doing server side copy
|
|
sasCopyStartSkew = 15 * time.Minute // how far back SAS should start to allow for clock skew
|
|
)
|
|
|
|
// setAcceptEncodingGzip is a per-call policy that sets Accept-Encoding: gzip
|
|
// on every request. This prevents the Go HTTP transport from automatically
|
|
// decompressing gzip-encoded blobs on download.
|
|
type setAcceptEncodingGzip struct{}
|
|
|
|
func (p setAcceptEncodingGzip) Do(req *policy.Request) (*http.Response, error) {
|
|
req.Raw().Header.Set("Accept-Encoding", "gzip")
|
|
return req.Next()
|
|
}
|
|
|
|
var (
|
|
errCantUpdateArchiveTierBlobs = fserrors.NoRetryError(errors.New("can't update archive tier blob without --azureblob-archive-tier-delete"))
|
|
|
|
// Take this when changing or reading metadata.
|
|
//
|
|
// It acts as global metadata lock so we don't bloat Object
|
|
// with an extra lock that will only very rarely be contended.
|
|
metadataMu sync.Mutex
|
|
)
|
|
|
|
// system metadata keys which this backend owns
|
|
var systemMetadataInfo = map[string]fs.MetadataHelp{
|
|
"cache-control": {
|
|
Help: "Cache-Control header",
|
|
Type: "string",
|
|
Example: "no-cache",
|
|
},
|
|
"content-disposition": {
|
|
Help: "Content-Disposition header",
|
|
Type: "string",
|
|
Example: "inline",
|
|
},
|
|
"content-encoding": {
|
|
Help: "Content-Encoding header",
|
|
Type: "string",
|
|
Example: "gzip",
|
|
},
|
|
"content-language": {
|
|
Help: "Content-Language header",
|
|
Type: "string",
|
|
Example: "en-US",
|
|
},
|
|
"content-type": {
|
|
Help: "Content-Type header",
|
|
Type: "string",
|
|
Example: "text/plain",
|
|
},
|
|
"tier": {
|
|
Help: "Tier of the object",
|
|
Type: "string",
|
|
Example: "Hot",
|
|
ReadOnly: true,
|
|
},
|
|
"mtime": {
|
|
Help: "Time of last modification, read from rclone metadata",
|
|
Type: "RFC 3339",
|
|
Example: "2006-01-02T15:04:05.999999999Z07:00",
|
|
},
|
|
}
|
|
|
|
// Register with Fs
|
|
func init() {
|
|
fs.Register(&fs.RegInfo{
|
|
Name: "azureblob",
|
|
Description: "Microsoft Azure Blob Storage",
|
|
NewFs: NewFs,
|
|
MetadataInfo: &fs.MetadataInfo{
|
|
System: systemMetadataInfo,
|
|
Help: `User metadata is stored as x-ms-meta- keys. Azure metadata keys are case insensitive and are always returned in lower case.`,
|
|
},
|
|
Options: slices.Concat(auth.ConfigOptions, []fs.Option{{
|
|
Name: "upload_cutoff",
|
|
Help: "Cutoff for switching to chunked upload (<= 256 MiB) (deprecated).",
|
|
Advanced: true,
|
|
}, {
|
|
Name: "chunk_size",
|
|
Help: `Upload chunk size.
|
|
|
|
Note that this is stored in memory and there may be up to
|
|
"--transfers" * "--azureblob-upload-concurrency" chunks stored at once
|
|
in memory.`,
|
|
Default: defaultChunkSize,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "upload_concurrency",
|
|
Help: `Concurrency for multipart uploads.
|
|
|
|
This is the number of chunks of the same file that are uploaded
|
|
concurrently.
|
|
|
|
If you are uploading small numbers of large files over high-speed
|
|
links and these uploads do not fully utilize your bandwidth, then
|
|
increasing this may help to speed up the transfers.
|
|
|
|
In tests, upload speed increases almost linearly with upload
|
|
concurrency. For example to fill a gigabit pipe it may be necessary to
|
|
raise this to 64. Note that this will use more memory.
|
|
|
|
Note that chunks are stored in memory and there may be up to
|
|
"--transfers" * "--azureblob-upload-concurrency" chunks stored at once
|
|
in memory.`,
|
|
Default: 16,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "copy_cutoff",
|
|
Help: `Cutoff for switching to multipart copy.
|
|
|
|
Any files larger than this that need to be server-side copied will be
|
|
copied in chunks of chunk_size using the put block list API.
|
|
|
|
Files smaller than this limit will be copied with the Copy Blob API.`,
|
|
Default: 8 * fs.Mebi,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "copy_concurrency",
|
|
Help: `Concurrency for multipart copy.
|
|
|
|
This is the number of chunks of the same file that are copied
|
|
concurrently.
|
|
|
|
These chunks are not buffered in memory and Microsoft recommends
|
|
setting this value to greater than 1000 in the azcopy documentation.
|
|
|
|
https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azcopy-optimize#increase-concurrency
|
|
|
|
In tests, copy speed increases almost linearly with copy
|
|
concurrency.`,
|
|
Default: 512,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "copy_total_concurrency",
|
|
Help: `Global concurrency limit for multipart copy chunks.
|
|
|
|
This limits the total number of multipart copy chunks running at once
|
|
across all files.
|
|
|
|
Set to 0 to disable this limiter.`,
|
|
Default: 0,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "use_copy_blob",
|
|
Help: `Whether to use the Copy Blob API when copying to the same storage account.
|
|
|
|
If true (the default) then rclone will use the Copy Blob API for
|
|
copies to the same storage account even when the size is above the
|
|
copy_cutoff.
|
|
|
|
Rclone assumes that the same storage account means the same config
|
|
and does not check for the same storage account in different configs.
|
|
|
|
There should be no need to change this value.
|
|
`,
|
|
Default: true,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "list_chunk",
|
|
Help: `Size of blob list.
|
|
|
|
This sets the number of blobs requested in each listing chunk. Default
|
|
is the maximum, 5000. "List blobs" requests are permitted 2 minutes
|
|
per megabyte to complete. If an operation is taking longer than 2
|
|
minutes per megabyte on average, it will time out (
|
|
[source](https://docs.microsoft.com/en-us/rest/api/storageservices/setting-timeouts-for-blob-service-operations#exceptions-to-default-timeout-interval)
|
|
). This can be used to limit the number of blobs items to return, to
|
|
avoid the time out.`,
|
|
Default: maxListChunkSize,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "use_arrow_list",
|
|
Help: `Use the experimental Apache Arrow listing format.
|
|
|
|
If set, directory listings are fetched using the ListBlobs Apache
|
|
Arrow response format instead of XML. This can be faster for very
|
|
large containers.
|
|
|
|
This is EXPERIMENTAL and requires the "Blob Listing with Apache Arrow"
|
|
preview feature to be enabled on the storage account. It is NOT
|
|
supported on accounts with a hierarchical namespace (ADLS Gen2) -
|
|
those return a 409 error. If the feature is not enabled the server
|
|
returns XML and the listing transparently falls back to the normal XML
|
|
path (logged at debug level).
|
|
|
|
Not supported with connection_string auth - falls back to normal
|
|
listing.`,
|
|
Default: false,
|
|
Advanced: true,
|
|
Hide: fs.OptionHideBoth,
|
|
}, {
|
|
Name: "list_parallelism",
|
|
Help: `Number of parallel shards to list a directory with.
|
|
|
|
EXPERIMENTAL. If set greater than 1, the blob name keyspace of each
|
|
directory is split into this many ranges which are listed concurrently
|
|
using the Arrow startFrom/endBefore range parameters. This can
|
|
dramatically speed up listing containers with millions of objects, for
|
|
both recursive (ListR) and single directory listings.
|
|
|
|
This has no effect unless "use_arrow_list" is also set, as Arrow is the
|
|
only listing path that supports server-side name ranges. The default of
|
|
0 (or 1) lists sequentially.`,
|
|
Default: 0,
|
|
Advanced: true,
|
|
Hide: fs.OptionHideBoth,
|
|
}, {
|
|
Name: "access_tier",
|
|
Help: `Access tier of blob: hot, cool, cold or archive.
|
|
|
|
Archived blobs can be restored by setting access tier to hot, cool or
|
|
cold. Leave blank if you intend to use default access tier, which is
|
|
set at account level
|
|
|
|
If there is no "access tier" specified, rclone doesn't apply any tier.
|
|
rclone performs "Set Tier" operation on blobs while uploading, if objects
|
|
are not modified, specifying "access tier" to new one will have no effect.
|
|
If blobs are in "archive tier" at remote, trying to perform data transfer
|
|
operations from remote will not be allowed. User should first restore by
|
|
tiering blob to "Hot", "Cool" or "Cold".`,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "archive_tier_delete",
|
|
Default: false,
|
|
Help: fmt.Sprintf(`Delete archive tier blobs before overwriting.
|
|
|
|
Archive tier blobs cannot be updated. So without this flag, if you
|
|
attempt to update an archive tier blob, then rclone will produce the
|
|
error:
|
|
|
|
%v
|
|
|
|
With this flag set then before rclone attempts to overwrite an archive
|
|
tier blob, it will delete the existing blob before uploading its
|
|
replacement. This has the potential for data loss if the upload fails
|
|
(unlike updating a normal blob) and also may cost more since deleting
|
|
archive tier blobs early may be chargable.
|
|
`, errCantUpdateArchiveTierBlobs),
|
|
Advanced: true,
|
|
}, {
|
|
Name: "disable_checksum",
|
|
Help: `Don't store MD5 checksum with object metadata.
|
|
|
|
Normally rclone will calculate the MD5 checksum of the input before
|
|
uploading it so it can add it to metadata on the object. This is great
|
|
for data integrity checking but can cause long delays for large files
|
|
to start uploading.`,
|
|
Default: false,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "memory_pool_flush_time",
|
|
Default: fs.Duration(time.Minute),
|
|
Advanced: true,
|
|
Hide: fs.OptionHideBoth,
|
|
Help: `How often internal memory buffer pools will be flushed. (no longer used)`,
|
|
}, {
|
|
Name: "memory_pool_use_mmap",
|
|
Default: false,
|
|
Advanced: true,
|
|
Hide: fs.OptionHideBoth,
|
|
Help: `Whether to use mmap buffers in internal memory pool. (no longer used)`,
|
|
}, {
|
|
Name: config.ConfigEncoding,
|
|
Help: config.ConfigEncodingHelp,
|
|
Advanced: true,
|
|
Default: (encoder.EncodeInvalidUtf8 |
|
|
encoder.EncodeSlash |
|
|
encoder.EncodeCtl |
|
|
encoder.EncodeDel |
|
|
encoder.EncodeBackSlash |
|
|
encoder.EncodeRightPeriod),
|
|
}, {
|
|
Name: "public_access",
|
|
Help: "Public access level of a container: blob or container.",
|
|
Default: "",
|
|
Examples: []fs.OptionExample{
|
|
{
|
|
Value: "",
|
|
Help: "The container and its blobs can be accessed only with an authorized request.\nIt's a default value.",
|
|
}, {
|
|
Value: string(container.PublicAccessTypeBlob),
|
|
Help: "Blob data within this container can be read via anonymous request.",
|
|
}, {
|
|
Value: string(container.PublicAccessTypeContainer),
|
|
Help: "Allow full public read access for container and blob data.",
|
|
},
|
|
},
|
|
Advanced: true,
|
|
}, {
|
|
Name: "directory_markers",
|
|
Default: false,
|
|
Advanced: true,
|
|
Help: `Upload an empty object with a trailing slash when a new directory is created
|
|
|
|
Empty folders are unsupported for bucket based remotes, this option
|
|
creates an empty object ending with "/", to persist the folder.
|
|
|
|
This object also has the metadata "` + dirMetaKey + ` = ` + dirMetaValue + `" to conform to
|
|
the Microsoft standard.
|
|
`,
|
|
}, {
|
|
Name: "no_check_container",
|
|
Help: `If set, don't attempt to check the container exists or create it.
|
|
|
|
This can be useful when trying to minimise the number of transactions
|
|
rclone does if you know the container exists already.
|
|
`,
|
|
Default: false,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "no_head_object",
|
|
Help: `If set, do not do HEAD before GET when getting objects.`,
|
|
Default: false,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "delete_snapshots",
|
|
Help: `Set to specify how to deal with snapshots on blob deletion.`,
|
|
Examples: []fs.OptionExample{
|
|
{
|
|
Value: "",
|
|
Help: "By default, the delete operation fails if a blob has snapshots",
|
|
}, {
|
|
Value: string(blob.DeleteSnapshotsOptionTypeInclude),
|
|
Help: "Specify 'include' to remove the root blob and all its snapshots",
|
|
}, {
|
|
Value: string(blob.DeleteSnapshotsOptionTypeOnly),
|
|
Help: "Specify 'only' to remove only the snapshots but keep the root blob.",
|
|
},
|
|
},
|
|
Default: "",
|
|
Exclusive: true,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "decompress",
|
|
Help: `If set this will decompress gzip encoded objects.
|
|
|
|
It is possible to upload objects to Azure Blob Storage with "Content-Encoding: gzip"
|
|
set. Normally rclone will download these files as compressed objects.
|
|
|
|
If this flag is set then rclone will decompress these files with
|
|
"Content-Encoding: gzip" as they are received. This means that rclone
|
|
can't check the size and hash but the file contents will be decompressed.
|
|
`,
|
|
Advanced: true,
|
|
Default: false,
|
|
}}),
|
|
})
|
|
}
|
|
|
|
// Options defines the configuration for this backend
|
|
type Options struct {
|
|
auth.Options
|
|
ChunkSize fs.SizeSuffix `config:"chunk_size"`
|
|
CopyCutoff fs.SizeSuffix `config:"copy_cutoff"`
|
|
CopyConcurrency int `config:"copy_concurrency"`
|
|
CopyTotalConcurrency int `config:"copy_total_concurrency"`
|
|
UseCopyBlob bool `config:"use_copy_blob"`
|
|
UploadConcurrency int `config:"upload_concurrency"`
|
|
ListChunkSize uint `config:"list_chunk"`
|
|
UseArrowList bool `config:"use_arrow_list"`
|
|
ListParallelism int `config:"list_parallelism"`
|
|
AccessTier string `config:"access_tier"`
|
|
ArchiveTierDelete bool `config:"archive_tier_delete"`
|
|
DisableCheckSum bool `config:"disable_checksum"`
|
|
Enc encoder.MultiEncoder `config:"encoding"`
|
|
PublicAccess string `config:"public_access"`
|
|
DirectoryMarkers bool `config:"directory_markers"`
|
|
NoCheckContainer bool `config:"no_check_container"`
|
|
NoHeadObject bool `config:"no_head_object"`
|
|
DeleteSnapshots string `config:"delete_snapshots"`
|
|
Decompress bool `config:"decompress"`
|
|
}
|
|
|
|
// Fs represents a remote azure server
|
|
type Fs struct {
|
|
name string // name of this remote
|
|
root string // the path we are working on if any
|
|
opt Options // parsed config options
|
|
ci *fs.ConfigInfo // global config
|
|
features *fs.Features // optional features
|
|
cntSVCcacheMu sync.Mutex // mutex to protect cntSVCcache and arrowCntSVCcache
|
|
cntSVCcache map[string]*container.Client // reference to containerClient per container
|
|
arrowCntSVCcache map[string]*arrowlist.Client // reference to arrowlist client per container
|
|
arrowClientOpts *arrowlist.ClientOptions // client options for the arrowlist clients
|
|
arrowXMLFallback atomic.Bool // set once the server answers XML so parallel listing is not retried
|
|
svc *service.Client // client to access azblob
|
|
cred azcore.TokenCredential // how to generate tokens (may be nil)
|
|
usingSharedKeyCred bool // set if using shared key credentials
|
|
anonymous bool // if this is anonymous access
|
|
rootContainer string // container part of root (if any)
|
|
rootDirectory string // directory part of root (if any)
|
|
isLimited bool // if limited to one container
|
|
cache *bucket.Cache // cache for container creation status
|
|
pacer *fs.Pacer // To pace and retry the API calls
|
|
uploadToken *pacer.TokenDispenser // control concurrency
|
|
copyToken *pacer.TokenDispenser // global multipart copy concurrency limiter
|
|
publicAccess container.PublicAccessType // Container Public Access Level
|
|
|
|
warnCompressed sync.Once // warn once about compressed files
|
|
|
|
// user delegation cache
|
|
userDelegationMu sync.Mutex
|
|
userDelegation *service.UserDelegationCredential
|
|
userDelegationExpiry time.Time
|
|
}
|
|
|
|
// Object describes an azure object
|
|
type Object struct {
|
|
fs *Fs // what this object is part of
|
|
remote string // The remote path
|
|
modTime time.Time // The modified time of the object if known
|
|
md5 string // MD5 hash if known
|
|
size int64 // Size of the object
|
|
mimeType string // Content-Type of the object
|
|
accessTier blob.AccessTier // Blob Access Tier
|
|
meta map[string]string // blob metadata - take metadataMu when accessing
|
|
tags map[string]string // blob tags
|
|
contentEncoding *string // Content-Encoding of the object
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// Name of the remote (as passed into NewFs)
|
|
func (f *Fs) Name() string {
|
|
return f.name
|
|
}
|
|
|
|
// Root of the remote (as passed into NewFs)
|
|
func (f *Fs) Root() string {
|
|
return f.root
|
|
}
|
|
|
|
// String converts this Fs to a string
|
|
func (f *Fs) String() string {
|
|
if f.rootContainer == "" {
|
|
return "Azure root"
|
|
}
|
|
if f.rootDirectory == "" {
|
|
return fmt.Sprintf("Azure container %s", f.rootContainer)
|
|
}
|
|
return fmt.Sprintf("Azure container %s path %s", f.rootContainer, f.rootDirectory)
|
|
}
|
|
|
|
// Features returns the optional features of this Fs
|
|
func (f *Fs) Features() *fs.Features {
|
|
return f.features
|
|
}
|
|
|
|
// parsePath parses a remote 'url'
|
|
func parsePath(path string) (root string) {
|
|
root = strings.Trim(path, "/")
|
|
return
|
|
}
|
|
|
|
// split returns container and containerPath from the rootRelativePath
|
|
// relative to f.root
|
|
func (f *Fs) split(rootRelativePath string) (containerName, containerPath string) {
|
|
containerName, containerPath = bucket.Split(bucket.Join(f.root, rootRelativePath))
|
|
if f.opt.DirectoryMarkers && strings.HasSuffix(containerPath, "//") {
|
|
containerPath = containerPath[:len(containerPath)-1]
|
|
}
|
|
return f.opt.Enc.FromStandardName(containerName), f.opt.Enc.FromStandardPath(containerPath)
|
|
}
|
|
|
|
// split returns container and containerPath from the object
|
|
func (o *Object) split() (container, containerPath string) {
|
|
return o.fs.split(o.remote)
|
|
}
|
|
|
|
// validateAccessTier checks if azureblob supports user supplied tier
|
|
func validateAccessTier(tier string) bool {
|
|
return strings.EqualFold(tier, string(blob.AccessTierHot)) ||
|
|
strings.EqualFold(tier, string(blob.AccessTierCool)) ||
|
|
strings.EqualFold(tier, string(blob.AccessTierCold)) ||
|
|
strings.EqualFold(tier, string(blob.AccessTierArchive))
|
|
}
|
|
|
|
// validatePublicAccess checks if azureblob supports use supplied public access level
|
|
func validatePublicAccess(publicAccess string) bool {
|
|
switch publicAccess {
|
|
case "",
|
|
string(container.PublicAccessTypeBlob),
|
|
string(container.PublicAccessTypeContainer):
|
|
// valid cases
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// retryErrorCodes is a slice of error codes that we will retry
|
|
var retryErrorCodes = []int{
|
|
401, // Unauthorized (e.g. "Token has expired")
|
|
408, // Request Timeout
|
|
429, // Rate exceeded.
|
|
500, // Get occasional 500 Internal Server Error
|
|
503, // Service Unavailable
|
|
504, // Gateway Time-out
|
|
}
|
|
|
|
// shouldRetry returns a boolean as to whether this resp and err
|
|
// deserve to be retried. It returns the err as a convenience
|
|
func (f *Fs) shouldRetry(ctx context.Context, err error) (bool, error) {
|
|
if fserrors.ContextError(ctx, &err) {
|
|
return false, err
|
|
}
|
|
var storageErr *azcore.ResponseError
|
|
if errors.As(err, &storageErr) {
|
|
// General errors from:
|
|
// https://learn.microsoft.com/en-us/rest/api/storageservices/common-rest-api-error-codes
|
|
// Blob specific errors from:
|
|
// https://learn.microsoft.com/en-us/rest/api/storageservices/blob-service-error-codes
|
|
switch storageErr.ErrorCode {
|
|
case "InvalidBlobOrBlock":
|
|
// These errors happen sometimes in multipart uploads
|
|
// because of block concurrency issues
|
|
return true, err
|
|
case "InternalError":
|
|
// The server encountered an internal error. Please retry the request.
|
|
return true, err
|
|
case "OperationTimedOut":
|
|
// The operation could not be completed within the permitted time. The
|
|
// operation may or may not have succeeded on the server side. Please query
|
|
// the server state before retrying the operation.
|
|
return true, err
|
|
case "ServerBusy":
|
|
// The server is currently unable to receive requests. Please retry your
|
|
// request.
|
|
return true, err
|
|
}
|
|
statusCode := storageErr.StatusCode
|
|
if slices.Contains(retryErrorCodes, statusCode) {
|
|
return true, err
|
|
}
|
|
}
|
|
return fserrors.ShouldRetry(err), err
|
|
}
|
|
|
|
func checkUploadChunkSize(cs fs.SizeSuffix) error {
|
|
const minChunkSize = fs.SizeSuffixBase
|
|
if cs < minChunkSize {
|
|
return fmt.Errorf("%s is less than %s", cs, minChunkSize)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *Fs) setUploadChunkSize(cs fs.SizeSuffix) (old fs.SizeSuffix, err error) {
|
|
err = checkUploadChunkSize(cs)
|
|
if err == nil {
|
|
old, f.opt.ChunkSize = f.opt.ChunkSize, cs
|
|
}
|
|
return
|
|
}
|
|
|
|
func (f *Fs) setCopyCutoff(cs fs.SizeSuffix) (old fs.SizeSuffix, err error) {
|
|
err = checkUploadChunkSize(cs)
|
|
if err == nil {
|
|
old, f.opt.CopyCutoff = f.opt.CopyCutoff, cs
|
|
}
|
|
return
|
|
}
|
|
|
|
// setRoot changes the root of the Fs
|
|
func (f *Fs) setRoot(root string) {
|
|
f.root = parsePath(root)
|
|
f.rootContainer, f.rootDirectory = bucket.Split(f.root)
|
|
}
|
|
|
|
// NewFs constructs an Fs from the path, container:path
|
|
func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, error) {
|
|
// Parse config into Options struct
|
|
opt := new(Options)
|
|
err := configstruct.Set(m, opt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = checkUploadChunkSize(opt.ChunkSize)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("chunk size: %w", err)
|
|
}
|
|
if opt.ListChunkSize > maxListChunkSize {
|
|
return nil, fmt.Errorf("blob list size can't be greater than %v - was %v", maxListChunkSize, opt.ListChunkSize)
|
|
}
|
|
if opt.ListParallelism > 1 && !opt.UseArrowList {
|
|
fs.Logf(nil, "azureblob: list_parallelism has no effect without use_arrow_list - listing sequentially")
|
|
}
|
|
|
|
if opt.AccessTier == "" {
|
|
opt.AccessTier = string(defaultAccessTier)
|
|
} else if !validateAccessTier(opt.AccessTier) {
|
|
return nil, fmt.Errorf("supported access tiers are %s, %s, %s and %s",
|
|
string(blob.AccessTierHot), string(blob.AccessTierCool), string(blob.AccessTierCold), string(blob.AccessTierArchive))
|
|
}
|
|
|
|
if !validatePublicAccess((opt.PublicAccess)) {
|
|
return nil, fmt.Errorf("supported public access level are %s and %s",
|
|
string(container.PublicAccessTypeBlob), string(container.PublicAccessTypeContainer))
|
|
}
|
|
|
|
ci := fs.GetConfig(ctx)
|
|
f := &Fs{
|
|
name: name,
|
|
opt: *opt,
|
|
ci: ci,
|
|
pacer: fs.NewPacer(ctx, pacer.NewS3(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant))),
|
|
uploadToken: pacer.NewTokenDispenser(ci.Transfers),
|
|
copyToken: pacer.NewTokenDispenser(opt.CopyTotalConcurrency),
|
|
cache: bucket.NewCache(),
|
|
cntSVCcache: make(map[string]*container.Client, 1),
|
|
arrowCntSVCcache: make(map[string]*arrowlist.Client, 1),
|
|
}
|
|
f.publicAccess = container.PublicAccessType(opt.PublicAccess)
|
|
f.setRoot(root)
|
|
f.features = (&fs.Features{
|
|
ReadMimeType: true,
|
|
WriteMimeType: true,
|
|
ReadMetadata: true,
|
|
WriteMetadata: true,
|
|
UserMetadata: true,
|
|
BucketBased: true,
|
|
BucketBasedRootOK: true,
|
|
SetTier: true,
|
|
GetTier: true,
|
|
ServerSideAcrossConfigs: true,
|
|
DoubleSlash: true,
|
|
}).Fill(ctx, f)
|
|
if opt.DirectoryMarkers {
|
|
f.features.CanHaveEmptyDirectories = true
|
|
fs.Debugf(f, "Using directory markers")
|
|
}
|
|
|
|
conf := auth.NewClientOpts[service.Client, service.ClientOptions, service.SharedKeyCredential]{
|
|
DefaultBaseURL: storageDefaultBaseURL,
|
|
RootContainer: f.rootContainer,
|
|
Blob: true,
|
|
NewClient: service.NewClient,
|
|
NewClientFromConnectionString: service.NewClientFromConnectionString,
|
|
NewClientWithNoCredential: service.NewClientWithNoCredential,
|
|
NewClientWithSharedKeyCredential: service.NewClientWithSharedKeyCredential,
|
|
NewSharedKeyCredential: service.NewSharedKeyCredential,
|
|
SetClientOptions: func(options *service.ClientOptions, policyClientOptions policy.ClientOptions) {
|
|
// Override the automatic decompression in the transport
|
|
// to download compressed files as-is
|
|
policyClientOptions.PerCallPolicies = append(policyClientOptions.PerCallPolicies, setAcceptEncodingGzip{})
|
|
options.ClientOptions = policyClientOptions
|
|
},
|
|
}
|
|
res, err := auth.NewClient(ctx, conf, &opt.Options)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
f.svc = res.Client
|
|
f.cred = res.Cred
|
|
f.usingSharedKeyCred = res.UsingSharedKeyCred
|
|
f.anonymous = res.Anonymous
|
|
|
|
// Client options for the arrowlist clients, using the same transport and
|
|
// gzip policy as the SDK clients built above.
|
|
f.arrowClientOpts = &arrowlist.ClientOptions{
|
|
ClientOptions: azcore.ClientOptions{
|
|
Transport: auth.Transporter(ctx),
|
|
PerCallPolicies: []policy.Policy{setAcceptEncodingGzip{}},
|
|
},
|
|
}
|
|
|
|
// if using Container level SAS put the container client into the cache
|
|
if opt.SASURL != "" && res.Container != "" {
|
|
_ = f.cntSVC(res.Container)
|
|
f.isLimited = true
|
|
}
|
|
|
|
if f.rootContainer != "" && f.rootDirectory != "" {
|
|
// Check to see if the (container,directory) is actually an existing file
|
|
oldRoot := f.root
|
|
newRoot, leaf := path.Split(oldRoot)
|
|
f.setRoot(newRoot)
|
|
_, err := f.NewObject(ctx, leaf)
|
|
if err != nil {
|
|
if err == fs.ErrorObjectNotFound || err == fs.ErrorNotAFile {
|
|
// File doesn't exist or is a directory so return old f
|
|
f.setRoot(oldRoot)
|
|
return f, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
// return an error with an fs which points to the parent
|
|
return f, fs.ErrorIsFile
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// return the container client for the container passed in
|
|
func (f *Fs) cntSVC(containerName string) (containerClient *container.Client) {
|
|
f.cntSVCcacheMu.Lock()
|
|
defer f.cntSVCcacheMu.Unlock()
|
|
var ok bool
|
|
if containerClient, ok = f.cntSVCcache[containerName]; !ok {
|
|
containerClient = f.svc.NewContainerClient(containerName)
|
|
f.cntSVCcache[containerName] = containerClient
|
|
}
|
|
return containerClient
|
|
}
|
|
|
|
// errArrowAuthUnsupported is returned by arrowCntSVC when the configured
|
|
// credentials can't be used for Arrow listing.
|
|
var errArrowAuthUnsupported = errors.New("credentials not supported for Arrow listing (connection_string auth is not supported)")
|
|
|
|
// return the arrowlist client for the container passed in
|
|
//
|
|
// Returns errArrowAuthUnsupported if the configured credentials can't be
|
|
// reused for the arrowlist client's pipeline.
|
|
func (f *Fs) arrowCntSVC(containerName string) (client *arrowlist.Client, err error) {
|
|
// The container URL includes any SAS token in its query
|
|
url := f.cntSVC(containerName).URL()
|
|
f.cntSVCcacheMu.Lock()
|
|
defer f.cntSVCcacheMu.Unlock()
|
|
if client, ok := f.arrowCntSVCcache[containerName]; ok {
|
|
return client, nil
|
|
}
|
|
switch {
|
|
case f.usingSharedKeyCred:
|
|
// Covers account+key and the emulator (auth fills in Account/Key)
|
|
var cred *arrowlist.SharedKeyCredential
|
|
cred, err = arrowlist.NewSharedKeyCredential(f.opt.Account, f.opt.Key)
|
|
if err == nil {
|
|
client, err = arrowlist.NewClientWithSharedKeyCredential(url, cred, f.arrowClientOpts)
|
|
}
|
|
case f.cred != nil:
|
|
client, err = arrowlist.NewClient(url, f.cred, f.arrowClientOpts)
|
|
case f.anonymous || f.opt.SASURL != "":
|
|
client, err = arrowlist.NewClientWithNoCredential(url, f.arrowClientOpts)
|
|
default:
|
|
return nil, errArrowAuthUnsupported
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
f.arrowCntSVCcache[containerName] = client
|
|
return client, nil
|
|
}
|
|
|
|
// Return an Object from a path
|
|
//
|
|
// If it can't be found it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) newObjectWithInfo(ctx context.Context, remote string, info *container.BlobItem) (fs.Object, error) {
|
|
o := &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
}
|
|
if info != nil {
|
|
err := o.decodeMetaDataFromBlob(info)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
} else if !o.fs.opt.NoHeadObject {
|
|
err := o.readMetaData(ctx) // reads info and headers, returning an error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return o, nil
|
|
}
|
|
|
|
// NewObject finds the Object at remote. If it can't be found
|
|
// it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) NewObject(ctx context.Context, remote string) (fs.Object, error) {
|
|
return f.newObjectWithInfo(ctx, remote, nil)
|
|
}
|
|
|
|
// getBlobSVC creates a blob client
|
|
func (f *Fs) getBlobSVC(container, containerPath string) *blob.Client {
|
|
return f.cntSVC(container).NewBlobClient(containerPath)
|
|
}
|
|
|
|
// getBlockBlobSVC creates a block blob client
|
|
func (f *Fs) getBlockBlobSVC(container, containerPath string) *blockblob.Client {
|
|
return f.cntSVC(container).NewBlockBlobClient(containerPath)
|
|
}
|
|
|
|
// updateMetadataWithModTime adds the modTime passed in to o.meta.
|
|
func (o *Object) updateMetadataWithModTime(modTime time.Time) {
|
|
metadataMu.Lock()
|
|
defer metadataMu.Unlock()
|
|
|
|
// Make sure o.meta is not nil
|
|
if o.meta == nil {
|
|
o.meta = make(map[string]string, 1)
|
|
}
|
|
|
|
// Set modTimeKey in it
|
|
o.meta[modTimeKey] = modTime.Format(timeFormatOut)
|
|
}
|
|
|
|
// parseXMsTags parses the value of the x-ms-tags header into a map.
|
|
// It expects comma-separated key=value pairs. Whitespace around keys and
|
|
// values is trimmed. Empty pairs and empty keys are rejected.
|
|
func parseXMsTags(s string) (map[string]string, error) {
|
|
if strings.TrimSpace(s) == "" {
|
|
return map[string]string{}, nil
|
|
}
|
|
out := make(map[string]string)
|
|
parts := strings.SplitSeq(s, ",")
|
|
for p := range parts {
|
|
p = strings.TrimSpace(p)
|
|
if p == "" {
|
|
continue
|
|
}
|
|
kv := strings.SplitN(p, "=", 2)
|
|
if len(kv) != 2 {
|
|
return nil, fmt.Errorf("invalid tag %q", p)
|
|
}
|
|
k := strings.TrimSpace(kv[0])
|
|
v := strings.TrimSpace(kv[1])
|
|
if k == "" {
|
|
return nil, fmt.Errorf("invalid tag key in %q", p)
|
|
}
|
|
out[k] = v
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// mapMetadataToAzure maps a generic metadata map to Azure HTTP headers,
|
|
// user metadata, tags and optional modTime override.
|
|
// Reserved x-ms-* keys (except x-ms-tags) are ignored for user metadata.
|
|
//
|
|
// Pass a logger to surface non-fatal parsing issues (e.g. bad mtime).
|
|
func mapMetadataToAzure(meta map[string]string, logf func(string, ...any)) (headers blob.HTTPHeaders, userMeta map[string]*string, tags map[string]string, modTime *time.Time, err error) {
|
|
if meta == nil {
|
|
return headers, nil, nil, nil, nil
|
|
}
|
|
tmp := make(map[string]string)
|
|
for k, v := range meta {
|
|
lowerKey := strings.ToLower(k)
|
|
switch lowerKey {
|
|
case "cache-control":
|
|
headers.BlobCacheControl = pString(v)
|
|
case "content-disposition":
|
|
headers.BlobContentDisposition = pString(v)
|
|
case "content-encoding":
|
|
headers.BlobContentEncoding = pString(v)
|
|
case "content-language":
|
|
headers.BlobContentLanguage = pString(v)
|
|
case "content-type":
|
|
headers.BlobContentType = pString(v)
|
|
case "x-ms-tags":
|
|
parsed, perr := parseXMsTags(v)
|
|
if perr != nil {
|
|
return headers, nil, nil, nil, perr
|
|
}
|
|
// allocate only if there are tags
|
|
if len(parsed) > 0 {
|
|
tags = parsed
|
|
}
|
|
case "mtime":
|
|
// Accept multiple layouts for tolerance
|
|
var parsed time.Time
|
|
var pErr error
|
|
for _, layout := range []string{time.RFC3339Nano, time.RFC3339, timeFormatOut} {
|
|
parsed, pErr = time.Parse(layout, v)
|
|
if pErr == nil {
|
|
modTime = &parsed
|
|
break
|
|
}
|
|
}
|
|
// Log and ignore if unparseable
|
|
if modTime == nil && logf != nil {
|
|
logf("metadata: couldn't parse mtime %q: %v", v, pErr)
|
|
}
|
|
case "tier":
|
|
// ignore - handled elsewhere
|
|
default:
|
|
// Filter out other reserved headers so they don't end up as user metadata
|
|
if strings.HasPrefix(lowerKey, "x-ms-") {
|
|
continue
|
|
}
|
|
tmp[lowerKey] = v
|
|
}
|
|
}
|
|
userMeta = toAzureMetaPtr(tmp)
|
|
return headers, userMeta, tags, modTime, nil
|
|
}
|
|
|
|
// toAzureMetaPtr converts a map[string]string to map[string]*string as used by Azure SDK
|
|
func toAzureMetaPtr(in map[string]string) map[string]*string {
|
|
if len(in) == 0 {
|
|
return nil
|
|
}
|
|
out := make(map[string]*string, len(in))
|
|
for k, v := range in {
|
|
vv := v
|
|
out[k] = &vv
|
|
}
|
|
return out
|
|
}
|
|
|
|
// assembleCopyParams prepares headers, metadata and tags for copy operations.
|
|
//
|
|
// It starts from the source properties, optionally overlays mapped metadata
|
|
// from rclone's metadata options, ensures mtime presence when mapping is
|
|
// enabled, and returns whether mapping was actually requested (hadMapping).
|
|
// assembleCopyParams prepares headers, metadata and tags for copy operations.
|
|
//
|
|
// If includeBaseMeta is true, start user metadata from the source's metadata
|
|
// and overlay mapped values. This matches multipart copy commit behavior.
|
|
// If false, only include mapped user metadata (no source baseline) which
|
|
// matches previous singlepart StartCopyFromURL semantics.
|
|
func assembleCopyParams(ctx context.Context, f *Fs, src fs.Object, srcProps *blob.GetPropertiesResponse, includeBaseMeta bool) (headers blob.HTTPHeaders, meta map[string]*string, tags map[string]string, hadMapping bool, err error) {
|
|
// Start from source properties
|
|
headers = blob.HTTPHeaders{
|
|
BlobCacheControl: srcProps.CacheControl,
|
|
BlobContentDisposition: srcProps.ContentDisposition,
|
|
BlobContentEncoding: srcProps.ContentEncoding,
|
|
BlobContentLanguage: srcProps.ContentLanguage,
|
|
BlobContentMD5: srcProps.ContentMD5,
|
|
BlobContentType: srcProps.ContentType,
|
|
}
|
|
// Optionally deep copy user metadata pointers from source. Normalise keys to
|
|
// lower-case to avoid duplicate x-ms-meta headers when we later inject/overlay
|
|
// metadata (Azure treats keys case-insensitively but Go's http.Header will
|
|
// join duplicate keys into a comma separated list, which breaks shared-key
|
|
// signing).
|
|
if includeBaseMeta && len(srcProps.Metadata) > 0 {
|
|
meta = make(map[string]*string, len(srcProps.Metadata))
|
|
for k, v := range srcProps.Metadata {
|
|
if v != nil {
|
|
vv := *v
|
|
meta[strings.ToLower(k)] = &vv
|
|
}
|
|
}
|
|
}
|
|
|
|
// Only consider mapping if metadata pipeline is enabled
|
|
if fs.GetConfig(ctx).Metadata {
|
|
mapped, mapErr := fs.GetMetadataOptions(ctx, f, src, fs.MetadataAsOpenOptions(ctx))
|
|
if mapErr != nil {
|
|
return headers, meta, nil, false, fmt.Errorf("failed to map metadata: %w", mapErr)
|
|
}
|
|
if mapped != nil {
|
|
// Map rclone metadata to Azure shapes
|
|
mappedHeaders, userMeta, mappedTags, mappedModTime, herr := mapMetadataToAzure(mapped, func(format string, args ...any) { fs.Debugf(f, format, args...) })
|
|
if herr != nil {
|
|
return headers, meta, nil, false, fmt.Errorf("metadata mapping: %w", herr)
|
|
}
|
|
hadMapping = true
|
|
// Overlay headers (only non-nil)
|
|
if mappedHeaders.BlobCacheControl != nil {
|
|
headers.BlobCacheControl = mappedHeaders.BlobCacheControl
|
|
}
|
|
if mappedHeaders.BlobContentDisposition != nil {
|
|
headers.BlobContentDisposition = mappedHeaders.BlobContentDisposition
|
|
}
|
|
if mappedHeaders.BlobContentEncoding != nil {
|
|
headers.BlobContentEncoding = mappedHeaders.BlobContentEncoding
|
|
}
|
|
if mappedHeaders.BlobContentLanguage != nil {
|
|
headers.BlobContentLanguage = mappedHeaders.BlobContentLanguage
|
|
}
|
|
if mappedHeaders.BlobContentType != nil {
|
|
headers.BlobContentType = mappedHeaders.BlobContentType
|
|
}
|
|
// Overlay user metadata
|
|
if len(userMeta) > 0 {
|
|
if meta == nil {
|
|
meta = make(map[string]*string, len(userMeta))
|
|
}
|
|
maps.Copy(meta, userMeta)
|
|
}
|
|
// Apply tags if any
|
|
if len(mappedTags) > 0 {
|
|
tags = mappedTags
|
|
}
|
|
// Ensure mtime present using mapped or source time
|
|
if _, ok := meta[modTimeKey]; !ok {
|
|
when := src.ModTime(ctx)
|
|
if mappedModTime != nil {
|
|
when = *mappedModTime
|
|
}
|
|
val := when.Format(time.RFC3339Nano)
|
|
if meta == nil {
|
|
meta = make(map[string]*string, 1)
|
|
}
|
|
meta[modTimeKey] = &val
|
|
}
|
|
// Ensure content-type fallback to source if not set by mapper
|
|
if headers.BlobContentType == nil {
|
|
headers.BlobContentType = srcProps.ContentType
|
|
}
|
|
} else {
|
|
// Mapping enabled but not provided: ensure mtime present based on source ModTime
|
|
if _, ok := meta[modTimeKey]; !ok {
|
|
when := src.ModTime(ctx)
|
|
val := when.Format(time.RFC3339Nano)
|
|
if meta == nil {
|
|
meta = make(map[string]*string, 1)
|
|
}
|
|
meta[modTimeKey] = &val
|
|
}
|
|
}
|
|
}
|
|
|
|
return headers, meta, tags, hadMapping, nil
|
|
}
|
|
|
|
// applyMappedMetadata applies mapped metadata and headers to the object state for uploads.
|
|
//
|
|
// It reads `--metadata`, `--metadata-set`, and `--metadata-mapper` outputs via fs.GetMetadataOptions
|
|
// and updates o.meta, o.tags and ui.httpHeaders accordingly.
|
|
func (o *Object) applyMappedMetadata(ctx context.Context, src fs.ObjectInfo, ui *uploadInfo, options []fs.OpenOption) (modTime time.Time, err error) {
|
|
// Start from the source modtime; may be overridden by metadata
|
|
modTime = src.ModTime(ctx)
|
|
|
|
// Fetch mapped metadata if --metadata is enabled
|
|
meta, err := fs.GetMetadataOptions(ctx, o.fs, src, options)
|
|
if err != nil {
|
|
return modTime, err
|
|
}
|
|
if meta == nil {
|
|
// No metadata processing requested
|
|
return modTime, nil
|
|
}
|
|
|
|
// Map metadata using common helper
|
|
headers, userMeta, tags, mappedModTime, err := mapMetadataToAzure(meta, func(format string, args ...any) { fs.Debugf(o, format, args...) })
|
|
if err != nil {
|
|
return modTime, err
|
|
}
|
|
// Merge headers into ui
|
|
if headers.BlobCacheControl != nil {
|
|
ui.httpHeaders.BlobCacheControl = headers.BlobCacheControl
|
|
}
|
|
if headers.BlobContentDisposition != nil {
|
|
ui.httpHeaders.BlobContentDisposition = headers.BlobContentDisposition
|
|
}
|
|
if headers.BlobContentEncoding != nil {
|
|
ui.httpHeaders.BlobContentEncoding = headers.BlobContentEncoding
|
|
}
|
|
if headers.BlobContentLanguage != nil {
|
|
ui.httpHeaders.BlobContentLanguage = headers.BlobContentLanguage
|
|
}
|
|
if headers.BlobContentType != nil {
|
|
ui.httpHeaders.BlobContentType = headers.BlobContentType
|
|
}
|
|
|
|
// Apply user metadata to o.meta with a single critical section
|
|
if len(userMeta) > 0 {
|
|
metadataMu.Lock()
|
|
if o.meta == nil {
|
|
o.meta = make(map[string]string, len(userMeta))
|
|
}
|
|
for k, v := range userMeta {
|
|
if v != nil {
|
|
o.meta[k] = *v
|
|
}
|
|
}
|
|
metadataMu.Unlock()
|
|
}
|
|
|
|
// Apply tags
|
|
if len(tags) > 0 {
|
|
if o.tags == nil {
|
|
o.tags = make(map[string]string, len(tags))
|
|
}
|
|
maps.Copy(o.tags, tags)
|
|
}
|
|
|
|
if mappedModTime != nil {
|
|
modTime = *mappedModTime
|
|
}
|
|
|
|
return modTime, nil
|
|
}
|
|
|
|
// Returns whether file is a directory marker or not
|
|
func isDirectoryMarker(size int64, metadata map[string]*string, remote string) bool {
|
|
// Directory markers are 0 length
|
|
if size == 0 {
|
|
endsWithSlash := strings.HasSuffix(remote, "/")
|
|
if endsWithSlash || remote == "" {
|
|
return true
|
|
}
|
|
// Note that metadata with hdi_isfolder = true seems to be a
|
|
// defacto standard for marking blobs as directories.
|
|
// Note also that the metadata hasn't been normalised to lower case yet
|
|
for k, v := range metadata {
|
|
if v != nil && strings.EqualFold(k, dirMetaKey) && *v == dirMetaValue {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// listFn is called from list to handle an object
|
|
type listFn func(remote string, object *container.BlobItem, isDirectory bool) error
|
|
|
|
// list lists the objects into the function supplied from
|
|
// the container and root supplied
|
|
//
|
|
// dir is the starting directory, "" for root
|
|
//
|
|
// The remote has prefix removed from it and if addContainer is set then
|
|
// it adds the container to the start.
|
|
func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, addContainer bool, recurse bool, maxResults int32, fn listFn) error {
|
|
if f.cache.IsDeleted(containerName) {
|
|
return fs.ErrorDirNotFound
|
|
}
|
|
if prefix != "" {
|
|
prefix += "/"
|
|
}
|
|
if directory != "" && (prefix == "" && !bucket.IsAllSlashes(directory) || prefix != "" && !strings.HasSuffix(directory, "/")) {
|
|
directory += "/"
|
|
}
|
|
delimiter := ""
|
|
if !recurse {
|
|
delimiter = "/"
|
|
}
|
|
|
|
opts := &arrowlist.ListBlobsHierarchyOptions{
|
|
ListBlobsHierarchyOptions: container.ListBlobsHierarchyOptions{
|
|
// Copy, Metadata, Snapshots, UncommittedBlobs, Deleted, Tags, Versions, LegalHold, ImmutabilityPolicy, DeletedWithVersions bool
|
|
Include: container.ListBlobsInclude{
|
|
Copy: false,
|
|
Metadata: true,
|
|
Snapshots: false,
|
|
UncommittedBlobs: false,
|
|
Deleted: false,
|
|
},
|
|
Prefix: &directory,
|
|
MaxResults: &maxResults,
|
|
},
|
|
}
|
|
// Experimental: request the Apache Arrow listing format. The arrowlist
|
|
// pager requests an Arrow IPC stream and decodes it, falling back to XML
|
|
// if the account doesn't have Arrow listing enabled. Skip the
|
|
// maxResults==1 probe (isEmpty) which doesn't benefit.
|
|
useArrow := f.opt.UseArrowList && maxResults != 1
|
|
if useArrow {
|
|
opts.UseArrowFormat = to.Ptr(true)
|
|
}
|
|
|
|
var foundItems int
|
|
var err error
|
|
if useArrow && f.opt.ListParallelism > 1 && !f.arrowXMLFallback.Load() {
|
|
// Arrow exposes server-side startFrom/endBefore name ranges, so the
|
|
// keyspace can be sharded and listed concurrently.
|
|
foundItems, err = f.listArrowParallel(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
|
|
} else {
|
|
foundItems, err = f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if f.opt.DirectoryMarkers && foundItems == 0 && directory != "" {
|
|
// Determine whether the directory exists or not by whether it has a marker
|
|
_, err := f.readMetaData(ctx, containerName, directory)
|
|
if err != nil {
|
|
if err == fs.ErrorObjectNotFound {
|
|
return fs.ErrorDirNotFound
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// listBlobsPager runs the hierarchy listing described by opts, calling fn for
|
|
// each blob and subdirectory, and returns the number of raw items seen.
|
|
// delimiter selects flat (recurse) vs hierarchical listing. If opts requests
|
|
// the Apache Arrow format the listing goes through the arrowlist pager
|
|
// (falling back to the SDK pager if the credentials don't support it); if the
|
|
// service then answers with XML (Arrow listing not enabled) a debug message
|
|
// is logged.
|
|
func (f *Fs) listBlobsPager(ctx context.Context, containerName, directory, prefix string, addContainer bool, opts *arrowlist.ListBlobsHierarchyOptions, delimiter string, fn listFn) (foundItems int, err error) {
|
|
useArrow := opts.UseArrowFormat != nil && *opts.UseArrowFormat
|
|
var pager *runtime.Pager[container.ListBlobsHierarchyResponse]
|
|
if useArrow {
|
|
arrowSVC, err := f.arrowCntSVC(containerName)
|
|
if err != nil {
|
|
fs.Debugf(f, "Not using Arrow listing: %v", err)
|
|
useArrow = false
|
|
} else {
|
|
pager = arrowSVC.NewListBlobsHierarchyPager(delimiter, opts)
|
|
}
|
|
}
|
|
if pager == nil {
|
|
pager = f.cntSVC(containerName).NewListBlobsHierarchyPager(delimiter, &opts.ListBlobsHierarchyOptions)
|
|
}
|
|
checkedArrow := false
|
|
for pager.More() {
|
|
var response container.ListBlobsHierarchyResponse
|
|
err := f.pacer.Call(func() (bool, error) {
|
|
var err error
|
|
response, err = pager.NextPage(ctx)
|
|
//response, err = f.srv.ListBlobsHierarchySegment(ctx, marker, delimiter, options)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
|
|
// If Arrow was requested but the service answered with XML, Arrow
|
|
// listing is not enabled on this account; the listing is still correct
|
|
// but not accelerated.
|
|
if useArrow && !checkedArrow && err == nil {
|
|
checkedArrow = true
|
|
if response.ContentType == nil || !strings.HasPrefix(*response.ContentType, arrowlist.ArrowContentType) {
|
|
fs.Debugf(f, "Apache Arrow listing requested but server returned XML - Blob Listing with Apache Arrow may not be enabled on this account")
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
// Check http error code along with service code, current SDK doesn't populate service code correctly sometimes
|
|
if storageErr, ok := err.(*azcore.ResponseError); ok && (storageErr.ErrorCode == string(bloberror.ContainerNotFound) || storageErr.StatusCode == http.StatusNotFound) {
|
|
return foundItems, fs.ErrorDirNotFound
|
|
}
|
|
return foundItems, err
|
|
}
|
|
// Advance marker to next
|
|
// marker = response.NextMarker
|
|
foundItems += len(response.Segment.BlobItems)
|
|
for i := range response.Segment.BlobItems {
|
|
file := response.Segment.BlobItems[i]
|
|
// Finish if file name no longer has prefix
|
|
// if prefix != "" && !strings.HasPrefix(file.Name, prefix) {
|
|
// return nil
|
|
// }
|
|
if file.Name == nil {
|
|
fs.Debugf(f, "Nil name received")
|
|
continue
|
|
}
|
|
remote := f.opt.Enc.ToStandardPath(*file.Name)
|
|
if !strings.HasPrefix(remote, prefix) {
|
|
fs.Debugf(f, "Odd name received %q", remote)
|
|
continue
|
|
}
|
|
// ContentLength is documented as nullable in the Arrow listing schema
|
|
size := int64(-1)
|
|
if file.Properties.ContentLength != nil {
|
|
size = *file.Properties.ContentLength
|
|
}
|
|
isDirectory := isDirectoryMarker(size, file.Metadata, remote)
|
|
if isDirectory {
|
|
// Don't insert the root directory
|
|
if remote == f.opt.Enc.ToStandardPath(directory) {
|
|
continue
|
|
}
|
|
// process directory markers as directories
|
|
remote, _ = strings.CutSuffix(remote, "/")
|
|
}
|
|
remote = remote[len(prefix):]
|
|
if addContainer {
|
|
remote = path.Join(containerName, remote)
|
|
}
|
|
// Send object
|
|
err = fn(remote, file, isDirectory)
|
|
if err != nil {
|
|
return foundItems, err
|
|
}
|
|
}
|
|
// Send the subdirectories
|
|
foundItems += len(response.Segment.BlobPrefixes)
|
|
for _, blobPrefix := range response.Segment.BlobPrefixes {
|
|
if blobPrefix.Name == nil {
|
|
fs.Debugf(f, "Nil prefix received")
|
|
continue
|
|
}
|
|
remote := f.opt.Enc.ToStandardPath(*blobPrefix.Name)
|
|
if !strings.HasPrefix(remote, prefix) {
|
|
fs.Debugf(f, "Odd directory name received %q", remote)
|
|
continue
|
|
}
|
|
// Don't insert the root directory. The Arrow listing returns a
|
|
// directory marker blob with the same name as the listed directory
|
|
// as a BlobPrefix, whereas XML returns it as a blob (handled
|
|
// above).
|
|
if remote == f.opt.Enc.ToStandardPath(directory) {
|
|
continue
|
|
}
|
|
remote = remote[len(prefix):]
|
|
// Trim one slash off the remote name
|
|
remote, _ = strings.CutSuffix(remote, "/")
|
|
if remote == "" || bucket.IsAllSlashes(remote) {
|
|
remote += "/"
|
|
}
|
|
if addContainer {
|
|
remote = path.Join(containerName, remote)
|
|
}
|
|
// Send object
|
|
err = fn(remote, nil, true)
|
|
if err != nil {
|
|
return foundItems, err
|
|
}
|
|
}
|
|
}
|
|
return foundItems, nil
|
|
}
|
|
|
|
// Convert a list item into a DirEntry
|
|
func (f *Fs) itemToDirEntry(ctx context.Context, remote string, object *container.BlobItem, isDirectory bool) (fs.DirEntry, error) {
|
|
if isDirectory {
|
|
d := fs.NewDir(remote, time.Time{})
|
|
return d, nil
|
|
}
|
|
o, err := f.newObjectWithInfo(ctx, remote, object)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return o, nil
|
|
}
|
|
|
|
// Check to see if this is a limited container and the container is not found
|
|
func (f *Fs) containerOK(container string) bool {
|
|
if !f.isLimited {
|
|
return true
|
|
}
|
|
f.cntSVCcacheMu.Lock()
|
|
defer f.cntSVCcacheMu.Unlock()
|
|
for limitedContainer := range f.cntSVCcache {
|
|
if container == limitedContainer {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// listDir lists a single directory
|
|
func (f *Fs) listDir(ctx context.Context, containerName, directory, prefix string, addContainer bool, callback func(fs.DirEntry) error) (err error) {
|
|
if !f.containerOK(containerName) {
|
|
return fs.ErrorDirNotFound
|
|
}
|
|
err = f.list(ctx, containerName, directory, prefix, addContainer, false, int32(f.opt.ListChunkSize), func(remote string, object *container.BlobItem, isDirectory bool) error {
|
|
entry, err := f.itemToDirEntry(ctx, remote, object, isDirectory)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if entry != nil {
|
|
return callback(entry)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// container must be present if listing succeeded
|
|
f.cache.MarkOK(containerName)
|
|
return nil
|
|
}
|
|
|
|
// listContainers returns all the containers to out
|
|
func (f *Fs) listContainers(ctx context.Context) (entries fs.DirEntries, err error) {
|
|
if f.isLimited {
|
|
f.cntSVCcacheMu.Lock()
|
|
for container := range f.cntSVCcache {
|
|
d := fs.NewDir(container, time.Time{})
|
|
entries = append(entries, d)
|
|
}
|
|
f.cntSVCcacheMu.Unlock()
|
|
return entries, nil
|
|
}
|
|
err = f.listContainersToFn(func(Name string, LastModified time.Time) error {
|
|
d := fs.NewDir(f.opt.Enc.ToStandardName(Name), LastModified)
|
|
f.cache.MarkOK(Name)
|
|
entries = append(entries, d)
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return entries, nil
|
|
}
|
|
|
|
// List the objects and directories in dir into entries. The
|
|
// entries can be returned in any order but should be for a
|
|
// complete directory.
|
|
//
|
|
// dir should be "" to list the root, and should not have
|
|
// trailing slashes.
|
|
//
|
|
// This should return ErrDirNotFound if the directory isn't
|
|
// found.
|
|
func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err error) {
|
|
return list.WithListP(ctx, dir, f)
|
|
}
|
|
|
|
// ListP lists the objects and directories of the Fs starting
|
|
// from dir non recursively into out.
|
|
//
|
|
// dir should be "" to start from the root, and should not
|
|
// have trailing slashes.
|
|
//
|
|
// This should return ErrDirNotFound if the directory isn't
|
|
// found.
|
|
//
|
|
// It should call callback for each tranche of entries read.
|
|
// These need not be returned in any particular order. If
|
|
// callback returns an error then the listing will stop
|
|
// immediately.
|
|
func (f *Fs) ListP(ctx context.Context, dir string, callback fs.ListRCallback) error {
|
|
list := list.NewHelper(callback)
|
|
container, directory := f.split(dir)
|
|
if container == "" {
|
|
if directory != "" {
|
|
return fs.ErrorListBucketRequired
|
|
}
|
|
entries, err := f.listContainers(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, entry := range entries {
|
|
err = list.Add(entry)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
} else {
|
|
err := f.listDir(ctx, container, directory, f.rootDirectory, f.rootContainer == "", list.Add)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
}
|
|
return list.Flush()
|
|
}
|
|
|
|
// ListR lists the objects and directories of the Fs starting
|
|
// from dir recursively into out.
|
|
//
|
|
// dir should be "" to start from the root, and should not
|
|
// have trailing slashes.
|
|
//
|
|
// This should return ErrDirNotFound if the directory isn't
|
|
// found.
|
|
//
|
|
// It should call callback for each tranche of entries read.
|
|
// These need not be returned in any particular order. If
|
|
// callback returns an error then the listing will stop
|
|
// immediately.
|
|
//
|
|
// Don't implement this unless you have a more efficient way
|
|
// of listing recursively that doing a directory traversal.
|
|
func (f *Fs) ListR(ctx context.Context, dir string, callback fs.ListRCallback) (err error) {
|
|
containerName, directory := f.split(dir)
|
|
list := list.NewHelper(callback)
|
|
listR := func(containerName, directory, prefix string, addContainer bool) error {
|
|
return f.list(ctx, containerName, directory, prefix, addContainer, true, int32(f.opt.ListChunkSize), func(remote string, object *container.BlobItem, isDirectory bool) error {
|
|
entry, err := f.itemToDirEntry(ctx, remote, object, isDirectory)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return list.Add(entry)
|
|
})
|
|
}
|
|
if containerName == "" {
|
|
entries, err := f.listContainers(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, entry := range entries {
|
|
err = list.Add(entry)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
container := entry.Remote()
|
|
err = listR(container, "", f.rootDirectory, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// container must be present if listing succeeded
|
|
f.cache.MarkOK(container)
|
|
}
|
|
} else {
|
|
if !f.containerOK(containerName) {
|
|
return fs.ErrorDirNotFound
|
|
}
|
|
err = listR(containerName, directory, f.rootDirectory, f.rootContainer == "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// container must be present if listing succeeded
|
|
f.cache.MarkOK(containerName)
|
|
}
|
|
return list.Flush()
|
|
}
|
|
|
|
// listContainerFn is called from listContainersToFn to handle a container
|
|
type listContainerFn func(Name string, LastModified time.Time) error
|
|
|
|
// listContainersToFn lists the containers to the function supplied
|
|
func (f *Fs) listContainersToFn(fn listContainerFn) error {
|
|
max := int32(f.opt.ListChunkSize)
|
|
pager := f.svc.NewListContainersPager(&service.ListContainersOptions{
|
|
Include: service.ListContainersInclude{Metadata: true, Deleted: true},
|
|
MaxResults: &max,
|
|
})
|
|
ctx := context.Background()
|
|
for pager.More() {
|
|
var response service.ListContainersResponse
|
|
err := f.pacer.Call(func() (bool, error) {
|
|
var err error
|
|
response, err = pager.NextPage(ctx)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, cnt := range response.ContainerItems {
|
|
if cnt == nil || cnt.Name == nil || cnt.Properties == nil || cnt.Properties.LastModified == nil {
|
|
fs.Debugf(f, "nil returned in container info")
|
|
}
|
|
err = fn(*cnt.Name, *cnt.Properties.LastModified)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Put the object into the container
|
|
//
|
|
// Copy the reader in to the new object which is returned.
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
|
// Temporary Object under construction
|
|
fs := &Object{
|
|
fs: f,
|
|
remote: src.Remote(),
|
|
}
|
|
return fs, fs.Update(ctx, in, src, options...)
|
|
}
|
|
|
|
// PutStream uploads to the remote path with the modTime given of indeterminate size
|
|
func (f *Fs) PutStream(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
|
return f.Put(ctx, in, src, options...)
|
|
}
|
|
|
|
// Create directory marker file and parents
|
|
func (f *Fs) createDirectoryMarker(ctx context.Context, container, dir string) error {
|
|
if !f.opt.DirectoryMarkers || container == "" {
|
|
return nil
|
|
}
|
|
|
|
// Object to be uploaded
|
|
o := &Object{
|
|
fs: f,
|
|
modTime: time.Now(),
|
|
meta: map[string]string{
|
|
dirMetaKey: dirMetaValue,
|
|
},
|
|
}
|
|
|
|
for {
|
|
_, containerPath := f.split(dir)
|
|
// Don't create the directory marker if it is the bucket or at the very root
|
|
if containerPath == "" {
|
|
break
|
|
}
|
|
o.remote = dir + "/"
|
|
|
|
// Check to see if object already exists
|
|
_, err := f.readMetaData(ctx, container, containerPath+"/")
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
|
|
// Upload it if not
|
|
fs.Debugf(o, "Creating directory marker")
|
|
content := io.Reader(strings.NewReader(""))
|
|
err = o.Update(ctx, content, o)
|
|
if err != nil {
|
|
return fmt.Errorf("creating directory marker failed: %w", err)
|
|
}
|
|
|
|
// Now check parent directory exists
|
|
dir = path.Dir(dir)
|
|
if dir == "/" || dir == "." {
|
|
break
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Mkdir creates the container if it doesn't exist
|
|
func (f *Fs) Mkdir(ctx context.Context, dir string) error {
|
|
container, _ := f.split(dir)
|
|
e := f.makeContainer(ctx, container)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
return f.createDirectoryMarker(ctx, container, dir)
|
|
}
|
|
|
|
// mkdirParent creates the parent bucket/directory if it doesn't exist
|
|
func (f *Fs) mkdirParent(ctx context.Context, remote string) error {
|
|
remote, _ = strings.CutSuffix(remote, "/")
|
|
dir := path.Dir(remote)
|
|
if dir == "/" || dir == "." {
|
|
dir = ""
|
|
}
|
|
return f.Mkdir(ctx, dir)
|
|
}
|
|
|
|
// makeContainer creates the container if it doesn't exist
|
|
func (f *Fs) makeContainer(ctx context.Context, container string) error {
|
|
if f.opt.NoCheckContainer {
|
|
return nil
|
|
}
|
|
return f.cache.Create(container, func() error {
|
|
// If this is a SAS URL limited to a container then assume it is already created
|
|
if f.isLimited {
|
|
return nil
|
|
}
|
|
opt := service.CreateContainerOptions{
|
|
// Optional. Specifies a user-defined name-value pair associated with the blob.
|
|
//Metadata map[string]string
|
|
|
|
// Optional. Specifies the encryption scope settings to set on the container.
|
|
//CpkScopeInfo *CpkScopeInfo
|
|
}
|
|
if f.publicAccess != "" {
|
|
// Specifies whether data in the container may be accessed publicly and the level of access
|
|
opt.Access = &f.publicAccess
|
|
}
|
|
// now try to create the container
|
|
return f.pacer.Call(func() (bool, error) {
|
|
_, err := f.svc.CreateContainer(ctx, container, &opt)
|
|
if err != nil {
|
|
if storageErr, ok := err.(*azcore.ResponseError); ok {
|
|
switch bloberror.Code(storageErr.ErrorCode) {
|
|
case bloberror.ContainerAlreadyExists:
|
|
return false, nil
|
|
case bloberror.ContainerBeingDeleted:
|
|
// From https://docs.microsoft.com/en-us/rest/api/storageservices/delete-container
|
|
// When a container is deleted, a container with the same name cannot be created
|
|
// for at least 30 seconds; the container may not be available for more than 30
|
|
// seconds if the service is still processing the request.
|
|
time.Sleep(12 * time.Second) // default 10 retries will be 120 seconds
|
|
f.cache.MarkDeleted(container)
|
|
return true, err
|
|
case bloberror.AuthorizationFailure:
|
|
// Assume that the user does not have permission to
|
|
// create the container and carry on anyway.
|
|
fs.Debugf(f, "Tried to create container but got %s error - carrying on assuming container exists. Use no_check_container to stop this check..", storageErr.ErrorCode)
|
|
return false, nil
|
|
}
|
|
}
|
|
}
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
}, nil)
|
|
}
|
|
|
|
// isEmpty checks to see if a given (container, directory) is empty and returns an error if not
|
|
func (f *Fs) isEmpty(ctx context.Context, containerName, directory string) (err error) {
|
|
empty := true
|
|
err = f.list(ctx, containerName, directory, f.rootDirectory, f.rootContainer == "", true, 1, func(remote string, object *container.BlobItem, isDirectory bool) error {
|
|
empty = false
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !empty {
|
|
return fs.ErrorDirectoryNotEmpty
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// deleteContainer deletes the container. It can delete a full
|
|
// container so use isEmpty if you don't want that.
|
|
func (f *Fs) deleteContainer(ctx context.Context, containerName string) error {
|
|
return f.cache.Remove(containerName, func() error {
|
|
getOptions := container.GetPropertiesOptions{}
|
|
delOptions := container.DeleteOptions{}
|
|
return f.pacer.Call(func() (bool, error) {
|
|
_, err := f.cntSVC(containerName).GetProperties(ctx, &getOptions)
|
|
if err == nil {
|
|
_, err = f.cntSVC(containerName).Delete(ctx, &delOptions)
|
|
}
|
|
|
|
if err != nil {
|
|
// Check http error code along with service code, current SDK doesn't populate service code correctly sometimes
|
|
if storageErr, ok := err.(*azcore.ResponseError); ok && (storageErr.ErrorCode == string(bloberror.ContainerNotFound) || storageErr.StatusCode == http.StatusNotFound) {
|
|
return false, fs.ErrorDirNotFound
|
|
}
|
|
|
|
return f.shouldRetry(ctx, err)
|
|
}
|
|
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
})
|
|
}
|
|
|
|
// Rmdir deletes the container if the fs is at the root
|
|
//
|
|
// Returns an error if it isn't empty
|
|
func (f *Fs) Rmdir(ctx context.Context, dir string) error {
|
|
container, directory := f.split(dir)
|
|
// Remove directory marker file
|
|
if f.opt.DirectoryMarkers && container != "" && directory != "" {
|
|
o := &Object{
|
|
fs: f,
|
|
remote: dir + "/",
|
|
}
|
|
fs.Debugf(o, "Removing directory marker")
|
|
err := o.Remove(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("removing directory marker failed: %w", err)
|
|
}
|
|
}
|
|
if container == "" || directory != "" {
|
|
return nil
|
|
}
|
|
err := f.isEmpty(ctx, container, directory)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return f.deleteContainer(ctx, container)
|
|
}
|
|
|
|
// Precision of the remote
|
|
func (f *Fs) Precision() time.Duration {
|
|
return time.Nanosecond
|
|
}
|
|
|
|
// Hashes returns the supported hash sets.
|
|
func (f *Fs) Hashes() hash.Set {
|
|
return hash.Set(hash.MD5)
|
|
}
|
|
|
|
// Purge deletes all the files and directories including the old versions.
|
|
func (f *Fs) Purge(ctx context.Context, dir string) error {
|
|
container, directory := f.split(dir)
|
|
if container == "" {
|
|
return errors.New("can't purge from root")
|
|
}
|
|
if directory != "" {
|
|
// Delegate to caller if not root of a container
|
|
return fs.ErrorCantPurge
|
|
}
|
|
return f.deleteContainer(ctx, container)
|
|
}
|
|
|
|
// Get a user delegation which is valid for at least sasCopyValidity
|
|
//
|
|
// This value is cached in f
|
|
func (f *Fs) getUserDelegation(ctx context.Context) (*service.UserDelegationCredential, error) {
|
|
f.userDelegationMu.Lock()
|
|
defer f.userDelegationMu.Unlock()
|
|
|
|
if f.userDelegation != nil && time.Until(f.userDelegationExpiry) > sasCopyValidity {
|
|
return f.userDelegation, nil
|
|
}
|
|
|
|
// Validity window
|
|
start := time.Now().UTC().Add(-sasCopyStartSkew)
|
|
expiry := start.Add(2 * sasCopyValidity)
|
|
startStr := start.Format(time.RFC3339)
|
|
expiryStr := expiry.Format(time.RFC3339)
|
|
|
|
// Acquire user delegation key from the service client
|
|
info := service.KeyInfo{
|
|
Start: &startStr,
|
|
Expiry: &expiryStr,
|
|
}
|
|
userDelegationKey, err := f.svc.GetUserDelegationCredential(ctx, info, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get user delegation key: %w", err)
|
|
}
|
|
|
|
f.userDelegation = userDelegationKey
|
|
f.userDelegationExpiry = expiry
|
|
return f.userDelegation, nil
|
|
}
|
|
|
|
// getAuth gets auth to copy o.
|
|
//
|
|
// tokenOK is used to signal that token based auth (Microsoft Entra
|
|
// ID) is acceptable.
|
|
//
|
|
// This will return srcURL to read the object, which may be a SAS URL.
|
|
//
|
|
// If noAuth is set then the srcURL returned will be a plain object
|
|
// URL (not a SAS) and token will be empty.
|
|
//
|
|
// If tokenOK is true it may also return a token for the auth.
|
|
func (o *Object) getAuth(ctx context.Context, noAuth bool) (srcURL string, err error) {
|
|
f := o.fs
|
|
srcBlobSVC := o.getBlobSVC()
|
|
srcURL = srcBlobSVC.URL()
|
|
|
|
switch {
|
|
case noAuth:
|
|
// If same storage account then no auth needed
|
|
case f.cred != nil:
|
|
// Generate a User Delegation SAS URL using Azure AD credentials
|
|
userDelegationKey, err := f.getUserDelegation(ctx)
|
|
if err != nil {
|
|
return "", fmt.Errorf("sas creation: %w", err)
|
|
}
|
|
|
|
// Build the SAS values
|
|
perms := sas.BlobPermissions{Read: true}
|
|
container, containerPath := o.split()
|
|
start := time.Now().UTC().Add(-sasCopyStartSkew)
|
|
expiry := start.Add(sasCopyValidity)
|
|
vals := sas.BlobSignatureValues{
|
|
StartTime: start,
|
|
ExpiryTime: expiry,
|
|
Permissions: perms.String(),
|
|
ContainerName: container,
|
|
BlobName: containerPath,
|
|
}
|
|
|
|
// Sign with the delegation key
|
|
queryParameters, err := vals.SignWithUserDelegation(userDelegationKey)
|
|
if err != nil {
|
|
return "", fmt.Errorf("signing SAS with user delegation failed: %w", err)
|
|
}
|
|
|
|
// Append the SAS to the URL
|
|
srcURL = srcBlobSVC.URL() + "?" + queryParameters.Encode()
|
|
case f.usingSharedKeyCred:
|
|
// Generate a short lived SAS URL if using shared key credentials
|
|
expiry := time.Now().Add(sasCopyValidity)
|
|
sasOptions := blob.GetSASURLOptions{}
|
|
srcURL, err = srcBlobSVC.GetSASURL(sas.BlobPermissions{Read: true}, expiry, &sasOptions)
|
|
if err != nil {
|
|
return srcURL, fmt.Errorf("failed to create SAS URL: %w", err)
|
|
}
|
|
case f.anonymous || f.opt.SASURL != "":
|
|
// If using a SASURL or anonymous, no need for any extra auth
|
|
default:
|
|
return srcURL, errors.New("unknown authentication type")
|
|
}
|
|
return srcURL, nil
|
|
}
|
|
|
|
// Do multipart parallel copy.
|
|
//
|
|
// This uses these APIs:
|
|
//
|
|
// - PutBlockFromURL - https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-from-url
|
|
// - PutBlockList - https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list
|
|
func (f *Fs) copyMultipart(ctx context.Context, remote, dstContainer, dstPath string, src *Object) (dst fs.Object, err error) {
|
|
srcProperties, err := src.readMetaDataAlways(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("multipart copy: failed to read source object: %w", err)
|
|
}
|
|
|
|
// Create the dst object by altering a copy of the src object
|
|
obj := *src
|
|
o := &obj
|
|
o.fs = f
|
|
o.remote = remote
|
|
|
|
srcURL, err := src.getAuth(ctx, false)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("multipart copy: %w", err)
|
|
}
|
|
|
|
bic, err := newBlockIDCreator()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
dstBlockBlobSVC := f.getBlockBlobSVC(dstContainer, dstPath)
|
|
|
|
defer atexit.OnError(&err, func() {
|
|
// Try to abort the upload, but ignore the error.
|
|
fs.Debugf(o, "Cancelling multipart copy")
|
|
_ = o.clearUncommittedBlocks(ctx)
|
|
})()
|
|
|
|
var (
|
|
srcSize = src.size
|
|
partSize = int64(chunksize.Calculator(o, src.size, blockblob.MaxBlocks, f.opt.ChunkSize))
|
|
numParts = (srcSize + partSize - 1) / partSize
|
|
blockIDs = make([]string, numParts) // list of blocks for finalize
|
|
g, gCtx = errgroup.WithContext(ctx)
|
|
checker = newCheckForInvalidBlockOrBlob("copy", o)
|
|
account = transferaccounter.Get(ctx)
|
|
)
|
|
g.SetLimit(f.opt.CopyConcurrency)
|
|
|
|
fs.Debugf(o, "Starting multipart copy with %d parts of size %v", numParts, fs.SizeSuffix(partSize))
|
|
account.Start()
|
|
for partNum := uint64(0); partNum < uint64(numParts); partNum++ {
|
|
// Fail fast, in case an errgroup managed function returns an error
|
|
// gCtx is cancelled. There is no point in uploading all the other parts.
|
|
if gCtx.Err() != nil {
|
|
break
|
|
}
|
|
if f.opt.CopyTotalConcurrency > 0 {
|
|
f.copyToken.Get()
|
|
}
|
|
partNum := partNum // for closure
|
|
g.Go(func() error {
|
|
if f.opt.CopyTotalConcurrency > 0 {
|
|
defer f.copyToken.Put()
|
|
}
|
|
blockID := bic.newBlockID(partNum)
|
|
options := blockblob.StageBlockFromURLOptions{
|
|
Range: blob.HTTPRange{
|
|
Offset: int64(partNum) * partSize,
|
|
Count: partSize,
|
|
},
|
|
// Specifies the authorization scheme and signature for the copy source.
|
|
// We use SAS URLs as this doesn't seem to work always
|
|
// CopySourceAuthorization: token,
|
|
// CPKInfo *blob.CPKInfo
|
|
// CPKScopeInfo *blob.CPKScopeInfo
|
|
}
|
|
// Partial last block
|
|
if remaining := srcSize - options.Range.Offset; remaining < options.Range.Count {
|
|
options.Range.Count = remaining
|
|
}
|
|
fs.Debugf(o, "multipart copy: starting chunk %d size %v offset %v/%v", partNum, fs.SizeSuffix(options.Range.Count), fs.SizeSuffix(options.Range.Offset), fs.SizeSuffix(srcSize))
|
|
err := f.pacer.Call(func() (bool, error) {
|
|
checker.start()
|
|
_, err := dstBlockBlobSVC.StageBlockFromURL(ctx, blockID, srcURL, &options)
|
|
checker.stop()
|
|
if err != nil {
|
|
if checker.checkErr(ctx, err) {
|
|
return true, err
|
|
}
|
|
return f.shouldRetry(ctx, err)
|
|
}
|
|
return false, nil
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("multipart copy: failed to copy chunk %d with %v bytes: %w", partNum+1, -1, err)
|
|
}
|
|
blockIDs[partNum] = blockID
|
|
account.Add(options.Range.Count)
|
|
return nil
|
|
})
|
|
}
|
|
err = g.Wait()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Prepare metadata/headers/tags for destination
|
|
// For multipart commit, include base metadata from source then overlay mapped
|
|
commitHeaders, commitMeta, commitTags, _, err := assembleCopyParams(ctx, f, src, srcProperties, true)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("multipart copy: %w", err)
|
|
}
|
|
|
|
// Convert metadata from source or mapper
|
|
options := blockblob.CommitBlockListOptions{
|
|
Metadata: commitMeta,
|
|
Tags: commitTags,
|
|
Tier: parseTier(f.opt.AccessTier),
|
|
HTTPHeaders: &commitHeaders,
|
|
}
|
|
|
|
// Finalise the upload session
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
_, err := dstBlockBlobSVC.CommitBlockList(ctx, blockIDs, &options)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to complete multipart copy: %w", err)
|
|
}
|
|
|
|
fs.Debugf(o, "multipart copy finished")
|
|
return f.NewObject(ctx, remote)
|
|
}
|
|
|
|
// Do single part copy.
|
|
//
|
|
// This uses these APIs:
|
|
//
|
|
// - Copy Blob - https://docs.microsoft.com/rest/api/storageservices/copy-blob
|
|
// - Get Blob Properties - https://docs.microsoft.com/rest/api/storageservices/get-blob-properties
|
|
func (f *Fs) copySinglepart(ctx context.Context, remote, dstContainer, dstPath string, src *Object) (dst fs.Object, err error) {
|
|
dstBlobSVC := f.getBlobSVC(dstContainer, dstPath)
|
|
|
|
// Get the source auth - none needed for same storage account
|
|
srcURL, err := src.getAuth(ctx, f == src.fs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("single part copy: source auth: %w", err)
|
|
}
|
|
|
|
// Prepare mapped metadata/tags/headers if requested
|
|
options := blob.StartCopyFromURLOptions{
|
|
Tier: parseTier(f.opt.AccessTier),
|
|
}
|
|
var postHeaders *blob.HTTPHeaders
|
|
// Read source properties and assemble params; this also handles the case when mapping is disabled
|
|
srcProps, err := src.readMetaDataAlways(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("single part copy: read source properties: %w", err)
|
|
}
|
|
// For singlepart copy, do not include base metadata from source in StartCopyFromURL
|
|
headers, meta, tags, hadMapping, aerr := assembleCopyParams(ctx, f, src, srcProps, false)
|
|
if aerr != nil {
|
|
return nil, fmt.Errorf("single part copy: %w", aerr)
|
|
}
|
|
// Apply tags and post-copy headers only when mapping requested changes
|
|
if len(tags) > 0 {
|
|
options.BlobTags = make(map[string]string, len(tags))
|
|
maps.Copy(options.BlobTags, tags)
|
|
}
|
|
if hadMapping {
|
|
// Only set metadata explicitly when mapping was requested; otherwise
|
|
// let the service copy source metadata (including mtime) automatically.
|
|
if len(meta) > 0 {
|
|
options.Metadata = meta
|
|
}
|
|
postHeaders = &headers
|
|
}
|
|
var startCopy blob.StartCopyFromURLResponse
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
startCopy, err = dstBlobSVC.StartCopyFromURL(ctx, srcURL, &options)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("single part copy: copy blob: %w", err)
|
|
}
|
|
|
|
// Poll for completion if necessary
|
|
//
|
|
// The for loop is never executed for same storage account copies.
|
|
copyStatus := startCopy.CopyStatus
|
|
getOptions := blob.GetPropertiesOptions{}
|
|
pollTime := 100 * time.Millisecond
|
|
for copyStatus != nil && string(*copyStatus) == string(container.CopyStatusTypePending) {
|
|
time.Sleep(pollTime)
|
|
var getMetadata blob.GetPropertiesResponse
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
getMetadata, err = dstBlobSVC.GetProperties(ctx, &getOptions)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
copyStatus = getMetadata.CopyStatus
|
|
pollTime = min(2*pollTime, time.Second)
|
|
}
|
|
|
|
// If mapper requested header changes, set them post-copy
|
|
if postHeaders != nil {
|
|
blb := f.getBlobSVC(dstContainer, dstPath)
|
|
_, setErr := blb.SetHTTPHeaders(ctx, *postHeaders, nil)
|
|
if setErr != nil {
|
|
return nil, fmt.Errorf("single part copy: failed to set headers: %w", setErr)
|
|
}
|
|
}
|
|
// Metadata (when requested) is set via StartCopyFromURL options.Metadata
|
|
|
|
return f.NewObject(ctx, remote)
|
|
}
|
|
|
|
// Copy src to this remote using server-side copy operations.
|
|
//
|
|
// This is stored with the remote path given.
|
|
//
|
|
// It returns the destination Object and a possible error.
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantCopy
|
|
func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
|
|
dstContainer, dstPath := f.split(remote)
|
|
err := f.mkdirParent(ctx, remote)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
srcObj, ok := src.(*Object)
|
|
if !ok {
|
|
fs.Debugf(src, "Can't copy - not same remote type")
|
|
return nil, fs.ErrorCantCopy
|
|
}
|
|
|
|
// Assume we are copying to a different storage account if we
|
|
// are copying across configs.
|
|
sameStorageAccount := f == srcObj.fs
|
|
|
|
// If we are using Microsoft Entra ID token based auth then
|
|
// copySinglepart does not work
|
|
usingEntraID := f.cred != nil
|
|
|
|
// Use multipart copy if size > cutoff
|
|
// or using Entra ID and we are not using the same storage account
|
|
useMultiPart := srcObj.size >= int64(f.opt.CopyCutoff) || (usingEntraID && !sameStorageAccount)
|
|
|
|
// Force the use of copy blob if on the same storage account
|
|
// and the user hasn't forbidden it.
|
|
if f.opt.UseCopyBlob && sameStorageAccount {
|
|
useMultiPart = false
|
|
}
|
|
|
|
if useMultiPart {
|
|
return f.copyMultipart(ctx, remote, dstContainer, dstPath, srcObj)
|
|
}
|
|
return f.copySinglepart(ctx, remote, dstContainer, dstPath, srcObj)
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// Fs returns the parent Fs
|
|
func (o *Object) Fs() fs.Info {
|
|
return o.fs
|
|
}
|
|
|
|
// Return a string version
|
|
func (o *Object) String() string {
|
|
if o == nil {
|
|
return "<nil>"
|
|
}
|
|
return o.remote
|
|
}
|
|
|
|
// Remote returns the remote path
|
|
func (o *Object) Remote() string {
|
|
return o.remote
|
|
}
|
|
|
|
// Hash returns the MD5 of an object returning a lowercase hex string
|
|
func (o *Object) Hash(ctx context.Context, t hash.Type) (string, error) {
|
|
if t != hash.MD5 {
|
|
return "", hash.ErrUnsupported
|
|
}
|
|
// If decompressing, erase the hash
|
|
if o.size < 0 {
|
|
return "", nil
|
|
}
|
|
// Convert base64 encoded md5 into lower case hex
|
|
if o.md5 == "" {
|
|
return "", nil
|
|
}
|
|
data, err := base64.StdEncoding.DecodeString(o.md5)
|
|
if err != nil {
|
|
return "", fmt.Errorf("failed to decode Content-MD5: %q: %w", o.md5, err)
|
|
}
|
|
return hex.EncodeToString(data), nil
|
|
}
|
|
|
|
// Size returns the size of an object in bytes
|
|
func (o *Object) Size() int64 {
|
|
return o.size
|
|
}
|
|
|
|
// Set o.metadata from metadata
|
|
func (o *Object) setMetadata(metadata map[string]*string) {
|
|
metadataMu.Lock()
|
|
defer metadataMu.Unlock()
|
|
|
|
if len(metadata) > 0 {
|
|
// Lower case the metadata
|
|
o.meta = make(map[string]string, len(metadata))
|
|
for k, v := range metadata {
|
|
if v != nil {
|
|
o.meta[strings.ToLower(k)] = *v
|
|
}
|
|
}
|
|
// Set o.modTime from metadata if it exists and
|
|
// UseServerModTime isn't in use.
|
|
if modTime, ok := o.meta[modTimeKey]; !o.fs.ci.UseServerModTime && ok {
|
|
when, err := time.Parse(timeFormatIn, modTime)
|
|
if err != nil {
|
|
fs.Debugf(o, "Couldn't parse %v = %q: %v", modTimeKey, modTime, err)
|
|
}
|
|
o.modTime = when
|
|
}
|
|
} else {
|
|
o.meta = nil
|
|
}
|
|
}
|
|
|
|
// Get metadata from o.meta
|
|
func (o *Object) getMetadata() (metadata map[string]*string) {
|
|
metadataMu.Lock()
|
|
defer metadataMu.Unlock()
|
|
|
|
if len(o.meta) == 0 {
|
|
return nil
|
|
}
|
|
metadata = make(map[string]*string, len(o.meta))
|
|
for k, v := range o.meta {
|
|
metadata[k] = &v
|
|
}
|
|
return metadata
|
|
}
|
|
|
|
// Metadata returns metadata for an object
|
|
//
|
|
// It returns a combined view of system and user metadata.
|
|
func (o *Object) Metadata(ctx context.Context) (fs.Metadata, error) {
|
|
// Ensure metadata is loaded
|
|
if err := o.readMetaData(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
m := fs.Metadata{}
|
|
|
|
// System metadata we expose
|
|
if !o.modTime.IsZero() {
|
|
m["mtime"] = o.modTime.Format(time.RFC3339Nano)
|
|
}
|
|
if o.accessTier != "" {
|
|
m["tier"] = string(o.accessTier)
|
|
}
|
|
|
|
// Merge user metadata (already lower-cased keys)
|
|
metadataMu.Lock()
|
|
maps.Copy(m, o.meta)
|
|
metadataMu.Unlock()
|
|
|
|
return m, nil
|
|
}
|
|
|
|
// decodeMetaDataFromPropertiesResponse sets the metadata from the data passed in
|
|
//
|
|
// Sets
|
|
//
|
|
// o.id
|
|
// o.modTime
|
|
// o.size
|
|
// o.md5
|
|
// o.meta
|
|
func (o *Object) decodeMetaDataFromPropertiesResponse(info *blob.GetPropertiesResponse) (err error) {
|
|
metadata := info.Metadata
|
|
var size int64
|
|
if info.ContentLength == nil {
|
|
size = -1
|
|
} else {
|
|
size = *info.ContentLength
|
|
}
|
|
if isDirectoryMarker(size, metadata, o.remote) {
|
|
return fs.ErrorNotAFile
|
|
}
|
|
// NOTE - Client library always returns MD5 as base64 decoded string, Object needs to maintain
|
|
// this as base64 encoded string.
|
|
o.md5 = base64.StdEncoding.EncodeToString(info.ContentMD5)
|
|
if info.ContentType == nil {
|
|
o.mimeType = ""
|
|
} else {
|
|
o.mimeType = *info.ContentType
|
|
}
|
|
o.size = size
|
|
if info.LastModified == nil {
|
|
o.modTime = time.Now()
|
|
} else {
|
|
o.modTime = *info.LastModified
|
|
}
|
|
if info.AccessTier == nil {
|
|
o.accessTier = blob.AccessTier("")
|
|
} else {
|
|
o.accessTier = blob.AccessTier(*info.AccessTier)
|
|
}
|
|
o.setMetadata(metadata)
|
|
o.contentEncoding = info.ContentEncoding
|
|
|
|
// If decompressing then size and md5sum are unknown
|
|
if o.fs.opt.Decompress && o.contentEncoding != nil && *o.contentEncoding == "gzip" {
|
|
o.size = -1
|
|
o.md5 = ""
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (o *Object) decodeMetaDataFromDownloadResponse(info *blob.DownloadStreamResponse) (err error) {
|
|
metadata := info.Metadata
|
|
var size int64
|
|
if info.ContentLength == nil {
|
|
size = -1
|
|
} else {
|
|
size = *info.ContentLength
|
|
}
|
|
if isDirectoryMarker(size, metadata, o.remote) {
|
|
return fs.ErrorNotAFile
|
|
}
|
|
// NOTE - Client library always returns MD5 as base64 decoded string, Object needs to maintain
|
|
// this as base64 encoded string.
|
|
//
|
|
// On a range request the service returns the whole-blob MD5 in the
|
|
// x-ms-blob-content-md5 header (BlobContentMD5) and leaves Content-MD5
|
|
// (ContentMD5) empty, so prefer BlobContentMD5. Never overwrite a known
|
|
// hash with an empty one, or a ranged read would drop the object's MD5.
|
|
if md5 := info.BlobContentMD5; len(md5) > 0 {
|
|
o.md5 = base64.StdEncoding.EncodeToString(md5)
|
|
} else if len(info.ContentMD5) > 0 {
|
|
o.md5 = base64.StdEncoding.EncodeToString(info.ContentMD5)
|
|
}
|
|
if info.ContentType == nil {
|
|
o.mimeType = ""
|
|
} else {
|
|
o.mimeType = *info.ContentType
|
|
}
|
|
o.size = size
|
|
if info.LastModified == nil {
|
|
o.modTime = time.Now()
|
|
} else {
|
|
o.modTime = *info.LastModified
|
|
}
|
|
// FIXME response doesn't appear to have AccessTier in?
|
|
// if info.AccessTier == nil {
|
|
// o.accessTier = blob.AccessTier("")
|
|
// } else {
|
|
// o.accessTier = blob.AccessTier(*info.AccessTier)
|
|
// }
|
|
o.setMetadata(metadata)
|
|
|
|
// If it was a Range request, the size is wrong, so correct it
|
|
if info.ContentRange != nil {
|
|
contentRange := *info.ContentRange
|
|
slash := strings.IndexRune(contentRange, '/')
|
|
if slash >= 0 {
|
|
i, err := strconv.ParseInt(contentRange[slash+1:], 10, 64)
|
|
if err == nil {
|
|
o.size = i
|
|
} else {
|
|
fs.Debugf(o, "Failed to find parse integer from in %q: %v", contentRange, err)
|
|
}
|
|
} else {
|
|
fs.Debugf(o, "Failed to find length in %q", contentRange)
|
|
}
|
|
}
|
|
o.contentEncoding = info.ContentEncoding
|
|
|
|
// If decompressing then size and md5sum are unknown
|
|
if o.fs.opt.Decompress && o.contentEncoding != nil && *o.contentEncoding == "gzip" {
|
|
o.size = -1
|
|
o.md5 = ""
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (o *Object) decodeMetaDataFromBlob(info *container.BlobItem) (err error) {
|
|
if info.Properties == nil {
|
|
return errors.New("nil Properties in decodeMetaDataFromBlob")
|
|
}
|
|
metadata := info.Metadata
|
|
var size int64
|
|
if info.Properties.ContentLength == nil {
|
|
size = -1
|
|
} else {
|
|
size = *info.Properties.ContentLength
|
|
}
|
|
if isDirectoryMarker(size, metadata, o.remote) {
|
|
return fs.ErrorNotAFile
|
|
}
|
|
// NOTE - Client library always returns MD5 as base64 decoded string, Object needs to maintain
|
|
// this as base64 encoded string.
|
|
o.md5 = base64.StdEncoding.EncodeToString(info.Properties.ContentMD5)
|
|
if info.Properties.ContentType == nil {
|
|
o.mimeType = ""
|
|
} else {
|
|
o.mimeType = *info.Properties.ContentType
|
|
}
|
|
o.size = size
|
|
if info.Properties.LastModified == nil {
|
|
o.modTime = time.Now()
|
|
} else {
|
|
o.modTime = *info.Properties.LastModified
|
|
}
|
|
if info.Properties.AccessTier == nil {
|
|
o.accessTier = blob.AccessTier("")
|
|
} else {
|
|
o.accessTier = *info.Properties.AccessTier
|
|
}
|
|
o.setMetadata(metadata)
|
|
o.contentEncoding = info.Properties.ContentEncoding
|
|
|
|
// If decompressing then size and md5sum are unknown
|
|
if o.fs.opt.Decompress && o.contentEncoding != nil && *o.contentEncoding == "gzip" {
|
|
o.size = -1
|
|
o.md5 = ""
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (o *Object) getTags() (tags map[string]string) {
|
|
if o.tags != nil {
|
|
return o.tags
|
|
}
|
|
|
|
return map[string]string{}
|
|
}
|
|
|
|
// getBlobSVC creates a blob client
|
|
func (o *Object) getBlobSVC() *blob.Client {
|
|
container, directory := o.split()
|
|
return o.fs.getBlobSVC(container, directory)
|
|
}
|
|
|
|
// getBlockBlobSVC creates a block blob client
|
|
func (o *Object) getBlockBlobSVC() *blockblob.Client {
|
|
container, directory := o.split()
|
|
return o.fs.getBlockBlobSVC(container, directory)
|
|
}
|
|
|
|
// clearMetaData clears enough metadata so readMetaData will re-read it
|
|
func (o *Object) clearMetaData() {
|
|
o.modTime = time.Time{}
|
|
}
|
|
|
|
// readMetaData gets the metadata if it hasn't already been fetched
|
|
func (f *Fs) readMetaData(ctx context.Context, container, containerPath string) (blobProperties *blob.GetPropertiesResponse, err error) {
|
|
if !f.containerOK(container) {
|
|
return nil, fs.ErrorObjectNotFound
|
|
}
|
|
blb := f.getBlobSVC(container, containerPath)
|
|
|
|
// Read metadata (this includes metadata)
|
|
options := blob.GetPropertiesOptions{}
|
|
var resp blob.GetPropertiesResponse
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
resp, err = blb.GetProperties(ctx, &options)
|
|
return f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
// On directories - GetProperties does not work and current SDK does not populate service code correctly hence check regular http response as well
|
|
if storageErr, ok := err.(*azcore.ResponseError); ok && (storageErr.ErrorCode == string(bloberror.BlobNotFound) || storageErr.StatusCode == http.StatusNotFound) {
|
|
return nil, fs.ErrorObjectNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
return &resp, nil
|
|
}
|
|
|
|
// readMetaDataAlways gets the metadata unconditionally and also the blob properties.
|
|
//
|
|
// Sets
|
|
//
|
|
// o.id
|
|
// o.modTime
|
|
// o.size
|
|
// o.md5
|
|
func (o *Object) readMetaDataAlways(ctx context.Context) (blobProperties *blob.GetPropertiesResponse, err error) {
|
|
container, containerPath := o.split()
|
|
blobProperties, err = o.fs.readMetaData(ctx, container, containerPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
err = o.decodeMetaDataFromPropertiesResponse(blobProperties)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return blobProperties, nil
|
|
}
|
|
|
|
// readMetaData gets the metadata if it hasn't already been fetched
|
|
//
|
|
// Sets
|
|
//
|
|
// o.id
|
|
// o.modTime
|
|
// o.size
|
|
// o.md5
|
|
func (o *Object) readMetaData(ctx context.Context) (err error) {
|
|
if !o.modTime.IsZero() {
|
|
return nil
|
|
}
|
|
_, err = o.readMetaDataAlways(ctx)
|
|
return err
|
|
}
|
|
|
|
// ModTime returns the modification time of the object
|
|
//
|
|
// It attempts to read the objects mtime and if that isn't present the
|
|
// LastModified returned in the http headers
|
|
func (o *Object) ModTime(ctx context.Context) (result time.Time) {
|
|
// The error is logged in readMetaData
|
|
_ = o.readMetaData(ctx)
|
|
return o.modTime
|
|
}
|
|
|
|
// SetModTime sets the modification time of the local fs object
|
|
func (o *Object) SetModTime(ctx context.Context, modTime time.Time) error {
|
|
o.updateMetadataWithModTime(modTime)
|
|
|
|
blb := o.getBlobSVC()
|
|
opt := blob.SetMetadataOptions{}
|
|
err := o.fs.pacer.Call(func() (bool, error) {
|
|
_, err := blb.SetMetadata(ctx, o.getMetadata(), &opt)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
o.modTime = modTime
|
|
return nil
|
|
}
|
|
|
|
// Storable returns if this object is storable
|
|
func (o *Object) Storable() bool {
|
|
return true
|
|
}
|
|
|
|
// Open an object for read
|
|
func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.ReadCloser, err error) {
|
|
// Offset and Count for range download
|
|
var offset int64
|
|
var count int64
|
|
if o.AccessTier() == blob.AccessTierArchive {
|
|
return nil, fmt.Errorf("blob in archive tier, you need to set tier to hot, cool, cold first")
|
|
}
|
|
fs.FixRangeOption(options, o.size)
|
|
for _, option := range options {
|
|
switch x := option.(type) {
|
|
case *fs.RangeOption:
|
|
offset, count = x.Decode(o.size)
|
|
if count < 0 {
|
|
count = o.size - offset
|
|
}
|
|
case *fs.SeekOption:
|
|
offset = x.Offset
|
|
default:
|
|
if option.Mandatory() {
|
|
fs.Logf(o, "Unsupported mandatory option: %v", option)
|
|
}
|
|
}
|
|
}
|
|
blb := o.getBlobSVC()
|
|
opt := blob.DownloadStreamOptions{
|
|
// When set to true and specified together with the Range, the service returns the MD5 hash for the range, as long as the
|
|
// range is less than or equal to 4 MB in size.
|
|
//RangeGetContentMD5 *bool
|
|
|
|
// Range specifies a range of bytes. The default value is all bytes.
|
|
//Range HTTPRange
|
|
Range: blob.HTTPRange{
|
|
Offset: offset,
|
|
Count: count,
|
|
},
|
|
|
|
// AccessConditions *AccessConditions
|
|
// CpkInfo *CpkInfo
|
|
// CpkScopeInfo *CpkScopeInfo
|
|
}
|
|
var downloadResponse blob.DownloadStreamResponse
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
downloadResponse, err = blb.DownloadStream(ctx, &opt)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open for download: %w", err)
|
|
}
|
|
err = o.decodeMetaDataFromDownloadResponse(&downloadResponse)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to decode metadata for download: %w", err)
|
|
}
|
|
|
|
// Decompress body if necessary
|
|
if downloadResponse.ContentEncoding != nil && *downloadResponse.ContentEncoding == "gzip" {
|
|
if o.fs.opt.Decompress {
|
|
return readers.NewGzipReader(downloadResponse.Body)
|
|
}
|
|
o.fs.warnCompressed.Do(func() {
|
|
fs.Logf(o, "Not decompressing 'Content-Encoding: gzip' compressed file. Use --azureblob-decompress to override")
|
|
})
|
|
}
|
|
|
|
return downloadResponse.Body, nil
|
|
}
|
|
|
|
// Converts a string into a pointer to a string
|
|
func pString(s string) *string {
|
|
return &s
|
|
}
|
|
|
|
// readSeekCloser joins an io.Reader and an io.Seeker and provides a no-op io.Closer
|
|
type readSeekCloser struct {
|
|
io.Reader
|
|
io.Seeker
|
|
}
|
|
|
|
// Close does nothing
|
|
func (rs *readSeekCloser) Close() error {
|
|
return nil
|
|
}
|
|
|
|
// a creator for blockIDs with an incrementing part and a random part
|
|
//
|
|
// The random part is to make sure that blockIDs don't collide between
|
|
// uploads. We need block IDs not to be shared between upload attempts
|
|
// so we can remove the uncommitted blocks properly on errors.
|
|
type blockIDCreator struct {
|
|
random [8]byte // randomness to make sure blocks don't collide
|
|
}
|
|
|
|
// create a new blockID creator with a random suffix
|
|
func newBlockIDCreator() (bic *blockIDCreator, err error) {
|
|
bic = &blockIDCreator{}
|
|
n, err := rand.Read(bic.random[:])
|
|
if err != nil {
|
|
return nil, fmt.Errorf("crypto rand failed: %w", err)
|
|
}
|
|
if n != len(bic.random) {
|
|
return nil, errors.New("crypto rand failed: short read")
|
|
}
|
|
return bic, nil
|
|
}
|
|
|
|
// create a new block ID for chunkNumber
|
|
func (bic *blockIDCreator) newBlockID(chunkNumber uint64) string {
|
|
var binaryBlockID [16]byte
|
|
// block counter as LSB first 8 bytes
|
|
binary.BigEndian.PutUint64(binaryBlockID[:8], chunkNumber)
|
|
// random bits at the end
|
|
copy(binaryBlockID[8:], bic.random[:])
|
|
// return base64 encoded value
|
|
return base64.StdEncoding.EncodeToString(binaryBlockID[:])
|
|
}
|
|
|
|
// Check the chunkNumber is correct in the id
|
|
func (bic *blockIDCreator) checkID(chunkNumber uint64, id string) error {
|
|
binaryBlockID, err := base64.StdEncoding.DecodeString(id)
|
|
if err != nil {
|
|
return fmt.Errorf("internal error: bad block ID: %w", err)
|
|
}
|
|
if len(binaryBlockID) != 16 {
|
|
return errors.New("internal error: bad block ID length")
|
|
}
|
|
gotChunkNumber := binary.BigEndian.Uint64(binaryBlockID[:8])
|
|
if chunkNumber != gotChunkNumber {
|
|
return fmt.Errorf("internal error: expecting decoded chunkNumber %d but got %d", chunkNumber, gotChunkNumber)
|
|
}
|
|
if !bytes.Equal(binaryBlockID[8:], bic.random[:]) {
|
|
return fmt.Errorf("internal error: random bytes are incorrect")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// record chunk number and id for Close
|
|
type azBlock struct {
|
|
chunkNumber uint64
|
|
id string
|
|
}
|
|
|
|
// Implements the fs.ChunkWriter interface
|
|
type azChunkWriter struct {
|
|
chunkSize int64
|
|
size int64
|
|
f *Fs
|
|
ui uploadInfo
|
|
blocksMu sync.Mutex // protects the below
|
|
blocks []azBlock // list of blocks for finalize
|
|
o *Object
|
|
bic *blockIDCreator
|
|
checker *checkForInvalidBlockOrBlob
|
|
}
|
|
|
|
// OpenChunkWriter returns the chunk size and a ChunkWriter
|
|
//
|
|
// Pass in the remote and the src object
|
|
// You can also use options to hint at the desired chunk size
|
|
func (f *Fs) OpenChunkWriter(ctx context.Context, remote string, src fs.ObjectInfo, options ...fs.OpenOption) (info fs.ChunkWriterInfo, writer fs.ChunkWriter, err error) {
|
|
// Temporary Object under construction
|
|
o := &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
}
|
|
ui, err := o.prepareUpload(ctx, src, options)
|
|
if err != nil {
|
|
return info, nil, fmt.Errorf("failed to prepare upload: %w", err)
|
|
}
|
|
|
|
// Calculate correct partSize
|
|
partSize := f.opt.ChunkSize
|
|
totalParts := -1
|
|
size := src.Size()
|
|
|
|
// Note that the max size of file is 4.75 TB (100 MB X 50,000
|
|
// blocks) and this is bigger than the max uncommitted block
|
|
// size (9.52 TB) so we do not need to part commit block lists
|
|
// or garbage collect uncommitted blocks.
|
|
//
|
|
// See: https://docs.microsoft.com/en-gb/rest/api/storageservices/put-block
|
|
|
|
// size can be -1 here meaning we don't know the size of the incoming file. We use ChunkSize
|
|
// buffers here (default 4MB). With a maximum number of parts (50,000) this will be a file of
|
|
// 195GB which seems like a not too unreasonable limit.
|
|
if size == -1 {
|
|
warnStreamUpload.Do(func() {
|
|
fs.Logf(f, "Streaming uploads using chunk size %v will have maximum file size of %v",
|
|
f.opt.ChunkSize, partSize*fs.SizeSuffix(blockblob.MaxBlocks))
|
|
})
|
|
} else {
|
|
partSize = chunksize.Calculator(remote, size, blockblob.MaxBlocks, f.opt.ChunkSize)
|
|
if partSize > fs.SizeSuffix(blockblob.MaxStageBlockBytes) {
|
|
return info, nil, fmt.Errorf("can't upload as it is too big %v - takes more than %d chunks of %v", fs.SizeSuffix(size), fs.SizeSuffix(blockblob.MaxBlocks), fs.SizeSuffix(blockblob.MaxStageBlockBytes))
|
|
}
|
|
totalParts = int(fs.SizeSuffix(size) / partSize)
|
|
if fs.SizeSuffix(size)%partSize != 0 {
|
|
totalParts++
|
|
}
|
|
}
|
|
|
|
fs.Debugf(o, "Multipart upload session started for %d parts of size %v", totalParts, partSize)
|
|
|
|
chunkWriter := &azChunkWriter{
|
|
chunkSize: int64(partSize),
|
|
size: size,
|
|
f: f,
|
|
ui: ui,
|
|
o: o,
|
|
checker: newCheckForInvalidBlockOrBlob("upload", o),
|
|
}
|
|
info = fs.ChunkWriterInfo{
|
|
ChunkSize: int64(partSize),
|
|
Concurrency: o.fs.opt.UploadConcurrency,
|
|
//LeavePartsOnError: o.fs.opt.LeavePartsOnError,
|
|
}
|
|
chunkWriter.bic, err = newBlockIDCreator()
|
|
if err != nil {
|
|
return info, nil, err
|
|
}
|
|
fs.Debugf(o, "open chunk writer: started multipart upload")
|
|
return info, chunkWriter, nil
|
|
}
|
|
|
|
// isInvalidBlockOrBlob looks for the InvalidBlockOrBlob error in err
|
|
// returning true if it is found
|
|
func isInvalidBlockOrBlob(err error) bool {
|
|
var storageErr *azcore.ResponseError
|
|
if errors.As(err, &storageErr) {
|
|
return storageErr.ErrorCode == string(bloberror.InvalidBlobOrBlock)
|
|
}
|
|
return false
|
|
}
|
|
|
|
// Struct to hold state for checking for InvalidBlockOrBlob
|
|
type checkForInvalidBlockOrBlob struct {
|
|
startMu sync.Mutex // hold when starting transactions
|
|
inFlight sync.WaitGroup // transactions in flight
|
|
what string // "copy" or "upload"
|
|
o *Object // object we are working on
|
|
cleared bool // set if we have cleared the uncommitted blocks - we only do this once
|
|
}
|
|
|
|
// Make InvalidBlockOrBlob checker
|
|
func newCheckForInvalidBlockOrBlob(what string, o *Object) *checkForInvalidBlockOrBlob {
|
|
return &checkForInvalidBlockOrBlob{
|
|
what: what,
|
|
o: o,
|
|
}
|
|
}
|
|
|
|
// start marks that there is a transaction in progress
|
|
func (c *checkForInvalidBlockOrBlob) start() {
|
|
c.startMu.Lock()
|
|
defer c.startMu.Unlock()
|
|
c.inFlight.Add(1)
|
|
}
|
|
|
|
// stop marks that this transaction has finished
|
|
func (c *checkForInvalidBlockOrBlob) stop() {
|
|
c.inFlight.Done()
|
|
}
|
|
|
|
// checkErr looks for the InvalidBlockOrBlob error in err, and if it
|
|
// is found, it clears uncommitted blocks in o to clear the error.
|
|
//
|
|
// It returns a bool indicating whether the error was found or not.
|
|
//
|
|
// See https://gauravmantri.com/2013/05/18/windows-azure-blob-storage-dealing-with-the-specified-blob-or-block-content-is-invalid-error/
|
|
func (c *checkForInvalidBlockOrBlob) checkErr(ctx context.Context, err error) (result bool) {
|
|
// defer log.Trace(c.o, "err=%#v, what=%q", err, c.what)("result=%v", &result)
|
|
if !isInvalidBlockOrBlob(err) {
|
|
return false
|
|
}
|
|
|
|
// Prevent more transactions starting
|
|
c.startMu.Lock()
|
|
defer c.startMu.Unlock()
|
|
|
|
if c.cleared {
|
|
fs.Debugf(c.o, "multipart %s: received %s error: already cleared", c.what, bloberror.InvalidBlobOrBlock)
|
|
return true
|
|
}
|
|
|
|
// Wait for any other outstanding transactions to finish
|
|
c.inFlight.Wait()
|
|
|
|
// Clear uncommitted blocks
|
|
fs.Debugf(c.o, "multipart %s: received %s error: clearing uncommitted blocks and retrying", c.what, bloberror.InvalidBlobOrBlock)
|
|
clearErr := c.o.clearUncommittedBlocks(ctx)
|
|
if clearErr != nil {
|
|
fs.Debugf(c.o, "multipart %s: error fixing %s: %v", c.what, bloberror.InvalidBlobOrBlock, clearErr)
|
|
}
|
|
fs.Debugf(c.o, "multipart %s: fixed %s", c.what, bloberror.InvalidBlobOrBlock)
|
|
c.cleared = true
|
|
|
|
return true
|
|
}
|
|
|
|
// WriteChunk will write chunk number with reader bytes, where chunk number >= 0
|
|
func (w *azChunkWriter) WriteChunk(ctx context.Context, chunkNumber int, reader io.ReadSeeker) (int64, error) {
|
|
if chunkNumber < 0 {
|
|
err := fmt.Errorf("invalid chunk number provided: %v", chunkNumber)
|
|
return -1, err
|
|
}
|
|
|
|
// Only account after the checksum reads have been done
|
|
if do, ok := reader.(pool.DelayAccountinger); ok {
|
|
// To figure out this number, do a transfer and if the accounted size is 0 or a
|
|
// multiple of what it should be, increase or decrease this number.
|
|
do.DelayAccounting(2)
|
|
}
|
|
|
|
// Upload the block, with MD5 for check
|
|
m := md5.New()
|
|
currentChunkSize, err := io.Copy(m, reader)
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
// If no data read, don't write the chunk
|
|
if currentChunkSize == 0 {
|
|
return 0, nil
|
|
}
|
|
md5sum := m.Sum(nil)
|
|
|
|
// Create a new blockID
|
|
blockID := w.bic.newBlockID(uint64(chunkNumber))
|
|
|
|
// Save the blockID for the commit
|
|
w.blocksMu.Lock()
|
|
w.blocks = append(w.blocks, azBlock{
|
|
chunkNumber: uint64(chunkNumber),
|
|
id: blockID,
|
|
})
|
|
w.blocksMu.Unlock()
|
|
|
|
err = w.f.pacer.Call(func() (bool, error) {
|
|
// rewind the reader on retry and after reading md5
|
|
_, err = reader.Seek(0, io.SeekStart)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
options := blockblob.StageBlockOptions{
|
|
// Specify the transactional md5 for the body, to be validated by the service.
|
|
TransactionalValidation: blob.TransferValidationTypeMD5(md5sum),
|
|
}
|
|
w.checker.start()
|
|
_, err = w.ui.blb.StageBlock(ctx, blockID, &readSeekCloser{Reader: reader, Seeker: reader}, &options)
|
|
w.checker.stop()
|
|
if err != nil {
|
|
if w.checker.checkErr(ctx, err) {
|
|
return true, err
|
|
}
|
|
if chunkNumber <= 8 {
|
|
return w.f.shouldRetry(ctx, err)
|
|
}
|
|
if fserrors.ContextError(ctx, &err) {
|
|
return false, err
|
|
}
|
|
// retry all chunks once have done the first few
|
|
return true, err
|
|
}
|
|
return false, nil
|
|
})
|
|
if err != nil {
|
|
return -1, fmt.Errorf("failed to upload chunk %d with %v bytes: %w", chunkNumber+1, currentChunkSize, err)
|
|
}
|
|
|
|
fs.Debugf(w.o, "multipart upload wrote chunk %d with %v bytes", chunkNumber+1, currentChunkSize)
|
|
return currentChunkSize, err
|
|
}
|
|
|
|
// Clear uncommitted blocks
|
|
//
|
|
// There isn't an API to clear uncommitted blocks.
|
|
//
|
|
// However they are released when a Commit is called. Doing this will
|
|
// instantiate the object so we don't want to overwrite an existing
|
|
// object.
|
|
//
|
|
// We will use this algorithm:
|
|
//
|
|
// Attempt to read committed blocks from the object
|
|
// If the object exists
|
|
// - Commit the existing blocks again
|
|
// - This should get rid of the uncommitted blocks without changing the existing object
|
|
// If the object does not exist then
|
|
// - Commit an empty block list
|
|
// - This will get rid of the uncommitted blocks
|
|
// - This will also create a 0 length blob
|
|
// - So delete the 0 length blob
|
|
func (o *Object) clearUncommittedBlocks(ctx context.Context) (err error) {
|
|
fs.Debugf(o, "Clearing uncommitted blocks")
|
|
var (
|
|
blockBlobSVC = o.getBlockBlobSVC()
|
|
objectExists = true
|
|
blockIDs []string
|
|
blockList blockblob.GetBlockListResponse
|
|
properties *blob.GetPropertiesResponse
|
|
options *blockblob.CommitBlockListOptions
|
|
)
|
|
|
|
properties, err = o.readMetaDataAlways(ctx)
|
|
if err == fs.ErrorObjectNotFound {
|
|
objectExists = false
|
|
} else if err != nil {
|
|
return fmt.Errorf("clear uncommitted blocks: failed to read metadata: %w", err)
|
|
}
|
|
|
|
if objectExists {
|
|
// Get the committed block list
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
blockList, err = blockBlobSVC.GetBlockList(ctx, blockblob.BlockListTypeAll, nil)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("clear uncommitted blocks: failed to read uncommitted block list: %w", err)
|
|
}
|
|
if len(blockList.UncommittedBlocks) == 0 {
|
|
fs.Debugf(o, "No uncommitted blocks - exiting")
|
|
return nil
|
|
}
|
|
fs.Debugf(o, "%d Uncommitted blocks found", len(blockList.UncommittedBlocks))
|
|
objectExists = true
|
|
uncommittedBlocks := make(map[string]struct{}, len(blockList.UncommittedBlocks))
|
|
for _, block := range blockList.UncommittedBlocks {
|
|
uncommittedBlocks[*block.Name] = struct{}{}
|
|
}
|
|
for _, block := range blockList.CommittedBlocks {
|
|
name := *block.Name
|
|
if _, found := uncommittedBlocks[name]; found {
|
|
return fmt.Errorf("clear uncommitted blocks: can't safely clear uncommitted blocks as committed and uncommitted IDs overlap. Delete the existing object to clear the uncommitted blocks")
|
|
}
|
|
blockIDs = append(blockIDs, name)
|
|
}
|
|
|
|
// Reconstruct metadata from existing object as CommitBlockList overwrites it
|
|
options = &blockblob.CommitBlockListOptions{
|
|
Metadata: properties.Metadata,
|
|
Tier: (*blob.AccessTier)(properties.AccessTier),
|
|
HTTPHeaders: &blob.HTTPHeaders{
|
|
BlobCacheControl: properties.CacheControl,
|
|
BlobContentDisposition: properties.ContentDisposition,
|
|
BlobContentEncoding: properties.ContentEncoding,
|
|
BlobContentLanguage: properties.ContentLanguage,
|
|
BlobContentMD5: properties.ContentMD5,
|
|
BlobContentType: properties.ContentType,
|
|
},
|
|
}
|
|
}
|
|
|
|
// Commit only the committed blocks
|
|
fs.Debugf(o, "Committing %d blocks to remove uncommitted blocks", len(blockIDs))
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
_, err := blockBlobSVC.CommitBlockList(ctx, blockIDs, options)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("clear uncommitted blocks: failed to commit block list: %w", err)
|
|
}
|
|
|
|
// If object didn't exist before, then delete it
|
|
if !objectExists {
|
|
fs.Debugf(o, "Removing empty object")
|
|
err = o.Remove(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("clear uncommitted blocks: failed to remove empty object: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Abort the multipart upload.
|
|
func (w *azChunkWriter) Abort(ctx context.Context) error {
|
|
return w.o.clearUncommittedBlocks(ctx)
|
|
}
|
|
|
|
// Close and finalise the multipart upload
|
|
func (w *azChunkWriter) Close(ctx context.Context) (err error) {
|
|
// sort the completed parts by part number
|
|
sort.Slice(w.blocks, func(i, j int) bool {
|
|
return w.blocks[i].chunkNumber < w.blocks[j].chunkNumber
|
|
})
|
|
|
|
// Create and check a list of block IDs
|
|
blockIDs := make([]string, len(w.blocks))
|
|
for i := range w.blocks {
|
|
if w.blocks[i].chunkNumber != uint64(i) {
|
|
return fmt.Errorf("internal error: expecting chunkNumber %d but got %d", i, w.blocks[i].chunkNumber)
|
|
}
|
|
err := w.bic.checkID(w.blocks[i].chunkNumber, w.blocks[i].id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
blockIDs[i] = w.blocks[i].id
|
|
}
|
|
|
|
options := blockblob.CommitBlockListOptions{
|
|
Metadata: w.o.getMetadata(),
|
|
Tags: w.o.getTags(),
|
|
Tier: parseTier(w.f.opt.AccessTier),
|
|
HTTPHeaders: &w.ui.httpHeaders,
|
|
}
|
|
|
|
// Finalise the upload session
|
|
err = w.f.pacer.Call(func() (bool, error) {
|
|
_, err := w.ui.blb.CommitBlockList(ctx, blockIDs, &options)
|
|
return w.f.shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("failed to complete multipart upload: %w", err)
|
|
}
|
|
fs.Debugf(w.o, "multipart upload finished")
|
|
return err
|
|
}
|
|
|
|
var warnStreamUpload sync.Once
|
|
|
|
// uploadMultipart uploads a file using multipart upload
|
|
//
|
|
// Write a larger blob, using CreateBlockBlob, PutBlock, and PutBlockList.
|
|
func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (ui uploadInfo, err error) {
|
|
chunkWriter, err := multipart.UploadMultipart(ctx, src, in, multipart.UploadMultipartOptions{
|
|
Open: o.fs,
|
|
OpenOptions: options,
|
|
})
|
|
if err != nil {
|
|
return ui, err
|
|
}
|
|
return chunkWriter.(*azChunkWriter).ui, nil
|
|
}
|
|
|
|
// uploadSinglepart uploads a short blob using a single part upload
|
|
func (o *Object) uploadSinglepart(ctx context.Context, in io.Reader, size int64, ui uploadInfo) (err error) {
|
|
chunkSize := int64(o.fs.opt.ChunkSize)
|
|
// fs.Debugf(o, "Single part upload starting of object %d bytes", size)
|
|
if size > chunkSize || size < 0 {
|
|
return fmt.Errorf("internal error: single part upload size too big %d > %d", size, chunkSize)
|
|
}
|
|
|
|
rw := multipart.NewRW()
|
|
defer fs.CheckClose(rw, &err)
|
|
|
|
n, err := io.CopyN(rw, in, size+1)
|
|
if err != nil && err != io.EOF {
|
|
return fmt.Errorf("single part upload read failed: %w", err)
|
|
}
|
|
if n != size {
|
|
return fmt.Errorf("single part upload: expecting to read %d bytes but read %d", size, n)
|
|
}
|
|
|
|
rs := &readSeekCloser{Reader: rw, Seeker: rw}
|
|
|
|
options := blockblob.UploadOptions{
|
|
Metadata: o.getMetadata(),
|
|
Tags: o.getTags(),
|
|
Tier: parseTier(o.fs.opt.AccessTier),
|
|
HTTPHeaders: &ui.httpHeaders,
|
|
}
|
|
|
|
return o.fs.pacer.Call(func() (bool, error) {
|
|
// rewind the reader on retry
|
|
_, err = rs.Seek(0, io.SeekStart)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
_, err = ui.blb.Upload(ctx, rs, &options)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
}
|
|
|
|
// Info needed for an upload
|
|
type uploadInfo struct {
|
|
blb *blockblob.Client
|
|
httpHeaders blob.HTTPHeaders
|
|
isDirMarker bool
|
|
}
|
|
|
|
// Prepare the object for upload
|
|
func (o *Object) prepareUpload(ctx context.Context, src fs.ObjectInfo, options []fs.OpenOption) (ui uploadInfo, err error) {
|
|
container, containerPath := o.split()
|
|
if container == "" || containerPath == "" {
|
|
return ui, fmt.Errorf("can't upload to root - need a container")
|
|
}
|
|
// Create parent dir/bucket if not saving directory marker
|
|
metadataMu.Lock()
|
|
_, ui.isDirMarker = o.meta[dirMetaKey]
|
|
metadataMu.Unlock()
|
|
if !ui.isDirMarker {
|
|
err = o.fs.mkdirParent(ctx, o.remote)
|
|
if err != nil {
|
|
return ui, err
|
|
}
|
|
}
|
|
// if ui.isDirMarker && strings.HasSuffix(containerPath, "//") {
|
|
// containerPath = containerPath[:len(containerPath)-1]
|
|
// }
|
|
|
|
// Start with default content-type based on source
|
|
ui.httpHeaders = blob.HTTPHeaders{
|
|
BlobContentType: pString(fs.MimeType(ctx, src)),
|
|
}
|
|
|
|
// Apply mapped metadata/headers/tags if requested
|
|
modTime, err := o.applyMappedMetadata(ctx, src, &ui, options)
|
|
if err != nil {
|
|
return ui, err
|
|
}
|
|
// Ensure mtime is set in metadata based on possibly overridden modTime
|
|
o.updateMetadataWithModTime(modTime)
|
|
|
|
// Compute the Content-MD5 of the file. As we stream all uploads it
|
|
// will be set in PutBlockList API call using the 'x-ms-blob-content-md5' header
|
|
if !o.fs.opt.DisableCheckSum {
|
|
if sourceMD5, _ := src.Hash(ctx, hash.MD5); sourceMD5 != "" {
|
|
sourceMD5bytes, err := hex.DecodeString(sourceMD5)
|
|
if err == nil {
|
|
ui.httpHeaders.BlobContentMD5 = sourceMD5bytes
|
|
} else {
|
|
fs.Debugf(o, "Failed to decode %q as MD5: %v", sourceMD5, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Apply upload options (also allows one to overwrite content-type)
|
|
for _, option := range options {
|
|
key, value := option.Header()
|
|
lowerKey := strings.ToLower(key)
|
|
switch lowerKey {
|
|
case "":
|
|
// ignore
|
|
case "x-ms-tags":
|
|
if o.tags == nil {
|
|
o.tags = make(map[string]string)
|
|
}
|
|
|
|
tags := strings.Split(value, ",")
|
|
for _, tag := range tags {
|
|
parts := strings.SplitN(tag, "=", 2)
|
|
if len(parts) != 2 {
|
|
return ui, fmt.Errorf("invalid tag %q", tag)
|
|
}
|
|
|
|
o.tags[parts[0]] = parts[1]
|
|
}
|
|
case "cache-control":
|
|
ui.httpHeaders.BlobCacheControl = pString(value)
|
|
case "content-disposition":
|
|
ui.httpHeaders.BlobContentDisposition = pString(value)
|
|
case "content-encoding":
|
|
ui.httpHeaders.BlobContentEncoding = pString(value)
|
|
case "content-language":
|
|
ui.httpHeaders.BlobContentLanguage = pString(value)
|
|
case "content-type":
|
|
ui.httpHeaders.BlobContentType = pString(value)
|
|
}
|
|
}
|
|
|
|
ui.blb = o.fs.getBlockBlobSVC(container, containerPath)
|
|
return ui, nil
|
|
}
|
|
|
|
// Update the object with the contents of the io.Reader, modTime and size
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (err error) {
|
|
if o.accessTier == blob.AccessTierArchive {
|
|
if o.fs.opt.ArchiveTierDelete {
|
|
fs.Debugf(o, "deleting archive tier blob before updating")
|
|
err = o.Remove(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to delete archive blob before updating: %w", err)
|
|
}
|
|
} else {
|
|
return errCantUpdateArchiveTierBlobs
|
|
}
|
|
}
|
|
|
|
size := src.Size()
|
|
multipartUpload := size < 0 || size > int64(o.fs.opt.ChunkSize)
|
|
var ui uploadInfo
|
|
|
|
if multipartUpload {
|
|
ui, err = o.uploadMultipart(ctx, in, src, options...)
|
|
} else {
|
|
ui, err = o.prepareUpload(ctx, src, options)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare upload: %w", err)
|
|
}
|
|
err = o.uploadSinglepart(ctx, in, size, ui)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Refresh metadata on object
|
|
if !ui.isDirMarker {
|
|
o.clearMetaData()
|
|
err = o.readMetaData(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// If tier is not changed or not specified, do not attempt to invoke `SetBlobTier` operation
|
|
if o.fs.opt.AccessTier == string(defaultAccessTier) || o.fs.opt.AccessTier == string(o.AccessTier()) {
|
|
return nil
|
|
}
|
|
|
|
// Now, set blob tier based on configured access tier
|
|
return o.SetTier(o.fs.opt.AccessTier)
|
|
}
|
|
|
|
// Remove an object
|
|
func (o *Object) Remove(ctx context.Context) error {
|
|
blb := o.getBlobSVC()
|
|
opt := blob.DeleteOptions{}
|
|
if o.fs.opt.DeleteSnapshots != "" {
|
|
action := blob.DeleteSnapshotsOptionType(o.fs.opt.DeleteSnapshots)
|
|
opt.DeleteSnapshots = &action
|
|
}
|
|
return o.fs.pacer.Call(func() (bool, error) {
|
|
_, err := blb.Delete(ctx, &opt)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
}
|
|
|
|
// MimeType of an Object if known, "" otherwise
|
|
func (o *Object) MimeType(ctx context.Context) string {
|
|
return o.mimeType
|
|
}
|
|
|
|
// AccessTier of an object, default is of type none
|
|
func (o *Object) AccessTier() blob.AccessTier {
|
|
return o.accessTier
|
|
}
|
|
|
|
// SetTier performs changing object tier
|
|
func (o *Object) SetTier(tier string) error {
|
|
if !validateAccessTier(tier) {
|
|
return fmt.Errorf("tier %s not supported by Azure Blob Storage", tier)
|
|
}
|
|
|
|
// Check if current tier already matches with desired tier
|
|
if o.GetTier() == tier {
|
|
return nil
|
|
}
|
|
desiredAccessTier := blob.AccessTier(tier)
|
|
blb := o.getBlobSVC()
|
|
ctx := context.Background()
|
|
priority := blob.RehydratePriorityStandard
|
|
opt := blob.SetTierOptions{
|
|
RehydratePriority: &priority,
|
|
}
|
|
err := o.fs.pacer.Call(func() (bool, error) {
|
|
_, err := blb.SetTier(ctx, desiredAccessTier, &opt)
|
|
return o.fs.shouldRetry(ctx, err)
|
|
})
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("failed to set Blob Tier: %w", err)
|
|
}
|
|
|
|
// Set access tier on local object also, this typically
|
|
// gets updated on get blob properties
|
|
o.accessTier = desiredAccessTier
|
|
fs.Debugf(o, "Successfully changed object tier to %s", tier)
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetTier returns object tier in azure as string
|
|
func (o *Object) GetTier() string {
|
|
return string(o.accessTier)
|
|
}
|
|
|
|
func parseTier(tier string) *blob.AccessTier {
|
|
if tier == "" {
|
|
return nil
|
|
}
|
|
msTier := blob.AccessTier(tier)
|
|
return &msTier
|
|
}
|
|
|
|
// Check the interfaces are satisfied
|
|
var (
|
|
_ fs.Fs = &Fs{}
|
|
_ fs.Copier = &Fs{}
|
|
_ fs.PutStreamer = &Fs{}
|
|
_ fs.Purger = &Fs{}
|
|
_ fs.ListRer = &Fs{}
|
|
_ fs.ListPer = &Fs{}
|
|
_ fs.OpenChunkWriter = &Fs{}
|
|
_ fs.Object = &Object{}
|
|
_ fs.MimeTyper = &Object{}
|
|
_ fs.GetTierer = &Object{}
|
|
_ fs.SetTierer = &Object{}
|
|
)
|