Konstantin Lebedev
2 months ago
38 changed files with 1379 additions and 564 deletions
-
15go.mod
-
45go.sum
-
5k8s/charts/seaweedfs/templates/s3/s3-deployment.yaml
-
7k8s/charts/seaweedfs/values.yaml
-
1test/s3/multipart/aws_upload.go
-
2unmaintained/diff_volume_servers/diff_volume_servers.go
-
5unmaintained/s3/presigned_put/presigned_put.go
-
2unmaintained/stream_read_volume/stream_read_volume.go
-
2unmaintained/stress_filer_upload/bench_filer_upload/bench_filer_upload.go
-
2unmaintained/stress_filer_upload/stress_filer_upload_actual/stress_filer_upload.go
-
4unmaintained/volume_tailer/volume_tailer.go
-
1weed/mq/broker/broker_grpc_pub.go
-
120weed/remote_storage/azure/azure_highlevel.go
-
283weed/remote_storage/azure/azure_storage_client.go
-
377weed/remote_storage/azure/azure_storage_client_test.go
-
92weed/replication/sink/azuresink/azure_sink.go
-
355weed/replication/sink/azuresink/azure_sink_test.go
-
2weed/s3api/policy_engine/types.go
-
6weed/s3api/s3api_object_handlers_put.go
-
56weed/s3api/s3api_object_handlers_put_test.go
-
4weed/s3api/s3err/s3-error.go
-
8weed/s3api/s3err/s3api_errors.go
-
5weed/server/filer_server_handlers_write.go
-
19weed/server/filer_server_handlers_write_autochunk.go
-
28weed/shell/command_volume_check_disk.go
-
7weed/util/constants/filer.go
@ -1,120 +0,0 @@ |
|||
package azure |
|||
|
|||
import ( |
|||
"context" |
|||
"crypto/rand" |
|||
"encoding/base64" |
|||
"errors" |
|||
"fmt" |
|||
"github.com/Azure/azure-pipeline-go/pipeline" |
|||
. "github.com/Azure/azure-storage-blob-go/azblob" |
|||
"io" |
|||
"sync" |
|||
) |
|||
|
|||
// copied from https://github.com/Azure/azure-storage-blob-go/blob/master/azblob/highlevel.go#L73:6
|
|||
// uploadReaderAtToBlockBlob was not public
|
|||
|
|||
// uploadReaderAtToBlockBlob uploads a buffer in blocks to a block blob.
|
|||
func uploadReaderAtToBlockBlob(ctx context.Context, reader io.ReaderAt, readerSize int64, |
|||
blockBlobURL BlockBlobURL, o UploadToBlockBlobOptions) (CommonResponse, error) { |
|||
if o.BlockSize == 0 { |
|||
// If bufferSize > (BlockBlobMaxStageBlockBytes * BlockBlobMaxBlocks), then error
|
|||
if readerSize > BlockBlobMaxStageBlockBytes*BlockBlobMaxBlocks { |
|||
return nil, errors.New("buffer is too large to upload to a block blob") |
|||
} |
|||
// If bufferSize <= BlockBlobMaxUploadBlobBytes, then Upload should be used with just 1 I/O request
|
|||
if readerSize <= BlockBlobMaxUploadBlobBytes { |
|||
o.BlockSize = BlockBlobMaxUploadBlobBytes // Default if unspecified
|
|||
} else { |
|||
o.BlockSize = readerSize / BlockBlobMaxBlocks // buffer / max blocks = block size to use all 50,000 blocks
|
|||
if o.BlockSize < BlobDefaultDownloadBlockSize { // If the block size is smaller than 4MB, round up to 4MB
|
|||
o.BlockSize = BlobDefaultDownloadBlockSize |
|||
} |
|||
// StageBlock will be called with blockSize blocks and a Parallelism of (BufferSize / BlockSize).
|
|||
} |
|||
} |
|||
|
|||
if readerSize <= BlockBlobMaxUploadBlobBytes { |
|||
// If the size can fit in 1 Upload call, do it this way
|
|||
var body io.ReadSeeker = io.NewSectionReader(reader, 0, readerSize) |
|||
if o.Progress != nil { |
|||
body = pipeline.NewRequestBodyProgress(body, o.Progress) |
|||
} |
|||
return blockBlobURL.Upload(ctx, body, o.BlobHTTPHeaders, o.Metadata, o.AccessConditions, o.BlobAccessTier, o.BlobTagsMap, o.ClientProvidedKeyOptions, o.ImmutabilityPolicyOptions) |
|||
} |
|||
|
|||
var numBlocks = uint16(((readerSize - 1) / o.BlockSize) + 1) |
|||
|
|||
blockIDList := make([]string, numBlocks) // Base-64 encoded block IDs
|
|||
progress := int64(0) |
|||
progressLock := &sync.Mutex{} |
|||
|
|||
err := DoBatchTransfer(ctx, BatchTransferOptions{ |
|||
OperationName: "uploadReaderAtToBlockBlob", |
|||
TransferSize: readerSize, |
|||
ChunkSize: o.BlockSize, |
|||
Parallelism: o.Parallelism, |
|||
Operation: func(offset int64, count int64, ctx context.Context) error { |
|||
// This function is called once per block.
|
|||
// It is passed this block's offset within the buffer and its count of bytes
|
|||
// Prepare to read the proper block/section of the buffer
|
|||
var body io.ReadSeeker = io.NewSectionReader(reader, offset, count) |
|||
blockNum := offset / o.BlockSize |
|||
if o.Progress != nil { |
|||
blockProgress := int64(0) |
|||
body = pipeline.NewRequestBodyProgress(body, |
|||
func(bytesTransferred int64) { |
|||
diff := bytesTransferred - blockProgress |
|||
blockProgress = bytesTransferred |
|||
progressLock.Lock() // 1 goroutine at a time gets a progress report
|
|||
progress += diff |
|||
o.Progress(progress) |
|||
progressLock.Unlock() |
|||
}) |
|||
} |
|||
|
|||
// Block IDs are unique values to avoid issue if 2+ clients are uploading blocks
|
|||
// at the same time causing PutBlockList to get a mix of blocks from all the clients.
|
|||
blockIDList[blockNum] = base64.StdEncoding.EncodeToString(newUUID().bytes()) |
|||
_, err := blockBlobURL.StageBlock(ctx, blockIDList[blockNum], body, o.AccessConditions.LeaseAccessConditions, nil, o.ClientProvidedKeyOptions) |
|||
return err |
|||
}, |
|||
}) |
|||
if err != nil { |
|||
return nil, err |
|||
} |
|||
// All put blocks were successful, call Put Block List to finalize the blob
|
|||
return blockBlobURL.CommitBlockList(ctx, blockIDList, o.BlobHTTPHeaders, o.Metadata, o.AccessConditions, o.BlobAccessTier, o.BlobTagsMap, o.ClientProvidedKeyOptions, o.ImmutabilityPolicyOptions) |
|||
} |
|||
|
|||
// The UUID reserved variants.
|
|||
const ( |
|||
reservedNCS byte = 0x80 |
|||
reservedRFC4122 byte = 0x40 |
|||
reservedMicrosoft byte = 0x20 |
|||
reservedFuture byte = 0x00 |
|||
) |
|||
|
|||
type uuid [16]byte |
|||
|
|||
// NewUUID returns a new uuid using RFC 4122 algorithm.
|
|||
func newUUID() (u uuid) { |
|||
u = uuid{} |
|||
// Set all bits to randomly (or pseudo-randomly) chosen values.
|
|||
rand.Read(u[:]) |
|||
u[8] = (u[8] | reservedRFC4122) & 0x7F // u.setVariant(ReservedRFC4122)
|
|||
|
|||
var version byte = 4 |
|||
u[6] = (u[6] & 0xF) | (version << 4) // u.setVersion(4)
|
|||
return |
|||
} |
|||
|
|||
// String returns an unparsed version of the generated UUID sequence.
|
|||
func (u uuid) String() string { |
|||
return fmt.Sprintf("%x-%x-%x-%x-%x", u[0:4], u[4:6], u[6:8], u[8:10], u[10:]) |
|||
} |
|||
|
|||
func (u uuid) bytes() []byte { |
|||
return u[:] |
|||
} |
|||
@ -0,0 +1,377 @@ |
|||
package azure |
|||
|
|||
import ( |
|||
"bytes" |
|||
"fmt" |
|||
"os" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" |
|||
) |
|||
|
|||
// TestAzureStorageClientBasic tests basic Azure storage client operations
|
|||
func TestAzureStorageClientBasic(t *testing.T) { |
|||
// Skip if credentials not available
|
|||
accountName := os.Getenv("AZURE_STORAGE_ACCOUNT") |
|||
accountKey := os.Getenv("AZURE_STORAGE_ACCESS_KEY") |
|||
testContainer := os.Getenv("AZURE_TEST_CONTAINER") |
|||
|
|||
if accountName == "" || accountKey == "" { |
|||
t.Skip("Skipping Azure storage test: AZURE_STORAGE_ACCOUNT or AZURE_STORAGE_ACCESS_KEY not set") |
|||
} |
|||
if testContainer == "" { |
|||
testContainer = "seaweedfs-test" |
|||
} |
|||
|
|||
// Create client
|
|||
maker := azureRemoteStorageMaker{} |
|||
conf := &remote_pb.RemoteConf{ |
|||
Name: "test-azure", |
|||
AzureAccountName: accountName, |
|||
AzureAccountKey: accountKey, |
|||
} |
|||
|
|||
client, err := maker.Make(conf) |
|||
if err != nil { |
|||
t.Fatalf("Failed to create Azure client: %v", err) |
|||
} |
|||
|
|||
azClient := client.(*azureRemoteStorageClient) |
|||
|
|||
// Test 1: Create bucket/container
|
|||
t.Run("CreateBucket", func(t *testing.T) { |
|||
err := azClient.CreateBucket(testContainer) |
|||
// Ignore error if bucket already exists
|
|||
if err != nil && !bloberror.HasCode(err, bloberror.ContainerAlreadyExists) { |
|||
t.Fatalf("Failed to create bucket: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test 2: List buckets
|
|||
t.Run("ListBuckets", func(t *testing.T) { |
|||
buckets, err := azClient.ListBuckets() |
|||
if err != nil { |
|||
t.Fatalf("Failed to list buckets: %v", err) |
|||
} |
|||
if len(buckets) == 0 { |
|||
t.Log("No buckets found (might be expected)") |
|||
} else { |
|||
t.Logf("Found %d buckets", len(buckets)) |
|||
} |
|||
}) |
|||
|
|||
// Test 3: Write file
|
|||
testContent := []byte("Hello from SeaweedFS Azure SDK migration test!") |
|||
testKey := fmt.Sprintf("/test-file-%d.txt", time.Now().Unix()) |
|||
loc := &remote_pb.RemoteStorageLocation{ |
|||
Name: "test-azure", |
|||
Bucket: testContainer, |
|||
Path: testKey, |
|||
} |
|||
|
|||
t.Run("WriteFile", func(t *testing.T) { |
|||
entry := &filer_pb.Entry{ |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Mtime: time.Now().Unix(), |
|||
Mime: "text/plain", |
|||
}, |
|||
Extended: map[string][]byte{ |
|||
"x-amz-meta-test-key": []byte("test-value"), |
|||
}, |
|||
} |
|||
|
|||
reader := bytes.NewReader(testContent) |
|||
remoteEntry, err := azClient.WriteFile(loc, entry, reader) |
|||
if err != nil { |
|||
t.Fatalf("Failed to write file: %v", err) |
|||
} |
|||
if remoteEntry == nil { |
|||
t.Fatal("Remote entry is nil") |
|||
} |
|||
if remoteEntry.RemoteSize != int64(len(testContent)) { |
|||
t.Errorf("Expected size %d, got %d", len(testContent), remoteEntry.RemoteSize) |
|||
} |
|||
}) |
|||
|
|||
// Test 4: Read file
|
|||
t.Run("ReadFile", func(t *testing.T) { |
|||
data, err := azClient.ReadFile(loc, 0, int64(len(testContent))) |
|||
if err != nil { |
|||
t.Fatalf("Failed to read file: %v", err) |
|||
} |
|||
if !bytes.Equal(data, testContent) { |
|||
t.Errorf("Content mismatch. Expected: %s, Got: %s", testContent, data) |
|||
} |
|||
}) |
|||
|
|||
// Test 5: Read partial file
|
|||
t.Run("ReadPartialFile", func(t *testing.T) { |
|||
data, err := azClient.ReadFile(loc, 0, 5) |
|||
if err != nil { |
|||
t.Fatalf("Failed to read partial file: %v", err) |
|||
} |
|||
expected := testContent[:5] |
|||
if !bytes.Equal(data, expected) { |
|||
t.Errorf("Content mismatch. Expected: %s, Got: %s", expected, data) |
|||
} |
|||
}) |
|||
|
|||
// Test 6: Update metadata
|
|||
t.Run("UpdateMetadata", func(t *testing.T) { |
|||
oldEntry := &filer_pb.Entry{ |
|||
Extended: map[string][]byte{ |
|||
"x-amz-meta-test-key": []byte("test-value"), |
|||
}, |
|||
} |
|||
newEntry := &filer_pb.Entry{ |
|||
Extended: map[string][]byte{ |
|||
"x-amz-meta-test-key": []byte("test-value"), |
|||
"x-amz-meta-new-key": []byte("new-value"), |
|||
}, |
|||
} |
|||
err := azClient.UpdateFileMetadata(loc, oldEntry, newEntry) |
|||
if err != nil { |
|||
t.Fatalf("Failed to update metadata: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test 7: Traverse (list objects)
|
|||
t.Run("Traverse", func(t *testing.T) { |
|||
foundFile := false |
|||
err := azClient.Traverse(loc, func(dir string, name string, isDir bool, remoteEntry *filer_pb.RemoteEntry) error { |
|||
if !isDir && name == testKey[1:] { // Remove leading slash
|
|||
foundFile = true |
|||
} |
|||
return nil |
|||
}) |
|||
if err != nil { |
|||
t.Fatalf("Failed to traverse: %v", err) |
|||
} |
|||
if !foundFile { |
|||
t.Log("Test file not found in traverse (might be expected due to path matching)") |
|||
} |
|||
}) |
|||
|
|||
// Test 8: Delete file
|
|||
t.Run("DeleteFile", func(t *testing.T) { |
|||
err := azClient.DeleteFile(loc) |
|||
if err != nil { |
|||
t.Fatalf("Failed to delete file: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test 9: Verify file deleted (should fail)
|
|||
t.Run("VerifyDeleted", func(t *testing.T) { |
|||
_, err := azClient.ReadFile(loc, 0, 10) |
|||
if !bloberror.HasCode(err, bloberror.BlobNotFound) { |
|||
t.Errorf("Expected BlobNotFound error, but got: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Clean up: Try to delete the test container
|
|||
// Comment out if you want to keep the container
|
|||
/* |
|||
t.Run("DeleteBucket", func(t *testing.T) { |
|||
err := azClient.DeleteBucket(testContainer) |
|||
if err != nil { |
|||
t.Logf("Warning: Failed to delete bucket: %v", err) |
|||
} |
|||
}) |
|||
*/ |
|||
} |
|||
|
|||
// TestToMetadata tests the metadata conversion function
|
|||
func TestToMetadata(t *testing.T) { |
|||
tests := []struct { |
|||
name string |
|||
input map[string][]byte |
|||
expected map[string]*string |
|||
}{ |
|||
{ |
|||
name: "basic metadata", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "key1": []byte("value1"), |
|||
s3_constants.AmzUserMetaPrefix + "key2": []byte("value2"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"key1": stringPtr("value1"), |
|||
"key2": stringPtr("value2"), |
|||
}, |
|||
}, |
|||
{ |
|||
name: "metadata with dashes", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "content-type": []byte("text/plain"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"content_2d_type": stringPtr("text/plain"), // dash (0x2d) -> _2d_
|
|||
}, |
|||
}, |
|||
{ |
|||
name: "non-metadata keys ignored", |
|||
input: map[string][]byte{ |
|||
"some-other-key": []byte("ignored"), |
|||
s3_constants.AmzUserMetaPrefix + "included": []byte("included"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"included": stringPtr("included"), |
|||
}, |
|||
}, |
|||
{ |
|||
name: "keys starting with digits", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "123key": []byte("value1"), |
|||
s3_constants.AmzUserMetaPrefix + "456-test": []byte("value2"), |
|||
s3_constants.AmzUserMetaPrefix + "789": []byte("value3"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"_123key": stringPtr("value1"), // starts with digit -> prefix _
|
|||
"_456_2d_test": stringPtr("value2"), // starts with digit AND has dash
|
|||
"_789": stringPtr("value3"), |
|||
}, |
|||
}, |
|||
{ |
|||
name: "uppercase and mixed case keys", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "My-Key": []byte("value1"), |
|||
s3_constants.AmzUserMetaPrefix + "UPPERCASE": []byte("value2"), |
|||
s3_constants.AmzUserMetaPrefix + "MiXeD-CaSe": []byte("value3"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"my_2d_key": stringPtr("value1"), // lowercase + dash -> _2d_
|
|||
"uppercase": stringPtr("value2"), |
|||
"mixed_2d_case": stringPtr("value3"), |
|||
}, |
|||
}, |
|||
{ |
|||
name: "keys with invalid characters", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "my.key": []byte("value1"), |
|||
s3_constants.AmzUserMetaPrefix + "key+plus": []byte("value2"), |
|||
s3_constants.AmzUserMetaPrefix + "key@symbol": []byte("value3"), |
|||
s3_constants.AmzUserMetaPrefix + "key-with.": []byte("value4"), |
|||
s3_constants.AmzUserMetaPrefix + "key/slash": []byte("value5"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"my_2e_key": stringPtr("value1"), // dot (0x2e) -> _2e_
|
|||
"key_2b_plus": stringPtr("value2"), // plus (0x2b) -> _2b_
|
|||
"key_40_symbol": stringPtr("value3"), // @ (0x40) -> _40_
|
|||
"key_2d_with_2e_": stringPtr("value4"), // dash and dot
|
|||
"key_2f_slash": stringPtr("value5"), // slash (0x2f) -> _2f_
|
|||
}, |
|||
}, |
|||
{ |
|||
name: "collision prevention", |
|||
input: map[string][]byte{ |
|||
s3_constants.AmzUserMetaPrefix + "my-key": []byte("value1"), |
|||
s3_constants.AmzUserMetaPrefix + "my.key": []byte("value2"), |
|||
s3_constants.AmzUserMetaPrefix + "my_key": []byte("value3"), |
|||
}, |
|||
expected: map[string]*string{ |
|||
"my_2d_key": stringPtr("value1"), // dash (0x2d)
|
|||
"my_2e_key": stringPtr("value2"), // dot (0x2e)
|
|||
"my_key": stringPtr("value3"), // underscore is valid, no encoding
|
|||
}, |
|||
}, |
|||
{ |
|||
name: "empty input", |
|||
input: map[string][]byte{}, |
|||
expected: map[string]*string{}, |
|||
}, |
|||
} |
|||
|
|||
for _, tt := range tests { |
|||
t.Run(tt.name, func(t *testing.T) { |
|||
result := toMetadata(tt.input) |
|||
if len(result) != len(tt.expected) { |
|||
t.Errorf("Expected %d keys, got %d", len(tt.expected), len(result)) |
|||
} |
|||
for key, expectedVal := range tt.expected { |
|||
if resultVal, ok := result[key]; !ok { |
|||
t.Errorf("Expected key %s not found", key) |
|||
} else if resultVal == nil || expectedVal == nil { |
|||
if resultVal != expectedVal { |
|||
t.Errorf("For key %s: expected %v, got %v", key, expectedVal, resultVal) |
|||
} |
|||
} else if *resultVal != *expectedVal { |
|||
t.Errorf("For key %s: expected %s, got %s", key, *expectedVal, *resultVal) |
|||
} |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
func contains(s, substr string) bool { |
|||
return bytes.Contains([]byte(s), []byte(substr)) |
|||
} |
|||
|
|||
func stringPtr(s string) *string { |
|||
return &s |
|||
} |
|||
|
|||
// Benchmark tests
|
|||
func BenchmarkToMetadata(b *testing.B) { |
|||
input := map[string][]byte{ |
|||
"x-amz-meta-key1": []byte("value1"), |
|||
"x-amz-meta-key2": []byte("value2"), |
|||
"x-amz-meta-content-type": []byte("text/plain"), |
|||
"other-key": []byte("ignored"), |
|||
} |
|||
|
|||
b.ResetTimer() |
|||
for i := 0; i < b.N; i++ { |
|||
toMetadata(input) |
|||
} |
|||
} |
|||
|
|||
// Test that the maker implements the interface
|
|||
func TestAzureRemoteStorageMaker(t *testing.T) { |
|||
maker := azureRemoteStorageMaker{} |
|||
|
|||
if !maker.HasBucket() { |
|||
t.Error("Expected HasBucket() to return true") |
|||
} |
|||
|
|||
// Test with missing credentials
|
|||
conf := &remote_pb.RemoteConf{ |
|||
Name: "test", |
|||
} |
|||
_, err := maker.Make(conf) |
|||
if err == nil { |
|||
t.Error("Expected error with missing credentials") |
|||
} |
|||
} |
|||
|
|||
// Test error cases
|
|||
func TestAzureStorageClientErrors(t *testing.T) { |
|||
// Test with invalid credentials
|
|||
maker := azureRemoteStorageMaker{} |
|||
conf := &remote_pb.RemoteConf{ |
|||
Name: "test", |
|||
AzureAccountName: "invalid", |
|||
AzureAccountKey: "aW52YWxpZGtleQ==", // base64 encoded "invalidkey"
|
|||
} |
|||
|
|||
client, err := maker.Make(conf) |
|||
if err != nil { |
|||
t.Skip("Invalid credentials correctly rejected at client creation") |
|||
} |
|||
|
|||
// If client creation succeeded, operations should fail
|
|||
azClient := client.(*azureRemoteStorageClient) |
|||
loc := &remote_pb.RemoteStorageLocation{ |
|||
Name: "test", |
|||
Bucket: "nonexistent", |
|||
Path: "/test.txt", |
|||
} |
|||
|
|||
// These operations should fail with invalid credentials
|
|||
_, err = azClient.ReadFile(loc, 0, 10) |
|||
if err == nil { |
|||
t.Log("Expected error with invalid credentials on ReadFile, but got none (might be cached)") |
|||
} |
|||
} |
|||
@ -0,0 +1,355 @@ |
|||
package azuresink |
|||
|
|||
import ( |
|||
"os" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
) |
|||
|
|||
// MockConfiguration for testing
|
|||
type mockConfiguration struct { |
|||
values map[string]interface{} |
|||
} |
|||
|
|||
func newMockConfiguration() *mockConfiguration { |
|||
return &mockConfiguration{ |
|||
values: make(map[string]interface{}), |
|||
} |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetString(key string) string { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.(string) |
|||
} |
|||
return "" |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetBool(key string) bool { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.(bool) |
|||
} |
|||
return false |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetInt(key string) int { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.(int) |
|||
} |
|||
return 0 |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetInt64(key string) int64 { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.(int64) |
|||
} |
|||
return 0 |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetFloat64(key string) float64 { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.(float64) |
|||
} |
|||
return 0.0 |
|||
} |
|||
|
|||
func (m *mockConfiguration) GetStringSlice(key string) []string { |
|||
if v, ok := m.values[key]; ok { |
|||
return v.([]string) |
|||
} |
|||
return nil |
|||
} |
|||
|
|||
func (m *mockConfiguration) SetDefault(key string, value interface{}) { |
|||
if _, exists := m.values[key]; !exists { |
|||
m.values[key] = value |
|||
} |
|||
} |
|||
|
|||
// Test the AzureSink interface implementation
|
|||
func TestAzureSinkInterface(t *testing.T) { |
|||
sink := &AzureSink{} |
|||
|
|||
if sink.GetName() != "azure" { |
|||
t.Errorf("Expected name 'azure', got '%s'", sink.GetName()) |
|||
} |
|||
|
|||
// Test directory setting
|
|||
sink.dir = "/test/dir" |
|||
if sink.GetSinkToDirectory() != "/test/dir" { |
|||
t.Errorf("Expected directory '/test/dir', got '%s'", sink.GetSinkToDirectory()) |
|||
} |
|||
|
|||
// Test incremental setting
|
|||
sink.isIncremental = true |
|||
if !sink.IsIncremental() { |
|||
t.Error("Expected isIncremental to be true") |
|||
} |
|||
} |
|||
|
|||
// Test Azure sink initialization
|
|||
func TestAzureSinkInitialization(t *testing.T) { |
|||
accountName := os.Getenv("AZURE_STORAGE_ACCOUNT") |
|||
accountKey := os.Getenv("AZURE_STORAGE_ACCESS_KEY") |
|||
testContainer := os.Getenv("AZURE_TEST_CONTAINER") |
|||
|
|||
if accountName == "" || accountKey == "" { |
|||
t.Skip("Skipping Azure sink test: AZURE_STORAGE_ACCOUNT or AZURE_STORAGE_ACCESS_KEY not set") |
|||
} |
|||
if testContainer == "" { |
|||
testContainer = "seaweedfs-test" |
|||
} |
|||
|
|||
sink := &AzureSink{} |
|||
|
|||
err := sink.initialize(accountName, accountKey, testContainer, "/test") |
|||
if err != nil { |
|||
t.Fatalf("Failed to initialize Azure sink: %v", err) |
|||
} |
|||
|
|||
if sink.container != testContainer { |
|||
t.Errorf("Expected container '%s', got '%s'", testContainer, sink.container) |
|||
} |
|||
|
|||
if sink.dir != "/test" { |
|||
t.Errorf("Expected dir '/test', got '%s'", sink.dir) |
|||
} |
|||
|
|||
if sink.client == nil { |
|||
t.Error("Expected client to be initialized") |
|||
} |
|||
} |
|||
|
|||
// Test configuration-based initialization
|
|||
func TestAzureSinkInitializeFromConfig(t *testing.T) { |
|||
accountName := os.Getenv("AZURE_STORAGE_ACCOUNT") |
|||
accountKey := os.Getenv("AZURE_STORAGE_ACCESS_KEY") |
|||
testContainer := os.Getenv("AZURE_TEST_CONTAINER") |
|||
|
|||
if accountName == "" || accountKey == "" { |
|||
t.Skip("Skipping Azure sink config test: AZURE_STORAGE_ACCOUNT or AZURE_STORAGE_ACCESS_KEY not set") |
|||
} |
|||
if testContainer == "" { |
|||
testContainer = "seaweedfs-test" |
|||
} |
|||
|
|||
config := newMockConfiguration() |
|||
config.values["azure.account_name"] = accountName |
|||
config.values["azure.account_key"] = accountKey |
|||
config.values["azure.container"] = testContainer |
|||
config.values["azure.directory"] = "/test" |
|||
config.values["azure.is_incremental"] = true |
|||
|
|||
sink := &AzureSink{} |
|||
err := sink.Initialize(config, "azure.") |
|||
if err != nil { |
|||
t.Fatalf("Failed to initialize from config: %v", err) |
|||
} |
|||
|
|||
if !sink.IsIncremental() { |
|||
t.Error("Expected incremental to be true") |
|||
} |
|||
} |
|||
|
|||
// Test cleanKey function
|
|||
func TestCleanKey(t *testing.T) { |
|||
tests := []struct { |
|||
input string |
|||
expected string |
|||
}{ |
|||
{"/test/file.txt", "test/file.txt"}, |
|||
{"test/file.txt", "test/file.txt"}, |
|||
{"/", ""}, |
|||
{"", ""}, |
|||
{"/a/b/c", "a/b/c"}, |
|||
} |
|||
|
|||
for _, tt := range tests { |
|||
t.Run(tt.input, func(t *testing.T) { |
|||
result := cleanKey(tt.input) |
|||
if result != tt.expected { |
|||
t.Errorf("cleanKey(%q) = %q, want %q", tt.input, result, tt.expected) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// Test entry operations (requires valid credentials)
|
|||
func TestAzureSinkEntryOperations(t *testing.T) { |
|||
accountName := os.Getenv("AZURE_STORAGE_ACCOUNT") |
|||
accountKey := os.Getenv("AZURE_STORAGE_ACCESS_KEY") |
|||
testContainer := os.Getenv("AZURE_TEST_CONTAINER") |
|||
|
|||
if accountName == "" || accountKey == "" { |
|||
t.Skip("Skipping Azure sink entry test: credentials not set") |
|||
} |
|||
if testContainer == "" { |
|||
testContainer = "seaweedfs-test" |
|||
} |
|||
|
|||
sink := &AzureSink{} |
|||
err := sink.initialize(accountName, accountKey, testContainer, "/test") |
|||
if err != nil { |
|||
t.Fatalf("Failed to initialize: %v", err) |
|||
} |
|||
|
|||
// Test CreateEntry with directory (should be no-op)
|
|||
t.Run("CreateDirectory", func(t *testing.T) { |
|||
entry := &filer_pb.Entry{ |
|||
IsDirectory: true, |
|||
} |
|||
err := sink.CreateEntry("/test/dir", entry, nil) |
|||
if err != nil { |
|||
t.Errorf("CreateEntry for directory should not error: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test CreateEntry with file
|
|||
testKey := "/test-sink-file-" + time.Now().Format("20060102-150405") + ".txt" |
|||
t.Run("CreateFile", func(t *testing.T) { |
|||
entry := &filer_pb.Entry{ |
|||
IsDirectory: false, |
|||
Content: []byte("Test content for Azure sink"), |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Mtime: time.Now().Unix(), |
|||
}, |
|||
} |
|||
err := sink.CreateEntry(testKey, entry, nil) |
|||
if err != nil { |
|||
t.Fatalf("Failed to create entry: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test UpdateEntry
|
|||
t.Run("UpdateEntry", func(t *testing.T) { |
|||
oldEntry := &filer_pb.Entry{ |
|||
Content: []byte("Old content"), |
|||
} |
|||
newEntry := &filer_pb.Entry{ |
|||
Content: []byte("New content for update test"), |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Mtime: time.Now().Unix(), |
|||
}, |
|||
} |
|||
found, err := sink.UpdateEntry(testKey, oldEntry, "/test", newEntry, false, nil) |
|||
if err != nil { |
|||
t.Fatalf("Failed to update entry: %v", err) |
|||
} |
|||
if !found { |
|||
t.Error("Expected found to be true") |
|||
} |
|||
}) |
|||
|
|||
// Test DeleteEntry
|
|||
t.Run("DeleteFile", func(t *testing.T) { |
|||
err := sink.DeleteEntry(testKey, false, false, nil) |
|||
if err != nil { |
|||
t.Fatalf("Failed to delete entry: %v", err) |
|||
} |
|||
}) |
|||
|
|||
// Test DeleteEntry with directory marker
|
|||
testDirKey := "/test-dir-" + time.Now().Format("20060102-150405") |
|||
t.Run("DeleteDirectory", func(t *testing.T) { |
|||
// First create a directory marker
|
|||
entry := &filer_pb.Entry{ |
|||
IsDirectory: false, |
|||
Content: []byte(""), |
|||
} |
|||
err := sink.CreateEntry(testDirKey+"/", entry, nil) |
|||
if err != nil { |
|||
t.Logf("Warning: Failed to create directory marker: %v", err) |
|||
} |
|||
|
|||
// Then delete it
|
|||
err = sink.DeleteEntry(testDirKey, true, false, nil) |
|||
if err != nil { |
|||
t.Logf("Warning: Failed to delete directory: %v", err) |
|||
} |
|||
}) |
|||
} |
|||
|
|||
// Test CreateEntry with precondition (IfUnmodifiedSince)
|
|||
func TestAzureSinkPrecondition(t *testing.T) { |
|||
accountName := os.Getenv("AZURE_STORAGE_ACCOUNT") |
|||
accountKey := os.Getenv("AZURE_STORAGE_ACCESS_KEY") |
|||
testContainer := os.Getenv("AZURE_TEST_CONTAINER") |
|||
|
|||
if accountName == "" || accountKey == "" { |
|||
t.Skip("Skipping Azure sink precondition test: credentials not set") |
|||
} |
|||
if testContainer == "" { |
|||
testContainer = "seaweedfs-test" |
|||
} |
|||
|
|||
sink := &AzureSink{} |
|||
err := sink.initialize(accountName, accountKey, testContainer, "/test") |
|||
if err != nil { |
|||
t.Fatalf("Failed to initialize: %v", err) |
|||
} |
|||
|
|||
testKey := "/test-precondition-" + time.Now().Format("20060102-150405") + ".txt" |
|||
|
|||
// Create initial entry
|
|||
entry := &filer_pb.Entry{ |
|||
Content: []byte("Initial content"), |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Mtime: time.Now().Unix(), |
|||
}, |
|||
} |
|||
err = sink.CreateEntry(testKey, entry, nil) |
|||
if err != nil { |
|||
t.Fatalf("Failed to create initial entry: %v", err) |
|||
} |
|||
|
|||
// Try to create again with old mtime (should be skipped due to precondition)
|
|||
oldEntry := &filer_pb.Entry{ |
|||
Content: []byte("Should not overwrite"), |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Mtime: time.Now().Add(-1 * time.Hour).Unix(), // Old timestamp
|
|||
}, |
|||
} |
|||
err = sink.CreateEntry(testKey, oldEntry, nil) |
|||
// Should either succeed (skip) or fail with precondition error
|
|||
if err != nil { |
|||
t.Logf("Create with old mtime: %v (expected)", err) |
|||
} |
|||
|
|||
// Clean up
|
|||
sink.DeleteEntry(testKey, false, false, nil) |
|||
} |
|||
|
|||
// Benchmark tests
|
|||
func BenchmarkCleanKey(b *testing.B) { |
|||
keys := []string{ |
|||
"/simple/path.txt", |
|||
"no/leading/slash.txt", |
|||
"/", |
|||
"/complex/path/with/many/segments/file.txt", |
|||
} |
|||
|
|||
b.ResetTimer() |
|||
for i := 0; i < b.N; i++ { |
|||
cleanKey(keys[i%len(keys)]) |
|||
} |
|||
} |
|||
|
|||
// Test error handling with invalid credentials
|
|||
func TestAzureSinkInvalidCredentials(t *testing.T) { |
|||
sink := &AzureSink{} |
|||
|
|||
err := sink.initialize("invalid-account", "aW52YWxpZGtleQ==", "test-container", "/test") |
|||
if err != nil { |
|||
t.Skip("Invalid credentials correctly rejected at initialization") |
|||
} |
|||
|
|||
// If initialization succeeded, operations should fail
|
|||
entry := &filer_pb.Entry{ |
|||
Content: []byte("test"), |
|||
} |
|||
err = sink.CreateEntry("/test.txt", entry, nil) |
|||
if err == nil { |
|||
t.Log("Expected error with invalid credentials, but got none (might be cached)") |
|||
} |
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
package s3api |
|||
|
|||
import ( |
|||
"testing" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err" |
|||
"github.com/seaweedfs/seaweedfs/weed/util/constants" |
|||
) |
|||
|
|||
func TestFilerErrorToS3Error(t *testing.T) { |
|||
tests := []struct { |
|||
name string |
|||
errString string |
|||
expectedErr s3err.ErrorCode |
|||
}{ |
|||
{ |
|||
name: "MD5 mismatch error", |
|||
errString: constants.ErrMsgBadDigest, |
|||
expectedErr: s3err.ErrBadDigest, |
|||
}, |
|||
{ |
|||
name: "Context canceled error", |
|||
errString: "rpc error: code = Canceled desc = context canceled", |
|||
expectedErr: s3err.ErrInvalidRequest, |
|||
}, |
|||
{ |
|||
name: "Context canceled error (simple)", |
|||
errString: "context canceled", |
|||
expectedErr: s3err.ErrInvalidRequest, |
|||
}, |
|||
{ |
|||
name: "Directory exists error", |
|||
errString: "existing /path/to/file is a directory", |
|||
expectedErr: s3err.ErrExistingObjectIsDirectory, |
|||
}, |
|||
{ |
|||
name: "File exists error", |
|||
errString: "/path/to/file is a file", |
|||
expectedErr: s3err.ErrExistingObjectIsFile, |
|||
}, |
|||
{ |
|||
name: "Unknown error", |
|||
errString: "some random error", |
|||
expectedErr: s3err.ErrInternalError, |
|||
}, |
|||
} |
|||
|
|||
for _, tt := range tests { |
|||
t.Run(tt.name, func(t *testing.T) { |
|||
result := filerErrorToS3Error(tt.errString) |
|||
if result != tt.expectedErr { |
|||
t.Errorf("filerErrorToS3Error(%q) = %v, want %v", tt.errString, result, tt.expectedErr) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
package constants |
|||
|
|||
// Filer error messages
|
|||
const ( |
|||
ErrMsgOperationNotPermitted = "operation not permitted" |
|||
ErrMsgBadDigest = "The Content-Md5 you specified did not match what we received." |
|||
) |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue