seaweedfs/weed/remote_storage/azure/azure_storage_client.go

217 lines
7.2 KiB
Go
Raw Normal View History

2021-08-24 08:18:30 +00:00
package azure
import (
"context"
"fmt"
"github.com/Azure/azure-storage-blob-go/azblob"
"github.com/chrislusf/seaweedfs/weed/filer"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2021-08-26 22:18:34 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
2021-08-24 08:18:30 +00:00
"github.com/chrislusf/seaweedfs/weed/remote_storage"
"github.com/chrislusf/seaweedfs/weed/util"
"io"
"io/ioutil"
"net/url"
"os"
"reflect"
)
func init() {
remote_storage.RemoteStorageClientMakers["azure"] = new(azureRemoteStorageMaker)
}
type azureRemoteStorageMaker struct{}
func (s azureRemoteStorageMaker) HasBucket() bool {
return true
}
2021-08-26 22:18:34 +00:00
func (s azureRemoteStorageMaker) Make(conf *remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, error) {
2021-08-24 08:18:30 +00:00
client := &azureRemoteStorageClient{
conf: conf,
}
accountName, accountKey := conf.AzureAccountName, conf.AzureAccountKey
if len(accountName) == 0 || len(accountKey) == 0 {
accountName, accountKey = os.Getenv("AZURE_STORAGE_ACCOUNT"), os.Getenv("AZURE_STORAGE_ACCESS_KEY")
if len(accountName) == 0 || len(accountKey) == 0 {
return nil, fmt.Errorf("either AZURE_STORAGE_ACCOUNT or AZURE_STORAGE_ACCESS_KEY environment variable is not set")
}
}
// Use your Storage account's name and key to create a credential object.
credential, err := azblob.NewSharedKeyCredential(accountName, accountKey)
if err != nil {
return nil, fmt.Errorf("invalid Azure credential with account name:%s: %v", accountName, err)
}
// Create a request pipeline that is used to process HTTP(S) requests and responses.
p := azblob.NewPipeline(credential, azblob.PipelineOptions{})
// Create an ServiceURL object that wraps the service URL and a request pipeline.
u, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net", accountName))
client.serviceURL = azblob.NewServiceURL(*u, p)
return client, nil
}
type azureRemoteStorageClient struct {
2021-08-26 22:18:34 +00:00
conf *remote_pb.RemoteConf
2021-08-24 08:18:30 +00:00
serviceURL azblob.ServiceURL
}
var _ = remote_storage.RemoteStorageClient(&azureRemoteStorageClient{})
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
2021-08-24 08:18:30 +00:00
pathKey := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
// List the container that we have created above
for marker := (azblob.Marker{}); marker.NotDone(); {
// Get a result segment starting with the blob indicated by the current Marker.
listBlob, err := containerURL.ListBlobsFlatSegment(context.Background(), marker, azblob.ListBlobsSegmentOptions{
Prefix: pathKey,
})
if err != nil {
return fmt.Errorf("azure traverse %s%s: %v", loc.Bucket, loc.Path, err)
}
// ListBlobs returns the start of the next segment; you MUST use this to get
// the next segment (after processing the current result segment).
marker = listBlob.NextMarker
// Process the blobs returned in this result segment (if the segment is empty, the loop body won't execute)
for _, blobInfo := range listBlob.Segment.BlobItems {
key := blobInfo.Name
key = "/" + key
dir, name := util.FullPath(key).DirAndName()
err = visitFn(dir, name, false, &filer_pb.RemoteEntry{
RemoteMtime: blobInfo.Properties.LastModified.Unix(),
RemoteSize: *blobInfo.Properties.ContentLength,
RemoteETag: string(blobInfo.Properties.Etag),
StorageName: az.conf.Name,
})
if err != nil {
return fmt.Errorf("azure processing %s%s: %v", loc.Bucket, loc.Path, err)
}
}
}
return
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) {
2021-08-24 08:18:30 +00:00
key := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
blobURL := containerURL.NewBlockBlobURL(key)
downloadResponse, readErr := blobURL.Download(context.Background(), offset, size, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
if readErr != nil {
return nil, readErr
}
// NOTE: automatically retries are performed if the connection fails
bodyStream := downloadResponse.Body(azblob.RetryReaderOptions{MaxRetryRequests: 20})
defer bodyStream.Close()
data, err = ioutil.ReadAll(bodyStream)
if err != nil {
return nil, fmt.Errorf("failed to download file %s%s: %v", loc.Bucket, loc.Path, err)
}
return
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
2021-08-24 08:18:30 +00:00
return nil
}
func (az *azureRemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) {
return nil
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (remoteEntry *filer_pb.RemoteEntry, err error) {
2021-08-24 08:18:30 +00:00
key := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
blobURL := containerURL.NewBlockBlobURL(key)
readerAt, ok := reader.(io.ReaderAt)
if !ok {
return nil, fmt.Errorf("unexpected reader: readerAt expected")
}
fileSize := int64(filer.FileSize(entry))
_, err = uploadReaderAtToBlockBlob(context.Background(), readerAt, fileSize, blobURL, azblob.UploadToBlockBlobOptions{
BlockSize: 4 * 1024 * 1024,
Parallelism: 16})
if err != nil {
return nil, fmt.Errorf("azure upload to %s%s: %v", loc.Bucket, loc.Path, err)
}
metadata := toMetadata(entry.Extended)
if len(metadata) > 0 {
_, err = blobURL.SetMetadata(context.Background(), metadata, azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{})
if err != nil {
return nil, fmt.Errorf("azure set metadata on %s%s: %v", loc.Bucket, loc.Path, err)
}
}
// read back the remote entry
return az.readFileRemoteEntry(loc)
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) readFileRemoteEntry(loc *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
2021-08-24 08:18:30 +00:00
key := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
blobURL := containerURL.NewBlockBlobURL(key)
attr, err := blobURL.GetProperties(context.Background(), azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{})
if err != nil {
return nil, err
}
return &filer_pb.RemoteEntry{
RemoteMtime: attr.LastModified().Unix(),
RemoteSize: attr.ContentLength(),
RemoteETag: string(attr.ETag()),
StorageName: az.conf.Name,
}, nil
}
func toMetadata(attributes map[string][]byte) map[string]string {
metadata := make(map[string]string)
for k, v := range attributes {
metadata[k] = string(v)
}
return metadata
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) {
2021-08-24 08:18:30 +00:00
if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) {
return nil
}
metadata := toMetadata(newEntry.Extended)
key := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
_, err = containerURL.NewBlobURL(key).SetMetadata(context.Background(), metadata, azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{})
return
}
2021-08-26 22:18:34 +00:00
func (az *azureRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) {
2021-08-24 08:18:30 +00:00
key := loc.Path[1:]
containerURL := az.serviceURL.NewContainerURL(loc.Bucket)
if _, err = containerURL.NewBlobURL(key).Delete(context.Background(),
azblob.DeleteSnapshotsOptionInclude, azblob.BlobAccessConditions{}); err != nil {
return fmt.Errorf("azure delete %s%s: %v", loc.Bucket, loc.Path, err)
}
return
}