diff --git a/k8s/charts/seaweedfs/Chart.yaml b/k8s/charts/seaweedfs/Chart.yaml index 247b5bd06c..5248f746de 100644 --- a/k8s/charts/seaweedfs/Chart.yaml +++ b/k8s/charts/seaweedfs/Chart.yaml @@ -1,6 +1,6 @@ apiVersion: v1 description: SeaweedFS name: seaweedfs -appVersion: "4.49" +appVersion: "4.50" # Dev note: Trigger a helm chart release by `git tag -a helm-` -version: 4.49.0 +version: 4.50.0 diff --git a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml b/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml index 01210f6171..04321b49bb 100644 --- a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml +++ b/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml @@ -170,9 +170,6 @@ spec: {{- if .Values.filer.metricsIp }} -metricsIp={{ .Values.filer.metricsIp }} \ {{- end }} - {{- if .Values.filer.redirectOnRead }} - -redirectOnRead \ - {{- end }} {{- if .Values.filer.disableHttp }} -disableHttp \ {{- end }} diff --git a/k8s/charts/seaweedfs/values.yaml b/k8s/charts/seaweedfs/values.yaml index 6c6b8673db..2475590e53 100644 --- a/k8s/charts/seaweedfs/values.yaml +++ b/k8s/charts/seaweedfs/values.yaml @@ -740,9 +740,6 @@ filer: # encrypt data on volume servers encryptVolumeData: false - # Whether proxy or redirect to volume server during file GET request - redirectOnRead: false - # Limit sub dir listing size (default 100000) dirListLimit: 100000 diff --git a/note/star_history.svg b/note/star_history.svg index 064e5b443b..72170caf1c 100644 --- a/note/star_history.svg +++ b/note/star_history.svg @@ -6,7 +6,7 @@ - 2026-10-24T02:56:41.212930 + 2026-10-25T00:48:28.860698 image/svg+xml @@ -40,23 +40,23 @@ z - + - - + - + - + - + - + - + - + - + - + - + - + @@ -301,18 +301,18 @@ L 353.368045 26.215385 - + - + - + @@ -325,18 +325,18 @@ L 449.270683 26.215385 - + - + - + @@ -349,18 +349,18 @@ L 545.042127 26.215385 - + - + - + @@ -484,41 +484,41 @@ z - + - - + - + - + - + - + - + - + - + @@ -577,18 +577,18 @@ L 709.2 267.787451 - + - + - + @@ -599,18 +599,18 @@ L 709.2 223.857989 - + - + - + @@ -621,18 +621,18 @@ L 709.2 179.928527 - + - + - + @@ -643,18 +643,18 @@ L 709.2 135.999064 - + - + - + - + - + - + @@ -813,1009 +813,963 @@ z +" clip-path="url(#p5a5d3b85f6)" style="fill: none; stroke: #0969da; stroke-width: 1.6; stroke-linecap: square"/> + diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 1310793564..6826d4f3f1 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -3039,50 +3039,62 @@ pub async fn post_handler( let needle_etag = n.etag(); let needle_has_name = n.has_name(); - let write_result = write_upload_needle(&state, vid, n, fsync).await; - // Replicate to remote volume servers if this volume has replicas. // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1. + let needs_replication = !is_replicate && { + let store = state.store.read().unwrap(); + store + .find_volume(vid) + .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1) + }; + + // The fan-out runs concurrently with the local needle write, like Go's + // ReplicatedWrite: it reads the already-buffered request body, so the two + // sides share no mutable state. The ack still waits for both. + let replication = if needs_replication { + let state_clone = state.clone(); + let path_clone = path.clone(); + let query_clone = query.clone(); + let headers_clone = headers.clone(); + let body_clone = body.clone(); + Some(tokio::spawn(async move { + do_replicated_request( + &state_clone, + vid.0, + Method::POST, + &path_clone, + &query_clone, + &headers_clone, + Some(body_clone), + ) + .await + })) + } else { + None + }; + + let write_result = write_upload_needle(&state, vid, n, fsync).await; + // No leader-known gate: when the lookup cannot run, the write must fail - // rather than be acknowledged on fewer copies than required. - if !is_replicate && write_result.is_ok() { - let needs_replication = { - let store = state.store.read().unwrap(); - store - .find_volume(vid) - .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1) - }; - if needs_replication { - let state_clone = state.clone(); - let path_clone = path.clone(); - let query_clone = query.clone(); - let headers_clone = headers.clone(); - let body_clone = body.clone(); - let replication = tokio::spawn(async move { - do_replicated_request( - &state_clone, - vid.0, - Method::POST, - &path_clone, - &query_clone, - &headers_clone, - Some(body_clone), - ) - .await - }); - let replication_result = replication - .await + // rather than be acknowledged on fewer copies than required. The local + // write result takes precedence, matching Go's error ordering. + let replication_result = match replication { + Some(task) => Some( + task.await .map_err(|e| format!("replication task failed: {}", e)) - .flatten(); - if let Err(e) = replication_result { - tracing::error!("replicated write failed: {}", e); - return json_error_with_query( - StatusCode::INTERNAL_SERVER_ERROR, - format!("replication failed: {}", e), - Some(&query), - ); - } - } + .flatten(), + ), + None => None, + }; + if write_result.is_ok() + && let Some(Err(e)) = replication_result + { + tracing::error!("replicated write failed: {}", e); + return json_error_with_query( + StatusCode::INTERNAL_SERVER_ERROR, + format!("replication failed: {}", e), + Some(&query), + ); } let resp = match write_result { diff --git a/test/s3tables/catalog_trino/trino_catalog_test.go b/test/s3tables/catalog_trino/trino_catalog_test.go index 665bdbb6d2..8a50318069 100644 --- a/test/s3tables/catalog_trino/trino_catalog_test.go +++ b/test/s3tables/catalog_trino/trino_catalog_test.go @@ -496,9 +496,7 @@ func waitForTrino(t *testing.T, containerName string, timeout time.Duration) { return } else { lastOutput = output - outputStr := string(output) - if strings.Contains(outputStr, "No such container") || - strings.Contains(outputStr, "is not running") { + if trinoContainerExited(output) { containerDead = true break } @@ -507,18 +505,25 @@ func waitForTrino(t *testing.T, containerName string, timeout time.Duration) { } // If we can't connect to system catalog, try to at least connect to Trino server - cmd := exec.Command("docker", "exec", containerName, "trino", "--version") - if err := cmd.Run(); err == nil { + if output, err := exec.Command("docker", "exec", containerName, "trino", "--version").CombinedOutput(); err == nil { // Trino process is running, even if catalog isn't ready yet // Give it a bit more time time.Sleep(5 * time.Second) return + } else { + lastOutput = output + containerDead = containerDead || trinoContainerExited(output) } if containerDead { t.Fatalf("Trino container %s exited before becoming ready\nLast output:\n%s\n%s", containerName, string(lastOutput), trinoContainerDiagnostics(containerName)) } - t.Fatalf("Timed out waiting for Trino to be ready\nLast output:\n%s", string(lastOutput)) + t.Fatalf("Timed out waiting for Trino to be ready\nLast output:\n%s\n%s", string(lastOutput), trinoContainerDiagnostics(containerName)) +} + +func trinoContainerExited(output []byte) bool { + s := string(output) + return strings.Contains(s, "No such container") || strings.Contains(s, "is not running") } func trinoContainerDiagnostics(containerName string) string { diff --git a/test/volume_server/http/read_path_variants_test.go b/test/volume_server/http/read_path_variants_test.go index eac72079a9..a997080d36 100644 --- a/test/volume_server/http/read_path_variants_test.go +++ b/test/volume_server/http/read_path_variants_test.go @@ -170,11 +170,11 @@ func TestConditionalHeaderPrecedenceAndInvalidIfModifiedSince(t *testing.T) { precedenceReq.Header.Set("If-None-Match", "\"definitely-different-etag\"") precedenceResp := framework.DoRequest(t, client, precedenceReq) precedenceBody := framework.ReadAllAndClose(t, precedenceResp) - if precedenceResp.StatusCode != http.StatusNotModified { - t.Fatalf("conditional precedence expected 304, got %d", precedenceResp.StatusCode) + if precedenceResp.StatusCode != http.StatusOK { + t.Fatalf("mismatched If-None-Match overrides If-Modified-Since: expected 200, got %d", precedenceResp.StatusCode) } - if len(precedenceBody) != 0 { - t.Fatalf("conditional precedence expected empty body, got %d bytes", len(precedenceBody)) + if string(precedenceBody) != string(payload) { + t.Fatalf("mismatched If-None-Match body mismatch: got %q want %q", string(precedenceBody), string(payload)) } invalidIMSReq := mustNewRequest(t, http.MethodGet, clusterHarness.VolumeAdminURL()+"/"+fid) diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index 0b83bacc89..acbeb755a0 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/weed/filer/etcd/etcd_store.go @@ -181,7 +181,7 @@ func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_ func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index 897c70152a..1745d72df7 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -438,46 +438,39 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc) } - var notPrefixed []*Entry - lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) { - notPrefixed = append(notPrefixed, entry) - return true, nil - }) - if err != nil { - return - } - + // lastFileName must stay at the last name delivered to eachEntryFunc; the + // raw scan cursor runs ahead over non-matching names and cannot be returned. count := int64(0) - for count < limit && len(notPrefixed) > 0 { - for _, entry := range notPrefixed { - if strings.HasPrefix(entry.Name(), prefix) { - count++ - res, resErr := eachEntryFunc(entry) - - if resErr != nil { - err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr) - return - } - - if !res { - return - } - if count >= limit { - break - } - } + scanName, scanInclusive := startFileName, includeStartFile + for count < limit { + var page []*Entry + if scanName, err = actualStore.ListDirectoryEntries(ctx, dirPath, scanName, scanInclusive, limit, func(entry *Entry) (bool, error) { + page = append(page, entry) + return true, nil + }); err != nil { + return } - if count < limit && lastFileName < prefix { - notPrefixed = notPrefixed[:0] - lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) (bool, error) { - notPrefixed = append(notPrefixed, entry) - return true, nil - }) - if err != nil { + scanInclusive = false + if len(page) == 0 { + return + } + for _, entry := range page { + if !strings.HasPrefix(entry.Name(), prefix) { + continue + } + count++ + res, resErr := eachEntryFunc(entry) + if resErr != nil { + err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr) return } - } else { - break + lastFileName = entry.Name() + if !res || count >= limit { + return + } + } + if scanName >= prefix && !strings.HasPrefix(scanName, prefix) { + return } } return diff --git a/weed/filer/filerstore_wrapper_test.go b/weed/filer/filerstore_wrapper_test.go index 0b180f63e5..6351d6978f 100644 --- a/weed/filer/filerstore_wrapper_test.go +++ b/weed/filer/filerstore_wrapper_test.go @@ -205,3 +205,41 @@ func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) { assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx())) assert.NoError(t, wrapper.RollbackTransaction(expiredCtx())) } + +type noPrefixListStore struct { + *stubFilerStore +} + +func (s *noPrefixListStore) ListDirectoryPrefixedEntries(context.Context, util.FullPath, string, bool, int64, string, ListEachEntryFunc) (string, error) { + return "", ErrUnsupportedListDirectoryPrefixed +} + +// The wrapper fallback must return the last delivered name as the cursor, not +// the end of the raw page, and must keep paging while prefix matches can still +// follow the raw cursor. +func TestPrefixFilterEntriesCursorStaysAtLastDelivered(t *testing.T) { + store := &noPrefixListStore{newStubFilerStore()} + wrapper := NewFilerStoreWrapper(store) + ctx := context.Background() + for _, name := range []string{"a", "b1", "b2", "b3", "b4"} { + require.NoError(t, store.InsertEntry(ctx, &Entry{FullPath: util.FullPath("/d/" + name)})) + } + + var delivered []string + lastFileName, err := wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "", false, 2, "b", func(entry *Entry) (bool, error) { + delivered = append(delivered, entry.Name()) + return true, nil + }) + require.NoError(t, err) + assert.Equal(t, []string{"b1", "b2"}, delivered) + assert.Equal(t, "b2", lastFileName) + + delivered = delivered[:0] + lastFileName, err = wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), lastFileName, false, 10, "b", func(entry *Entry) (bool, error) { + delivered = append(delivered, entry.Name()) + return true, nil + }) + require.NoError(t, err) + assert.Equal(t, []string{"b3", "b4"}, delivered) + assert.Equal(t, "b4", lastFileName) +} diff --git a/weed/filer/hbase/hbase_store.go b/weed/filer/hbase/hbase_store.go index 7663fef9d3..cfa027f3c1 100644 --- a/weed/filer/hbase/hbase_store.go +++ b/weed/filer/hbase/hbase_store.go @@ -156,7 +156,11 @@ func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util. func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}} expectedPrefix := []byte(dirPath.Child(prefix)) - scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family)) + scanStart := expectedPrefix + if resumeKey := []byte(dirPath.Child(startFileName)); bytes.Compare(resumeKey, scanStart) > 0 { + scanStart = resumeKey + } + scan, err := hrpc.NewScanRange(ctx, store.table, scanStart, nil, hrpc.Families(family)) if err != nil { return lastFileName, err } @@ -188,7 +192,7 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa value := cell.Value - if fileName == startFileName && !includeStartFile { + if fileName < startFileName || (fileName == startFileName && !includeStartFile) { continue } diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 606166da90..341327e433 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/weed/filer/leveldb/leveldb_store.go @@ -221,7 +221,7 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/leveldb/leveldb_store_test.go b/weed/filer/leveldb/leveldb_store_test.go index b5384fc778..b55586ec98 100644 --- a/weed/filer/leveldb/leveldb_store_test.go +++ b/weed/filer/leveldb/leveldb_store_test.go @@ -110,3 +110,28 @@ func BenchmarkInsertEntry(b *testing.B) { store.InsertEntry(ctx, entry) } } + +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDBStore{} + store.initialize(t.TempDir()) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 5c05033b99..e861db47e1 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/weed/filer/leveldb2/leveldb2_store.go @@ -239,7 +239,7 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart, _ = genDirectoryKeyPrefix(dirPath, startFileName, store.dbCount) } diff --git a/weed/filer/leveldb2/leveldb2_store_test.go b/weed/filer/leveldb2/leveldb2_store_test.go index 9651711d92..8d1d46ccc3 100644 --- a/weed/filer/leveldb2/leveldb2_store_test.go +++ b/weed/filer/leveldb2/leveldb2_store_test.go @@ -151,6 +151,31 @@ func TestInitializeRejectsOversizedOption(t *testing.T) { } } +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDB2Store{} + store.initialize(t.TempDir(), 2) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} + func TestEmptyRoot(t *testing.T) { testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) dir := t.TempDir() diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 029d902715..0b8748347a 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/weed/filer/leveldb3/leveldb3_store.go @@ -370,7 +370,7 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(shortPath, startFileName) } diff --git a/weed/filer/leveldb3/leveldb3_store_test.go b/weed/filer/leveldb3/leveldb3_store_test.go index 62896a299f..a9f3be10da 100644 --- a/weed/filer/leveldb3/leveldb3_store_test.go +++ b/weed/filer/leveldb3/leveldb3_store_test.go @@ -182,3 +182,28 @@ func TestEmptyRoot(t *testing.T) { } } + +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDB3Store{} + store.initialize(t.TempDir()) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go index 1320495583..2f90230e33 100644 --- a/weed/filer/redis2/universal_redis_store.go +++ b/weed/filer/redis2/universal_redis_store.go @@ -15,6 +15,10 @@ import ( const ( DIR_LIST_MARKER = "\x00" + + // a listing whose callback stops early reads at most one batch more than + // it uses, and no single pipeline grows with the listing limit + listFetchBatchSize = 128 ) type UniversalRedis2Store struct { @@ -96,7 +100,12 @@ func (store *UniversalRedis2Store) UpdateEntry(ctx context.Context, entry *filer func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) { - data, err := store.Client.Get(ctx, store.getKey(string(fullpath))).Result() + return decodeEntry(fullpath, store.Client.Get(ctx, store.getKey(string(fullpath)))) +} + +func decodeEntry(fullpath util.FullPath, get *redis.StringCmd) (entry *filer.Entry, err error) { + + data, err := get.Result() if err == redis.Nil { return nil, filer_pb.ErrNotFound } @@ -202,9 +211,13 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir // fetch entry meta var entry *filer.Entry - for _, fileName := range members { + var gets []*redis.StringCmd + for i, fileName := range members { + if i%listFetchBatchSize == 0 { + gets = store.getEntries(ctx, dirPath, members[i:]) + } path := util.NewFullPath(string(dirPath), fileName) - entry, err = store.FindEntry(ctx, path) + entry, err = decodeEntry(path, gets[i%listFetchBatchSize]) lastFileName = fileName if err != nil { glog.V(0).InfofCtx(ctx, "list %s : %v", path, err) @@ -235,6 +248,23 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir return lastFileName, err } +// getEntries pipelines the GETs for up to listFetchBatchSize of members. Each +// command keeps its own error, so a missing value still reads as redis.Nil and +// fails only that member. +func (store *UniversalRedis2Store) getEntries(ctx context.Context, dirPath util.FullPath, members []string) []*redis.StringCmd { + if len(members) > listFetchBatchSize { + members = members[:listFetchBatchSize] + } + gets := make([]*redis.StringCmd, len(members)) + store.Client.Pipelined(ctx, func(pipe redis.Pipeliner) error { + for i, fileName := range members { + gets[i] = pipe.Get(ctx, store.getKey(string(util.NewFullPath(string(dirPath), fileName)))) + } + return nil + }) + return gets +} + func (store *UniversalRedis2Store) removeOrphanedDirectoryListMember(ctx context.Context, dirPath util.FullPath, fileName string) { // a directory converted to super large after accumulating members still has a legacy index if store.isSuperLargeDirectory(string(dirPath)) { diff --git a/weed/filer/redis2/universal_redis_store_test.go b/weed/filer/redis2/universal_redis_store_test.go index 1558538dad..0bce68201c 100644 --- a/weed/filer/redis2/universal_redis_store_test.go +++ b/weed/filer/redis2/universal_redis_store_test.go @@ -256,6 +256,95 @@ func TestListDirectoryEntriesDeletesLogicallyExpiredEntries(t *testing.T) { } } +func TestListDirectoryEntriesSkipsMissingAndExpiredMembersInOrder(t *testing.T) { + for _, keyPrefix := range []string{"", "sw:"} { + t.Run("keyPrefix="+keyPrefix, func(t *testing.T) { + store, dir := newTestStore(t, keyPrefix) + + insertTestEntry(t, store, dir.Child("a"), 0) + insertTestEntry(t, store, dir.Child("b"), 0) + insertLogicallyExpiredTestEntry(t, store, dir.Child("c")) + insertTestEntry(t, store, dir.Child("d"), 0) + + if err := store.Client.Del(context.Background(), store.getKey(string(dir.Child("b")))).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + + if names := listNames(t, store, dir); fmt.Sprint(names) != "[a d]" { + t.Fatalf("listed %v, want [a d]", names) + } + + if members := indexMembers(t, store, dir); fmt.Sprint(members) != "[a d]" { + t.Fatalf("directory index holds %v, want [a d]", members) + } + }) + } +} + +func TestListDirectoryEntriesAcrossFetchBatches(t *testing.T) { + store, dir := newTestStore(t, "") + + var want []string + for i := 0; i < listFetchBatchSize+2; i++ { + name := fmt.Sprintf("f%04d", i) + insertTestEntry(t, store, dir.Child(name), 0) + if i != listFetchBatchSize { + want = append(want, name) + } + } + + // the first member of the second batch + missing := dir.Child(fmt.Sprintf("f%04d", listFetchBatchSize)) + if err := store.Client.Del(context.Background(), store.getKey(string(missing))).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + + var names []string + if _, err := store.ListDirectoryEntries(context.Background(), dir, "", true, 1000, func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }); err != nil { + t.Fatalf("ListDirectoryEntries %s: %v", dir, err) + } + if fmt.Sprint(names) != fmt.Sprint(want) { + t.Fatalf("listed %v, want %v", names, want) + } + + if members := indexMembers(t, store, dir); fmt.Sprint(members) != fmt.Sprint(want) { + t.Fatalf("directory index holds %v, want %v", members, want) + } +} + +func TestListDirectoryEntriesStopsAtUnreadableMember(t *testing.T) { + store, dir := newTestStore(t, "") + ctx := context.Background() + + insertTestEntry(t, store, dir.Child("a"), 0) + insertTestEntry(t, store, dir.Child("b"), 0) + insertTestEntry(t, store, dir.Child("c"), 0) + + // a value of the wrong type makes its GET fail with something other than redis.Nil + key := store.getKey(string(dir.Child("b"))) + if err := store.Client.Del(ctx, key).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + if err := store.Client.HSet(ctx, key, "f", "v").Err(); err != nil { + t.Fatalf("plant hash: %v", err) + } + + var names []string + lastFileName, err := store.ListDirectoryEntries(ctx, dir, "", true, 100, func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err == nil { + t.Fatal("ListDirectoryEntries returned no error for an unreadable member") + } + if fmt.Sprint(names) != "[a]" || lastFileName != "b" { + t.Fatalf("listed %v up to %q, want [a] up to \"b\"", names, lastFileName) + } +} + func TestDeleteExpiredEntryKeepsRecreatedValue(t *testing.T) { store, dir := newTestStore(t, "") diff --git a/weed/filer/rocksdb/rocksdb_store.go b/weed/filer/rocksdb/rocksdb_store.go index 2283efa6fb..dd88d73aa3 100644 --- a/weed/filer/rocksdb/rocksdb_store.go +++ b/weed/filer/rocksdb/rocksdb_store.go @@ -243,7 +243,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/tikv/tikv_store.go b/weed/filer/tikv/tikv_store.go index d6ba8f2844..92c34d70d6 100644 --- a/weed/filer/tikv/tikv_store.go +++ b/weed/filer/tikv/tikv_store.go @@ -246,7 +246,7 @@ func (store *TikvStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat lastFileName := "" directoryPrefix := store.genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = store.genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/s3api/s3api_object_handlers_postpolicy.go b/weed/s3api/s3api_object_handlers_postpolicy.go index 1e8705ba11..8ebcc71097 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy.go +++ b/weed/s3api/s3api_object_handlers_postpolicy.go @@ -139,7 +139,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R // Forward validated POST form fields to the underlying PUT as headers. applyPostPolicyFormHeaders(r, formValues) - // Authorize the object like the PUT path; the coarse Write check above does not consult the bucket policy. + // Authorize the object like the PUT path. if errCode := s3a.iam.AuthorizeObjectWrite(r, identity, bucket, object); errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, errCode)) return @@ -382,25 +382,15 @@ func getRedirectPostRawQuery(bucket, key, etag string) string { } // Check to see if Policy is signed correctly, returning the signing identity. -// The verified form's access key is recorded in ctx's identity holder. The -// identity's Write permission on the bucket is checked only once the signature -// verifies, as on the header-signed path, so a form that fails verification -// learns nothing about the key's permissions. A verified identity that may not -// write is returned with ErrAccessDenied. +// The verified form's access key is recorded in ctx's identity holder. +// Write authorization happens later in the handler via AuthorizeObjectWrite, +// which evaluates the identity's attached IAM policies, session policies, and +// the bucket policy against the real object key; gating here on the legacy +// Actions projection would deny identities those mechanisms allow. func (iam *IdentityAccessManagement) doesPolicySignatureMatch(ctx context.Context, formValues http.Header) (*Identity, s3err.ErrorCode) { - var identity *Identity - var errCode s3err.ErrorCode // For SignV2 - Signature field will be valid if _, ok := formValues["Signature"]; ok { - identity, errCode = iam.doesPolicySignatureV2Match(ctx, formValues) - } else { - identity, errCode = iam.doesPolicySignatureV4Match(ctx, formValues) + return iam.doesPolicySignatureV2Match(ctx, formValues) } - if errCode != s3err.ErrNone { - return nil, errCode - } - if !identity.CanDo(s3_constants.ACTION_WRITE, formValues.Get("bucket"), "") { - return identity, s3err.ErrAccessDenied - } - return identity, s3err.ErrNone + return iam.doesPolicySignatureV4Match(ctx, formValues) } diff --git a/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go b/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go index 553ff2f0a5..a0e0f7e4ce 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go +++ b/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go @@ -32,6 +32,11 @@ const ( postAclAccessKey = "AKIAACLTESTTEST" postAclSecretKey = "secret-key-for-acl-tests" postAclPrincipal = "arn:aws:iam::000000000000:user/aclwriter" + + postPolicyAccessKey = "AKIAPOLICYONLY0000" + postPolicySecretKey = "secret-key-for-policy-only" + postReadAccessKey = "AKIAPOLICYREAD0000" + postReadSecretKey = "secret-key-for-policy-read" ) // newPostAuthzServer builds an S3ApiServer whose non-admin "tester" identity @@ -274,3 +279,118 @@ func TestAuthorizeObjectAclWrite(t *testing.T) { s3a.iam.AuthorizeObjectAclWrite(req, admin, postAuthzBucket, "acl-denied/secret.txt"), "an admin is not confined by the bucket policy") } + +// newPostPolicyOnlyServer builds an S3ApiServer with identities that hold no +// legacy Write action: "poster" carries an attached IAM policy allowing +// s3:PutObject on the test bucket, and "reader" one allowing only s3:GetObject. +func newPostPolicyOnlyServer(t *testing.T) *S3ApiServer { + t.Helper() + + iam := &IdentityAccessManagement{ + hashes: make(map[string]*sync.Pool), + hashCounters: make(map[string]*int32), + } + err := iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{ + Name: "poster", + Account: &iam_pb.Account{Id: "000000000000", DisplayName: "poster"}, + Credentials: []*iam_pb.Credential{{AccessKey: postPolicyAccessKey, SecretKey: postPolicySecretKey}}, + Actions: []string{"List:" + postAuthzBucket}, + PolicyNames: []string{"post-write"}, + }, { + Name: "reader", + Account: &iam_pb.Account{Id: "000000000000", DisplayName: "reader"}, + Credentials: []*iam_pb.Credential{{AccessKey: postReadAccessKey, SecretKey: postReadSecretKey}}, + Actions: []string{"Read:" + postAuthzBucket, "List:" + postAuthzBucket}, + PolicyNames: []string{"post-read"}, + }}, + Policies: []*iam_pb.Policy{{ + Name: "post-write", + Content: fmt.Sprintf(`{ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"], + "Resource": ["arn:aws:s3:::%s/*"] + }, { + "Effect": "Allow", + "Action": ["s3:ListBucket", "s3:GetBucketLocation"], + "Resource": ["arn:aws:s3:::%s"] + }] + }`, postAuthzBucket, postAuthzBucket), + }, { + Name: "post-read", + Content: fmt.Sprintf(`{ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": ["s3:GetObject"], + "Resource": ["arn:aws:s3:::%s/*"] + }] + }`, postAuthzBucket), + }}, + }) + require.NoError(t, err) + + s3a := &S3ApiServer{ + option: &S3ApiServerOption{BucketsPath: "/buckets"}, + iam: iam, + } + s3a.bucketRegistry = NewBucketRegistry(s3a) + s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: postAuthzBucket, IsTableBucket: false}) + return s3a +} + +// postFormValues canonicalizes the signed request's multipart fields the way +// extractPostPolicyFormValues does, for driving doesPolicySignatureMatch. +func postFormValues(t *testing.T, r *http.Request) http.Header { + t.Helper() + require.NoError(t, r.ParseMultipartForm(1<<20)) + formValues := make(http.Header) + for k, v := range r.MultipartForm.Value { + formValues[http.CanonicalHeaderKey(k)] = v + } + return formValues +} + +// TestPostPolicySignatureAllowsAttachedPolicyIdentity signs a POST Object form +// with a key whose holder has no legacy Write action - only an attached IAM +// policy granting s3:PutObject - and asserts the signature check accepts it +// and the shared object-write authorization allows the key. Gating the +// signature check on the Actions projection used to fail this with +// AccessDenied before authorization ever ran. +func TestPostPolicySignatureAllowsAttachedPolicyIdentity(t *testing.T) { + s3a := newPostPolicyOnlyServer(t) + + req := newSignedPostRequestFor(t, "uploads/pic.png", postPolicyAccessKey, postPolicySecretKey, nil) + identity, errCode := s3a.iam.doesPolicySignatureMatch(req.Context(), postFormValues(t, req)) + require.Equal(t, s3err.ErrNone, errCode, + "a verified POST signature must not require the legacy Write action") + require.NotNil(t, identity) + + assert.Equal(t, s3err.ErrNone, + s3a.iam.AuthorizeObjectWrite(req, identity, postAuthzBucket, "uploads/pic.png"), + "an attached IAM policy granting s3:PutObject must authorize the key like the PUT path") +} + +// TestPostPolicyAttachedPolicyDeny covers the deny leg: a verified identity +// whose attached policy does not grant s3:PutObject is still refused by +// AuthorizeObjectWrite and by the full handler. +func TestPostPolicyAttachedPolicyDeny(t *testing.T) { + s3a := newPostPolicyOnlyServer(t) + + req := newSignedPostRequestFor(t, "uploads/pic.png", postReadAccessKey, postReadSecretKey, nil) + identity, errCode := s3a.iam.doesPolicySignatureMatch(req.Context(), postFormValues(t, req)) + require.Equal(t, s3err.ErrNone, errCode) + require.NotNil(t, identity) + + assert.Equal(t, s3err.ErrAccessDenied, + s3a.iam.AuthorizeObjectWrite(req, identity, postAuthzBucket, "uploads/pic.png"), + "an attached policy without s3:PutObject must still deny the upload") + + rec := httptest.NewRecorder() + s3a.PostPolicyBucketHandler(rec, newSignedPostRequestFor(t, "uploads/pic.png", postReadAccessKey, postReadSecretKey, nil)) + assert.Equal(t, http.StatusForbidden, rec.Code, + "POST from a read-only identity must stay AccessDenied, body: %s", rec.Body.String()) + assert.Contains(t, rec.Body.String(), "AccessDenied") +} diff --git a/weed/shell/command_volume_configure_replication.go b/weed/shell/command_volume_configure_replication.go index 8add4aad80..cc85bb5f34 100644 --- a/weed/shell/command_volume_configure_replication.go +++ b/weed/shell/command_volume_configure_replication.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "io" + "sync" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -33,6 +34,9 @@ func (c *commandVolumeConfigureReplication) Help() string { This command changes a volume replication value. It should be followed by "volume.fix.replication". + Up to -maxParallelization volume servers are configured at a time, one volume at a time on each, + and replicas of the same volume on different servers take turns. + ` } @@ -46,6 +50,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id") replicationString := configureReplicationCommand.String("replication", "", "the intended replication value") collectionPattern := configureReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all") + maxParallelization := configureReplicationCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") if err = configureReplicationCommand.Parse(args); err != nil { return nil } @@ -75,6 +80,19 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman return err } + return configureReplicationOnVolumeServers(commandEnv, topologyInfo, volumeFilter, replicaPlacement.String(), *maxParallelization) +} + +func configureReplicationOnVolumeServers(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeFilter func(*master_pb.VolumeInformationMessage) bool, replication string, maxParallelization int) error { + type serverVolumes struct { + addr pb.ServerAddress + volumeIds []uint32 + } + var servers []serverVolumes + // VolumeConfigure unmounts and remounts the volume, so replicas take turns + // to keep at least one of them serving. + volumeLocks := make(map[uint32]*sync.Mutex) + // find all data nodes with volumes that needs replication change eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { var targetVolumeIds []uint32 @@ -82,33 +100,45 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman for _, v := range diskInfo.VolumeInfos { if volumeFilter(v) { targetVolumeIds = append(targetVolumeIds, v.Id) + if volumeLocks[v.Id] == nil { + volumeLocks[v.Id] = &sync.Mutex{} + } } } } - if len(targetVolumeIds) == 0 { - return - } - err = operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dn), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - for _, targetVolumeId := range targetVolumeIds { - resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ - VolumeId: targetVolumeId, - Replication: replicaPlacement.String(), - }) - if configureErr != nil { - return configureErr - } - if resp.Error != "" { - return errors.New(resp.Error) - } - } - return nil - }) - if err != nil { - return + if len(targetVolumeIds) > 0 { + servers = append(servers, serverVolumes{pb.NewServerAddressFromDataNode(dn), targetVolumeIds}) } }) - return err + ewg := NewErrorWaitGroup(maxParallelization) + for _, server := range servers { + ewg.Add(func() error { + err := operation.WithVolumeServerClient(false, server.addr, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { + for _, targetVolumeId := range server.volumeIds { + volumeLocks[targetVolumeId].Lock() + resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ + VolumeId: targetVolumeId, + Replication: replication, + }) + volumeLocks[targetVolumeId].Unlock() + if configureErr != nil { + return configureErr + } + if resp.Error != "" { + return errors.New(resp.Error) + } + } + return nil + }) + if err != nil { + return fmt.Errorf("volume server %s: %w", server.addr, err) + } + return nil + }) + } + + return ewg.Wait() } func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) (func(message *master_pb.VolumeInformationMessage) bool, error) { diff --git a/weed/shell/command_volume_configure_replication_test.go b/weed/shell/command_volume_configure_replication_test.go new file mode 100644 index 0000000000..f24e65b835 --- /dev/null +++ b/weed/shell/command_volume_configure_replication_test.go @@ -0,0 +1,226 @@ +package shell + +import ( + "context" + "fmt" + "maps" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +type configureReplicationTestVolumeServer struct { + volume_server_pb.UnimplementedVolumeServerServer + failure string + mu sync.Mutex + configured map[uint32]string +} + +func (s *configureReplicationTestVolumeServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.configured == nil { + s.configured = make(map[uint32]string) + } + s.configured[req.VolumeId] = req.Replication + return &volume_server_pb.VolumeConfigureResponse{Error: s.failure}, nil +} + +// configureReplicationOverlapServer can back several volume servers at once and +// records whether two calls for the same volume were ever in flight together. +type configureReplicationOverlapServer struct { + volume_server_pb.UnimplementedVolumeServerServer + mu sync.Mutex + inFlight map[uint32]int + overlapped bool +} + +func (s *configureReplicationOverlapServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.mu.Lock() + s.inFlight[req.VolumeId]++ + if s.inFlight[req.VolumeId] > 1 { + s.overlapped = true + } + s.mu.Unlock() + + time.Sleep(20 * time.Millisecond) + + s.mu.Lock() + s.inFlight[req.VolumeId]-- + s.mu.Unlock() + return &volume_server_pb.VolumeConfigureResponse{}, nil +} + +func startConfigureReplicationTestVolumeServer(t *testing.T, server volume_server_pb.VolumeServerServer, volumeIds ...uint32) *master_pb.DataNodeInfo { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, server) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + port := lis.Addr().(*net.TCPAddr).Port + disk := &master_pb.DiskInfo{} + for _, vid := range volumeIds { + disk.VolumeInfos = append(disk.VolumeInfos, &master_pb.VolumeInformationMessage{Id: vid}) + } + return &master_pb.DataNodeInfo{ + Id: fmt.Sprintf("127.0.0.1:%d", port), + GrpcPort: uint32(port), + DiskInfos: map[string]*master_pb.DiskInfo{"": disk}, + } +} + +func configureReplicationTestEnv(dataNodes ...*master_pb.DataNodeInfo) (*CommandEnv, *master_pb.TopologyInfo) { + commandEnv := &CommandEnv{option: &ShellOptions{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }} + topologyInfo := &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{ + RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dataNodes}}, + }}} + return commandEnv, topologyInfo +} + +func allVolumes(*master_pb.VolumeInformationMessage) bool { return true } + +func TestConfigureReplicationOnVolumeServersReportsEveryFailingServer(t *testing.T) { + for _, maxParallelization := range []int{1, DefaultMaxParallelization} { + t.Run(fmt.Sprintf("maxParallelization=%d", maxParallelization), func(t *testing.T) { + first := &configureReplicationTestVolumeServer{failure: "first server failed"} + healthy := &configureReplicationTestVolumeServer{} + last := &configureReplicationTestVolumeServer{failure: "last server failed"} + commandEnv, topologyInfo := configureReplicationTestEnv( + startConfigureReplicationTestVolumeServer(t, first, 1), + startConfigureReplicationTestVolumeServer(t, healthy, 2, 3), + startConfigureReplicationTestVolumeServer(t, last, 4), + ) + + err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization) + + if err == nil { + t.Fatal("expected an error") + } + for _, failure := range []string{first.failure, last.failure} { + if !strings.Contains(err.Error(), failure) { + t.Errorf("error %q does not mention %q", err, failure) + } + } + healthy.mu.Lock() + defer healthy.mu.Unlock() + if expected := map[uint32]string{2: "001", 3: "001"}; !maps.Equal(healthy.configured, expected) { + t.Errorf("healthy server configured %v, expected %v", healthy.configured, expected) + } + }) + } +} + +func TestConfigureReplicationOnVolumeServersConfiguresReplicasOneAtATime(t *testing.T) { + server := &configureReplicationOverlapServer{inFlight: make(map[uint32]int)} + commandEnv, topologyInfo := configureReplicationTestEnv( + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + ) + + if err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "002", DefaultMaxParallelization); err != nil { + t.Fatal(err) + } + + server.mu.Lock() + defer server.mu.Unlock() + if server.overlapped { + t.Error("replicas of one volume were configured at the same time") + } +} + +// configureReplicationGateServer blocks inside VolumeConfigure until the test +// releases it, so the test can observe how many servers run at once. +type configureReplicationGateServer struct { + volume_server_pb.UnimplementedVolumeServerServer + gate *configureReplicationGate +} + +type configureReplicationGate struct { + entered chan struct{} + release chan struct{} + mu sync.Mutex + active int + maxActive int + calls int +} + +func (s *configureReplicationGateServer) VolumeConfigure(_ context.Context, _ *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.gate.mu.Lock() + s.gate.calls++ + s.gate.active++ + if s.gate.active > s.gate.maxActive { + s.gate.maxActive = s.gate.active + } + s.gate.mu.Unlock() + + s.gate.entered <- struct{}{} + <-s.gate.release + + s.gate.mu.Lock() + s.gate.active-- + s.gate.mu.Unlock() + return &volume_server_pb.VolumeConfigureResponse{}, nil +} + +func TestConfigureReplicationOnVolumeServersRunsServersInParallelUpToTheLimit(t *testing.T) { + const servers = 4 + const maxParallelization = 2 + gate := &configureReplicationGate{ + entered: make(chan struct{}, servers), + release: make(chan struct{}), + } + var dataNodes []*master_pb.DataNodeInfo + for i := 0; i < servers; i++ { + dataNodes = append(dataNodes, startConfigureReplicationTestVolumeServer(t, + &configureReplicationGateServer{gate: gate}, uint32(i+1))) + } + commandEnv, topologyInfo := configureReplicationTestEnv(dataNodes...) + + done := make(chan error, 1) + go func() { + done <- configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization) + }() + + for i := 0; i < maxParallelization; i++ { + select { + case <-gate.entered: + case <-time.After(10 * time.Second): + t.Fatalf("only %d servers started; serial execution would hold the second call", i) + } + } + + select { + case <-gate.entered: + t.Fatalf("more than %d servers ran while calls were held open", maxParallelization) + case <-time.After(100 * time.Millisecond): + } + + close(gate.release) + if err := <-done; err != nil { + t.Fatal(err) + } + + gate.mu.Lock() + defer gate.mu.Unlock() + if gate.maxActive != maxParallelization { + t.Errorf("at most %d servers ran together, expected %d", gate.maxActive, maxParallelization) + } + if gate.calls != servers { + t.Errorf("%d servers ran, expected %d", gate.calls, servers) + } +} diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index fc7da99fb3..7ef9ca5694 100644 --- a/weed/topology/store_replicate.go +++ b/weed/topology/store_replicate.go @@ -9,6 +9,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -46,8 +47,8 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi } // with no local volume and no replica to forward to, the write would be acknowledged without being stored - hasLocalVolume := s.HasVolume(volumeId) - if !hasLocalVolume && len(remoteLocations) == 0 { + vol := s.GetVolume(volumeId) + if vol == nil && len(remoteLocations) == 0 { err = fmt.Errorf("volume %d not found on %s:%d", volumeId, s.Ip, s.Port) glog.V(0).Infoln(err) return @@ -68,109 +69,141 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi }(time.Now()) } - if hasLocalVolume { - start := time.Now() - - inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk) - inFlightGauge.Inc() - defer inFlightGauge.Dec() - - isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, true, fsync) - stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds()) - if err != nil { - stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc() - err = fmt.Errorf("failed to write to local disk: %w", err) - glog.V(0).Infoln(err) - return + // Snapshot the fields the fan-out reads before the concurrent local write + // mutates the needle (it stamps the volume TTL and flags), so the two cannot + // race; the fan-out then touches only these locals plus read-only n.Data. + replicaTtl := n.Ttl + if vol != nil && replicaTtl == needle.EMPTY_TTL && vol.Ttl != needle.EMPTY_TTL { + replicaTtl = vol.Ttl + } + replicaLastModified := n.LastModified + replicaIsChunked := n.IsChunkedManifest() + replicaIsCompressed := n.IsCompressed() + replicaName := string(n.Name) + replicaMime := string(n.Mime) + replicaData := n.Data + replicaPairMap := make(map[string]string) + if replicaCount > 0 && n.HasPairs() { + tmpMap := make(map[string]string) + if pe := json.Unmarshal(n.Pairs, &tmpMap); pe != nil { + stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc() + glog.V(0).Infoln("Unmarshal pairs error:", pe) + } + for k, v := range tmpMap { + replicaPairMap[needle.PairNamePrefix+k] = v } } // Observe replication targets histogram for all operations (including zero) stats.VolumeServerReplicationTargets.Observe(float64(replicaCount)) - if replicaCount > 0 { //send to other replica locations - start := time.Now() + // Run the local disk write and the replica fan-out concurrently against the + // buffered needle (latency max(local, replicas), not the sum); ack after both. + var wg sync.WaitGroup + var localErr, remoteErr error + var localIsUnchanged bool - inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas) - inFlightGauge.Inc() - defer inFlightGauge.Dec() + if vol != nil { + wg.Add(1) + go func() { + defer wg.Done() + start := time.Now() + inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk) + inFlightGauge.Inc() + defer inFlightGauge.Dec() - err = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error { - u := url.URL{ - Scheme: "http", - Host: location.Url, - Path: r.URL.Path, + localIsUnchanged, localErr = s.WriteVolumeNeedle(volumeId, n, true, fsync) + stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds()) + if localErr != nil { + stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc() + localErr = fmt.Errorf("failed to write to local disk: %w", localErr) + glog.V(0).Infoln(localErr) } - q := url.Values{ - "type": {"replicate"}, - "ttl": {n.Ttl.String()}, - } - if fsync { - q.Set("fsync", "true") - } - if n.LastModified > 0 { - q.Set("ts", strconv.FormatUint(n.LastModified, 10)) - } - if n.IsChunkedManifest() { - q.Set("cm", "true") - } - u.RawQuery = q.Encode() - - pairMap := make(map[string]string) - if n.HasPairs() { - tmpMap := make(map[string]string) - err := json.Unmarshal(n.Pairs, &tmpMap) - if err != nil { - stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc() - glog.V(0).Infoln("Unmarshal pairs error:", err) - } - for k, v := range tmpMap { - pairMap[needle.PairNamePrefix+k] = v - } - } - bytesBuffer := buffer_pool.SyncPoolGetBuffer() - defer buffer_pool.SyncPoolPutBuffer(bytesBuffer) - - // volume server do not know about encryption - // TODO optimize here to compress data only once - uploadOption := &operation.UploadOption{ - UploadUrl: u.String(), - Filename: string(n.Name), - Cipher: false, - IsInputCompressed: n.IsCompressed(), - IsReplication: true, - MimeType: string(n.Mime), - PairMap: pairMap, - Jwt: jwt, - Md5: contentMd5, - BytesBuffer: bytesBuffer, - MaxAttempts: 1, // fail fast on a dead replica; the client write retries - } - - uploader, err := operation.NewUploader() - if err != nil { - glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String()) - return err - } - _, err = uploader.UploadData(ctx, n.Data, uploadOption) - if err != nil { - glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String()) - } - return err - }) - stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds()) - if err != nil { - stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc() - stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc() - reason := classifyReplicationError(err) - stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc() - err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err) - glog.V(0).Infoln(err) - return false, err - } - stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc() + }() } - return + + if replicaCount > 0 { + wg.Add(1) + go func() { + defer wg.Done() + start := time.Now() + inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas) + inFlightGauge.Inc() + defer inFlightGauge.Dec() + + remoteErr = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error { + u := url.URL{ + Scheme: "http", + Host: location.Url, + Path: r.URL.Path, + } + q := url.Values{ + "type": {"replicate"}, + "ttl": {replicaTtl.String()}, + } + if fsync { + q.Set("fsync", "true") + } + if replicaLastModified > 0 { + q.Set("ts", strconv.FormatUint(replicaLastModified, 10)) + } + if replicaIsChunked { + q.Set("cm", "true") + } + u.RawQuery = q.Encode() + + bytesBuffer := buffer_pool.SyncPoolGetBuffer() + defer buffer_pool.SyncPoolPutBuffer(bytesBuffer) + + // volume server do not know about encryption + uploadOption := &operation.UploadOption{ + UploadUrl: u.String(), + Filename: replicaName, + Cipher: false, + IsInputCompressed: replicaIsCompressed, + IsReplication: true, + MimeType: replicaMime, + PairMap: replicaPairMap, + Jwt: jwt, + Md5: contentMd5, + BytesBuffer: bytesBuffer, + MaxAttempts: 1, // fail fast on a dead replica; the client write retries + } + + uploader, uploaderErr := operation.NewUploader() + if uploaderErr != nil { + glog.Errorf("replication-UploadData, err:%v, url:%s", uploaderErr, u.String()) + return uploaderErr + } + _, uploadErr := uploader.UploadData(ctx, replicaData, uploadOption) + if uploadErr != nil { + glog.Errorf("replication-UploadData, err:%v, url:%s", uploadErr, u.String()) + } + return uploadErr + }) + stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds()) + if remoteErr != nil { + stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc() + stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc() + reason := classifyReplicationError(remoteErr) + stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc() + remoteErr = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, remoteErr) + glog.V(0).Infoln(remoteErr) + } else { + stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc() + } + }() + } + + wg.Wait() + + if localErr != nil { + return false, localErr + } + if remoteErr != nil { + return false, remoteErr + } + return localIsUnchanged, nil } // ReplicatedDelete deletes a needle from the local volume and sends delete diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go index 3901db744f..74eadaa48c 100644 --- a/weed/topology/store_replicate_test.go +++ b/weed/topology/store_replicate_test.go @@ -16,8 +16,11 @@ import ( "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) @@ -230,6 +233,176 @@ func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) { } } +// newReplicateTestStore builds a single-disk store in dir, draining the notify +// channels so volume mounts never block on a heartbeat reader. +func newReplicateTestStore(t *testing.T, dir string) *storage.Store { + t.Helper() + store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, + stats.DefaultDiskIOProbeConfig()) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.DeletedVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + close(done) + store.Close() + }) + return store +} + +// TestReplicatedWriteOverlapsLocalAndReplicaWrites runs the write path with a +// real local volume and a replica that stalls mid-request: the fan-out starts +// while the local write is still free to run, the local write lands before the +// replica unblocks, and the acknowledged write waits for both legs. +func TestReplicatedWriteOverlapsLocalAndReplicaWrites(t *testing.T) { + replicaStarted := make(chan url.Values, 1) + releaseReplica := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseReplica) }) } + replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + replicaStarted <- r.URL.Query() + <-releaseReplica + w.WriteHeader(http.StatusCreated) + if _, err := w.Write([]byte(`{"size":21}`)); err != nil { + t.Errorf("replica write response: %v", err) + } + })) + defer replica.Close() + defer release() + replicaHost := strings.TrimPrefix(replica.URL, "http://") + + masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{ + locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}}, + }) + + store := newReplicateTestStore(t, t.TempDir()) + volumeId := needle.VolumeId(7) + operation.InvalidateVolumeIdLocationCache(volumeId.String()) + if err := store.AddVolume(volumeId, "", storage.NeedleMapInMemory, "001", "15d", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatal(err) + } + + data := []byte("overlapped write data") + n := &needle.Needle{ + Id: 1, + Data: data, + Checksum: needle.NewCRC(data), + Ttl: needle.EMPTY_TTL, + } + r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/7,01637037d6", nil) + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + + done := make(chan error, 1) + go func() { + _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, volumeId, n, r, "") + done <- err + }() + + var q url.Values + select { + case q = <-replicaStarted: + case <-time.After(5 * time.Second): + t.Fatal("replica never received the fan-out request") + } + if got := q.Get("type"); got != "replicate" { + t.Errorf("replica type = %q, want replicate", got) + } + if got := q.Get("ttl"); got != "15d" { + t.Errorf("replica ttl = %q, want the inherited volume ttl 15d", got) + } + + select { + case err := <-done: + t.Fatalf("write returned before the replica responded: %v", err) + case <-time.After(50 * time.Millisecond): + } + + deadline := time.Now().Add(5 * time.Second) + for { + readBack := &needle.Needle{Id: n.Id} + if _, err := store.ReadVolumeNeedle(volumeId, readBack, nil, nil); err == nil && string(readBack.Data) == string(data) { + break + } + if time.Now().After(deadline) { + t.Fatal("local write did not land while the replica was still blocked") + } + time.Sleep(5 * time.Millisecond) + } + + release() + if err := <-done; err != nil { + t.Fatalf("ReplicatedWrite: %v", err) + } +} + +// TestReplicatedWriteFailsWhenEitherLegFails checks that a failed local write +// or a failed replica fan-out fails the whole request even when the other leg +// succeeded. +func TestReplicatedWriteFailsWhenEitherLegFails(t *testing.T) { + for _, tc := range []struct { + name string + volumeId needle.VolumeId + readonlyLocal bool + replicaStatus int + }{ + {name: "local write fails", volumeId: 8, readonlyLocal: true, replicaStatus: http.StatusCreated}, + {name: "replica write fails", volumeId: 9, replicaStatus: http.StatusInternalServerError}, + } { + t.Run(tc.name, func(t *testing.T) { + replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.replicaStatus) + })) + defer replica.Close() + replicaHost := strings.TrimPrefix(replica.URL, "http://") + + masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{ + locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}}, + }) + + store := newReplicateTestStore(t, t.TempDir()) + operation.InvalidateVolumeIdLocationCache(tc.volumeId.String()) + if err := store.AddVolume(tc.volumeId, "", storage.NeedleMapInMemory, "001", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatal(err) + } + if tc.readonlyLocal { + if err := store.MarkVolumeReadonly(tc.volumeId, false, false); err != nil { + t.Fatal(err) + } + } + + data := []byte("write data") + n := &needle.Needle{ + Id: 1, + Data: data, + Checksum: needle.NewCRC(data), + Ttl: needle.EMPTY_TTL, + } + r := httptest.NewRequest(http.MethodPost, fmt.Sprintf("http://127.0.0.1:8080/%d,01637037d6", tc.volumeId), nil) + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + + if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, tc.volumeId, n, r, ""); err == nil { + t.Fatal("ReplicatedWrite succeeded despite a failed leg") + } + }) + } +} + // TestReplicatedWriteRejectsWriteWithNoTarget verifies that a write this server // cannot store and has no replica to forward to fails instead of being // acknowledged with nothing written. diff --git a/weed/util/version/constants.go b/weed/util/version/constants.go index 852d163e40..37a1520405 100644 --- a/weed/util/version/constants.go +++ b/weed/util/version/constants.go @@ -9,7 +9,7 @@ import ( var ( MAJOR_VERSION = int32(4) - MINOR_VERSION = int32(49) + MINOR_VERSION = int32(50) VERSION_NUMBER = fmt.Sprintf("%d.%02d", MAJOR_VERSION, MINOR_VERSION) VERSION = util.SizeLimit + " " + VERSION_NUMBER COMMIT = ""