| Job Type | Active Jobs | Recent Activities |
|---|---|---|
| Loading... |
| Job Type | Enabled | Detector | In Flight | Max Runtime | Exec Global | Exec/Worker | Executor Workers | Effective Exec | Last Run |
|---|---|---|---|---|---|---|---|---|---|
| Loading... |
| Worker | Address | Capabilities | Load |
|---|---|---|---|
| Loading... |
| Time | Job ID | Worker | Duration |
|---|---|---|---|
| No data |
| Time | Job ID | Worker | Error |
|---|---|---|---|
| No data |
| Job ID | Type | State | Progress | Worker | Updated | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Time | Job Type | Request ID | Worker | Stage | Source | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Job ID | Type | State | Progress | Worker | Updated | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Time | Job Type | Job ID | Source | Stage | Message |
|---|---|---|---|---|---|
| Loading... |
| Job Type | Active Jobs | Recent Activities |
|---|---|---|
| Loading... |
| Job Type | Enabled | Detector | In Flight | Max Runtime | Exec Global | Exec/Worker | Executor Workers | Effective Exec | Last Run |
|---|---|---|---|---|---|---|---|---|---|
| Loading... |
| Worker | Address | Capabilities | Load |
|---|---|---|---|
| Loading... |
| Time | Job ID | Worker | Duration |
|---|---|---|---|
| No data |
| Time | Job ID | Worker | Error |
|---|---|---|---|
| No data |
| Job ID | Type | State | Progress | Worker | Updated | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Time | Job Type | Request ID | Worker | Stage | Source | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Job ID | Type | State | Progress | Worker | Updated | Message |
|---|---|---|---|---|---|---|
| Loading... |
| Time | Job Type | Job ID | Source | Stage | Message |
|---|---|---|---|---|---|
| Loading... |
"+apiErr.Code+"")
+ require.Zero(t, probe.readsAtDrain, "body reads before the error response")
+ require.Zero(t, volume.storedCount(), "chunks stored on the volume server")
+}
+
+// An invalid storage class or tag set is answered before the handler touches
+// the body, whichever way the bucket versions: the content-type sniff of a PUT
+// without Content-Type blocks on the first read.
+func TestPutObjectRejectsInvalidHeadersBeforeReadingBody(t *testing.T) {
+ for _, state := range []struct{ name, versioning string }{
+ {"unversioned", ""},
+ {"versioning enabled", s3_constants.VersioningEnabled},
+ {"versioning suspended", s3_constants.VersioningSuspended},
+ } {
+ for _, h := range invalidPutHeaders {
+ t.Run(state.name+"/"+h.name, func(t *testing.T) {
+ s3a, filer, volume := newHeaderTestServer(t, state.versioning)
+ req, probe := newHeaderTestRequest("/b/o", "o", "hello world", h.header, h.value)
+
+ s3a.PutObjectHandler(probe, req)
+
+ requireRejected(t, probe, volume, h.want)
+ filer.mu.Lock()
+ defer filer.mu.Unlock()
+ require.Empty(t, filer.entries)
+ })
+ }
+ }
+}
+
+func TestPutObjectPartRejectsInvalidHeadersBeforeReadingBody(t *testing.T) {
+ for _, h := range invalidPutHeaders {
+ t.Run(h.name, func(t *testing.T) {
+ s3a, filer, volume := newHeaderTestServer(t, "")
+ uploadID := s3a.generateUploadID("o")
+ filer.mu.Lock()
+ filer.entries["/buckets/b/.uploads/"+uploadID] = &filer_pb.Entry{
+ Name: uploadID, IsDirectory: true,
+ Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("o")},
+ }
+ filer.mu.Unlock()
+ req, probe := newHeaderTestRequest("/b/o?partNumber=1&uploadId="+uploadID, "o", "hello world", h.header, h.value)
+
+ s3a.PutObjectPartHandler(probe, req)
+
+ requireRejected(t, probe, volume, h.want)
+ })
+ }
+}
+
+// A directory marker is validated like any other PUT.
+func TestPutObjectDirectoryMarkerRejectsInvalidHeaders(t *testing.T) {
+ for _, h := range invalidPutHeaders {
+ t.Run(h.name, func(t *testing.T) {
+ s3a, filer, volume := newHeaderTestServer(t, "")
+ req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", h.header, h.value)
+
+ s3a.PutObjectHandler(probe, req)
+
+ requireRejected(t, probe, volume, h.want)
+ filer.mu.Lock()
+ defer filer.mu.Unlock()
+ require.Nil(t, filer.entries["/buckets/b/dir"], "directory marker created")
+ })
+ }
+}
+
+func TestPutObjectDirectoryMarkerStoresHeaders(t *testing.T) {
+ s3a, filer, _ := newHeaderTestServer(t, "")
+ req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", s3_constants.AmzStorageClass, "STANDARD_IA")
+ req.Header.Set(s3_constants.AmzObjectTagging, "k1=v1&k2=v2")
+
+ s3a.PutObjectHandler(probe, req)
+
+ require.Equal(t, http.StatusOK, probe.Code, probe.Body.String())
+ filer.mu.Lock()
+ entry := filer.entries["/buckets/b/dir"]
+ require.NotNil(t, entry)
+ require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass]))
+ require.Equal(t, "v1", string(entry.Extended[S3TAG_PREFIX+"k1"]))
+ require.Equal(t, "v2", string(entry.Extended[S3TAG_PREFIX+"k2"]))
+ filer.mu.Unlock()
+
+ getReq := httptest.NewRequest(http.MethodGet, "/b/dir/?tagging", nil)
+ getReq = mux.SetURLVars(getReq, map[string]string{"bucket": "b", "object": "dir/"})
+ rec := httptest.NewRecorder()
+ s3a.GetObjectTaggingHandler(rec, getReq)
+ require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
+ var tagging Tagging
+ require.NoError(t, xml.Unmarshal(rec.Body.Bytes(), &tagging))
+ require.ElementsMatch(t, []Tag{{Key: "k1", Value: "v1"}, {Key: "k2", Value: "v2"}}, tagging.TagSet.Tag)
+}
+
+// A valid storage class still stores, and the content type is still sniffed.
+func TestPutObjectValidStorageClassSniffsContentType(t *testing.T) {
+ s3a, filer, volume := newHeaderTestServer(t, "")
+ req, probe := newHeaderTestRequest("/b/o", "o", "hello world", s3_constants.AmzStorageClass, "STANDARD_IA")
+
+ s3a.PutObjectHandler(probe, req)
+
+ require.Equal(t, http.StatusOK, probe.Code, probe.Body.String())
+ require.Equal(t, 1, volume.storedCount())
+ filer.mu.Lock()
+ defer filer.mu.Unlock()
+ entry := filer.entries["/buckets/b/o"]
+ require.NotNil(t, entry)
+ require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass]))
+ require.Equal(t, "text/plain; charset=utf-8", entry.Attributes.Mime)
+}
+
+// CreateMultipartUpload stores the storage class for the completed object,
+// so it must reject an invalid one as PutObject and CopyObject do.
+func TestParseS3MetadataRejectsInvalidStorageClass(t *testing.T) {
+ req := httptest.NewRequest(http.MethodPost, "http://s3/b/o?uploads", nil)
+ req.Header.Set(s3_constants.AmzStorageClass, "BOGUS")
+ _, code := ParseS3Metadata(req, nil, false)
+ require.Equal(t, s3err.ErrInvalidStorageClass, code)
+
+ req.Header.Set(s3_constants.AmzStorageClass, "GLACIER")
+ metadata, code := ParseS3Metadata(req, nil, false)
+ require.Equal(t, s3err.ErrNone, code)
+ require.Equal(t, "GLACIER", string(metadata[s3_constants.AmzStorageClass]))
+}
diff --git a/weed/server/filer_server_handlers_iam_grpc.go b/weed/server/filer_server_handlers_iam_grpc.go
index be98356614..508dcd2c3b 100644
--- a/weed/server/filer_server_handlers_iam_grpc.go
+++ b/weed/server/filer_server_handlers_iam_grpc.go
@@ -17,11 +17,11 @@ import (
)
// IamGrpcServer implements the IAM gRPC service on the filer.
-// Auth is opt-in: when jwt.filer_signing.key is set in security.toml the
-// service requires a Bearer token in the "authorization" metadata signed with
-// that key; when it is empty every RPC is accepted unauthenticated, matching
-// the rest of SeaweedFS's gRPC surface. Operators who expose the filer gRPC
-// port beyond a trusted network should configure the key.
+// When jwt.filer_signing.key is set, every RPC requires a Bearer token signed
+// with it; otherwise RPCs run unauthenticated, except the OIDC provider and
+// role RPCs, which are refused. The key gates only this service: the filer RPCs
+// on the same port can still read and write the IAM store under /etc/iam.
+// Protect a filer gRPC port reachable by untrusted clients with [grpc.filer] mTLS.
type IamGrpcServer struct {
iam_pb.UnimplementedSeaweedIdentityAccessManagementServer
credentialManager *credential.CredentialManager
diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go
index 17c6f4fdd9..bb05ce6217 100644
--- a/weed/shell/command_fs_distribute_chunks.go
+++ b/weed/shell/command_fs_distribute_chunks.go
@@ -23,6 +23,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/util"
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+ "google.golang.org/protobuf/proto"
)
func init() {
@@ -116,6 +117,7 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
if err != nil {
return fmt.Errorf("lookup %s: %v", path, err)
}
+ readEntry := proto.Clone(entry).(*filer_pb.Entry)
chunks, hadManifest, err := resolveEntryDataChunks(commandEnv, entry)
if err != nil {
@@ -180,6 +182,11 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
fmt.Fprintf(writer, "\nExecuting redistribution...\n")
defer util_http.GetGlobalHttpClient().CloseIdleConnections()
+ readFids := make(map[string]bool, len(chunks))
+ for _, chunk := range chunks {
+ readFids[chunk.GetFileIdString()] = true
+ }
+
movedCount, err := executeChunkMoves(commandEnv, writer, moves, volumeInfoMap)
if err != nil {
return err
@@ -191,8 +198,14 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
finalChunks := chunks
if hadManifest {
- // no chunk deleter here: a failed fold reports the blobs it saved
- remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), nil, chunks)
+ dropSaved := func(saved []*filer_pb.FileChunk) {
+ var blobs []orphanedNeedle
+ for _, chunk := range saved {
+ blobs = append(blobs, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: chunk.GetFileIdString()})
+ }
+ deleteOrphanedNeedles(commandEnv, util.FullPath(path), blobs, true)
+ }
+ remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), dropSaved, chunks)
if manErr != nil {
fmt.Fprintf(writer, "WARNING: re-manifestize failed: %v. Writing flat chunk list.\n", manErr)
} else {
@@ -203,12 +216,24 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
entry.Chunks = finalChunks
fmt.Fprintf(writer, "\nUpdating filer metadata (%d chunks moved)...\n", movedCount)
+ var stale bool
err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
- Directory: dir,
- Entry: entry,
- })
+ var uErr error
+ stale, uErr = updateEntryIfUnchanged(context.Background(), client, dir, entry, readEntry)
+ return uErr
})
+ if stale {
+ var copies []orphanedNeedle
+ for _, chunk := range slices.Concat(chunks, finalChunks) {
+ if fid := chunk.GetFileIdString(); !readFids[fid] {
+ readFids[fid] = true
+ copies = append(copies, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: fid})
+ }
+ }
+ deleteOrphanedNeedles(commandEnv, util.FullPath(path), copies, true)
+ fmt.Fprintf(writer, "%s was changed or deleted during redistribution; the filer kept that version and the moved copies were deleted.\n", path)
+ return err
+ }
if err != nil {
fmt.Fprintf(writer, "FAILED to update filer metadata: %v\n", err)
fmt.Fprintf(writer, "WARNING: New chunks were uploaded but metadata was not updated.\n")
diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go
index 99b6140943..26059da18e 100644
--- a/weed/shell/command_fs_merge_volumes.go
+++ b/weed/shell/command_fs_merge_volumes.go
@@ -6,6 +6,7 @@ import (
"flag"
"fmt"
"io"
+ "math"
"net/http"
"sort"
"strconv"
@@ -21,6 +22,8 @@ import (
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"golang.org/x/exp/maps"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
"github.com/seaweedfs/seaweedfs/weed/operation"
@@ -54,6 +57,9 @@ func (c *commandFsMergeVolumes) Help() string {
-toVolumeId accepts a comma-separated list. With -fromVolumeId, the source
chunks are distributed across the listed volumes by remaining capacity, so a
volume that does not fit into any single target can still be cleared.
+
+ -apply needs the shell lock (run "lock" first), so two runs never move the
+ same chunks at once.
`
}
@@ -72,6 +78,11 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
if err = fsMergeVolumesCommand.Parse(args); err != nil {
return err
}
+ if *apply {
+ if err = commandEnv.confirmIsLocked(args); err != nil {
+ return err
+ }
+ }
dir := *dirArg
if dir != "/" {
@@ -171,7 +182,16 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
// look like mergeVolumes hadn't done anything. Track the old fids
// and delete them below after the filer update commits, so the
// filer never points at a fid we already deleted.
- var movedSources []orphanedNeedle
+ var moved rewrittenNeedles
+ // The update only lands while the filer still holds the entry as
+ // the walk read it, so keep that copy before the first chunk is
+ // rewritten in place.
+ var readEntry *filer_pb.Entry
+ keepReadEntry := func() {
+ if readEntry == nil {
+ readEntry = proto.Clone(entry).(*filer_pb.Entry)
+ }
+ }
for i, chunk := range entry.Chunks {
if chunk.IsChunkManifest {
if !c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.VolumeId)) {
@@ -190,14 +210,18 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
if !changed || !*apply {
continue
}
+ keepReadEntry()
entry.Chunks[i] = newChunk
entryChanged = true
- movedSources = append(movedSources, rewritten.sources...)
+ moved.sources = append(moved.sources, rewritten.sources...)
// The old manifest needle is always orphaned when we
// replace it with a freshly uploaded one, even when the
// rewrite was triggered by sub-chunk moves rather than the
// manifest volume itself being in the plan.
- movedSources = append(movedSources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid})
+ moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid})
+ moved.targets = append(moved.targets, rewritten.targets...)
+ moved.targets = append(moved.targets, orphanedNeedle{volumeId: newChunk.Fid.VolumeId, fileId: newChunk.GetFileIdString()})
+ moved.reservations = append(moved.reservations, rewritten.reservations...)
continue
}
@@ -206,6 +230,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
continue
}
seen.record(chunkVolumeId)
+ keepReadEntry()
oldFid := chunk.GetFileIdString()
oldVid := chunk.Fid.VolumeId
@@ -224,20 +249,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
continue
}
entryChanged = true
- movedSources = append(movedSources, orphanedNeedle{volumeId: oldVid, fileId: oldFid})
+ moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldVid, fileId: oldFid})
+ moved.targets = append(moved.targets, orphanedNeedle{volumeId: uint32(toVolumeId), fileId: chunk.GetFileIdString()})
+ moved.reservations = append(moved.reservations, reservation{src: chunkVolumeId, dst: toVolumeId, size: chunk.Size})
}
if entryChanged {
- if uErr := filer_pb.UpdateEntry(context.Background(), filerClient, &filer_pb.UpdateEntryRequest{
- Directory: string(parentPath),
- Entry: entry,
- }); uErr != nil {
+ stale, uErr := updateEntryIfUnchanged(context.Background(), filerClient, string(parentPath), entry, readEntry)
+ if stale {
+ fmt.Printf("skip %s: changed during merge\n", entryPath)
+ moved.targets = unreferencedTargets(context.Background(), filerClient, lookupFn, parentPath, entry.Name, moved.targets)
+ moved.abandon(commandEnv, entryPath, plan)
+ if len(entry.HardLinkId) > 0 {
+ processedHardLinks.Delete(string(entry.HardLinkId))
+ }
+ return nil
+ }
+ if uErr != nil {
fmt.Printf("failed to update %s: %v\n", entryPath, uErr)
// Filer still references the source fids. Deleting them
// now would lose data — abandon the cleanup for this
// entry and let fsck reconcile later.
return nil
}
- deleteOrphanedNeedles(commandEnv, entryPath, movedSources, false)
+ deleteOrphanedNeedles(commandEnv, entryPath, moved.sources, false)
}
return nil
})
@@ -313,6 +347,61 @@ type rewrittenNeedles struct {
reservations []reservation
}
+// abandon drops a rewrite the filer never took: the copies on the target
+// volumes belong to nobody, and the plan gets back the room they held, or a
+// multi-target merge keeps counting bytes it just deleted and skips later
+// chunks for no reason.
+func (r rewrittenNeedles) abandon(commandEnv *CommandEnv, entryPath util.FullPath, plan *mergePlan) {
+ deleteOrphanedNeedles(commandEnv, entryPath, r.targets, true)
+ for _, res := range r.reservations {
+ plan.release(res.src, res.dst, res.size)
+ }
+}
+
+// updateEntryIfUnchanged writes entry back only while the filer still holds
+// readEntry. An unconditional write would revert a change made since the read
+// and the filer would delete the newer chunks as garbage. stale reports that
+// the filer refused because the entry was changed or deleted, so nothing was
+// written and the copies made for this update are referenced by nothing.
+func updateEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, entry, readEntry *filer_pb.Entry) (stale bool, err error) {
+ err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
+ Directory: dir,
+ Entry: entry,
+ Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{
+ {Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: readEntry},
+ }},
+ })
+ switch status.Code(err) {
+ case codes.FailedPrecondition, codes.NotFound:
+ return true, err
+ }
+ return false, err
+}
+
+// unreferencedTargets drops the copies the filer's current entry points at.
+// moveChunk keeps the key and cookie, so an overlapping run that moved the same
+// chunk to the same volume wrote the same fid, and its committed update is what
+// refused this one. When the entry cannot be read, no copy is reported.
+func unreferencedTargets(ctx context.Context, client filer_pb.SeaweedFilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir util.FullPath, name string, targets []orphanedNeedle) []orphanedNeedle {
+ resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(dir), Name: name})
+ if errors.Is(err, filer_pb.ErrNotFound) {
+ return targets
+ }
+ if err != nil {
+ return nil
+ }
+ chunks := resp.Entry.GetChunks()
+ dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, chunks, 0, math.MaxInt64, nil)
+ if err != nil {
+ return nil
+ }
+ referenced := make(map[string]bool)
+ for _, chunk := range slices.Concat(chunks, dataChunks, manifestChunks) {
+ referenced[chunk.GetFileIdString()] = true
+ }
+ return slices.DeleteFunc(targets, func(t orphanedNeedle) bool { return referenced[t.fileId] })
+}
+
// deleteOrphanedNeedles fans out BatchDelete RPCs to every replica of each
// needle's volume. Errors are logged but never returned — the data is already
// orphan at this point, so a failed cleanup just leaves work for a later fsck.
@@ -740,16 +829,8 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk(
var rewritten rewrittenNeedles
// Every path out of here that is not the successful return abandons the
- // rewrite with the filer still pointing at the old manifest, so whatever
- // already landed on the target volumes belongs to nobody. Give the plan
- // back the room those copies were holding, or a multi-target merge keeps
- // counting bytes it just deleted and skips later chunks for no reason.
- abandon := func() {
- deleteOrphanedNeedles(commandEnv, entryPath, rewritten.targets, true)
- for _, r := range rewritten.reservations {
- plan.release(r.src, r.dst, r.size)
- }
- }
+ // rewrite with the filer still pointing at the old manifest.
+ abandon := func() { rewritten.abandon(commandEnv, entryPath, plan) }
anySubChanged := false
for i, sub := range subChunks {
if sub.IsChunkManifest {
diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go
index d02dd60b71..37500e2d45 100644
--- a/weed/shell/command_fs_merge_volumes_test.go
+++ b/weed/shell/command_fs_merge_volumes_test.go
@@ -1,12 +1,22 @@
package shell
import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "slices"
"strings"
"sync"
"testing"
+ "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+ "google.golang.org/protobuf/proto"
)
func newMergeCmd(limitMB uint64, vols ...*master_pb.VolumeInformationMessage) *commandFsMergeVolumes {
@@ -394,3 +404,112 @@ func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) {
t.Fatalf("expected no warning when all needles were seen, got %q", sb.String())
}
}
+
+type conditionalUpdateFiler struct {
+ filer_pb.SeaweedFilerClient
+ err error
+ req *filer_pb.UpdateEntryRequest
+ current *filer_pb.Entry
+ lookupErr error
+ lookupReq *filer_pb.LookupDirectoryEntryRequest
+}
+
+func (s *conditionalUpdateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+ s.lookupReq = req
+ if s.lookupErr != nil {
+ return nil, s.lookupErr
+ }
+ return &filer_pb.LookupDirectoryEntryResponse{Entry: s.current}, nil
+}
+
+func (s *conditionalUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
+ s.req = req
+ if s.err != nil {
+ return nil, s.err
+ }
+ return &filer_pb.UpdateEntryResponse{}, nil
+}
+
+// The write-back is conditioned on the entry as it was read, and only a refusal
+// that proves nothing was written counts as stale: the caller then deletes the
+// copies it made, which would lose data if the update might have landed.
+func TestUpdateEntryIfUnchanged(t *testing.T) {
+ read := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}}
+ moved := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 7}}}}
+
+ for _, tc := range []struct {
+ name string
+ err error
+ stale bool
+ }{
+ {"committed", nil, false},
+ {"changed", status.Error(codes.FailedPrecondition, "precondition failed"), true},
+ {"deleted", status.Error(codes.NotFound, "not found"), true},
+ {"outcome unknown", status.Error(codes.Unavailable, "filer down"), false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ fake := &conditionalUpdateFiler{err: tc.err}
+ stale, err := updateEntryIfUnchanged(context.Background(), fake, "/dir", moved, read)
+ if stale != tc.stale || (err != nil) != (tc.err != nil) {
+ t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err)
+ }
+ clauses := fake.req.GetCondition().GetClauses()
+ if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL ||
+ !proto.Equal(clauses[0].ExpectedEntry, read) || !proto.Equal(fake.req.Entry, moved) {
+ t.Fatalf("request %v must write the moved entry conditioned on the read one", fake.req)
+ }
+ })
+ }
+}
+
+// A refused update may have been refused by an overlapping run that moved the
+// same chunk to the same volume: moveChunk keeps the key and cookie, so that
+// run's committed entry points at the very fid this run would delete.
+func TestUnreferencedTargets(t *testing.T) {
+ committed := orphanedNeedle{volumeId: 9, fileId: "9,0101000007"}
+ own := orphanedNeedle{volumeId: 9, fileId: "9,0202000007"}
+ manifestCopy := orphanedNeedle{volumeId: 9, fileId: "9,0303000007"}
+ targets := func() []orphanedNeedle { return []orphanedNeedle{committed, own, manifestCopy} }
+ direct := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId}}}
+
+ blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId, Size: 10}}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write(blob)
+ }))
+ defer volumeServer.Close()
+ lookupFn := func(_ context.Context, fileId string) ([]string, error) {
+ if fileId != manifestCopy.fileId {
+ return nil, fmt.Errorf("volume of %s unavailable", fileId)
+ }
+ return []string{volumeServer.URL + "/" + fileId}, nil
+ }
+ viaManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: manifestCopy.fileId, IsChunkManifest: true, Size: 10}}}
+ unreadableManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: "9,0404000007", IsChunkManifest: true, Size: 10}}}
+
+ for _, tc := range []struct {
+ name string
+ current *filer_pb.Entry
+ lookupErr error
+ want []orphanedNeedle
+ }{
+ {"another run committed one copy", direct, nil, []orphanedNeedle{own, manifestCopy}},
+ {"another run committed copies through a manifest", viaManifest, nil, []orphanedNeedle{own}},
+ {"manifest unreadable", unreadableManifest, nil, nil},
+ {"entry deleted", direct, status.Error(codes.NotFound, "not found"), targets()},
+ {"entry unreadable", direct, status.Error(codes.Unavailable, "filer down"), nil},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ fake := &conditionalUpdateFiler{current: tc.current, lookupErr: tc.lookupErr}
+ got := unreferencedTargets(context.Background(), fake, lookupFn, "/dir", "f", targets())
+ if !slices.Equal(got, tc.want) {
+ t.Fatalf("got %v, want %v", got, tc.want)
+ }
+ if fake.lookupReq.GetDirectory() != "/dir" || fake.lookupReq.GetName() != "f" {
+ t.Fatalf("looked up %v, want /dir f", fake.lookupReq)
+ }
+ })
+ }
+}
diff --git a/weed/shell/command_fs_meta_change_volume_id.go b/weed/shell/command_fs_meta_change_volume_id.go
index ef5a9386a2..23bedd5f99 100644
--- a/weed/shell/command_fs_meta_change_volume_id.go
+++ b/weed/shell/command_fs_meta_change_volume_id.go
@@ -12,6 +12,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/util"
+ "google.golang.org/protobuf/proto"
)
func init() {
@@ -78,6 +79,7 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(*dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
if !entry.IsDirectory {
var hasChanges bool
+ var readEntry *filer_pb.Entry
for _, chunk := range entry.Chunks {
if chunk.IsChunkManifest {
fmt.Printf("Change volume id for large file is not implemented yet: %s/%s\n", parentPath, entry.Name)
@@ -85,6 +87,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
}
chunkVolumeId := chunk.Fid.VolumeId
if toVolumeId, found := mapping[needle.VolumeId(chunkVolumeId)]; found {
+ if readEntry == nil {
+ readEntry = proto.Clone(entry).(*filer_pb.Entry)
+ }
hasChanges = true
chunk.Fid.VolumeId = uint32(toVolumeId)
chunk.FileId = ""
@@ -93,10 +98,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
if hasChanges {
println("Updating", parentPath, entry.Name)
if *applyChanges {
- if updateErr := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
- Directory: string(parentPath),
- Entry: entry,
- }); updateErr != nil {
+ if stale, updateErr := updateEntryIfUnchanged(context.Background(), client, string(parentPath), entry, readEntry); stale {
+ fmt.Printf("skip %s/%s: changed since it was read\n", parentPath, entry.Name)
+ } else if updateErr != nil {
fmt.Printf("failed to update %s/%s: %v\n", parentPath, entry.Name, updateErr)
}
}
diff --git a/weed/shell/command_s3_versions_audit.go b/weed/shell/command_s3_versions_audit.go
index 1be0b16ae1..c7b5aad164 100644
--- a/weed/shell/command_s3_versions_audit.go
+++ b/weed/shell/command_s3_versions_audit.go
@@ -13,6 +13,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/seaweedfs/seaweedfs/weed/util"
+ "google.golang.org/protobuf/proto"
)
func init() {
@@ -228,6 +229,7 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient
if entry.Extended == nil {
return nil
}
+ readEntry := proto.Clone(entry).(*filer_pb.Entry)
delete(entry.Extended, s3_constants.ExtLatestVersionIdKey)
delete(entry.Extended, s3_constants.ExtLatestVersionFileNameKey)
// Also clear the cached list metadata so a stale size/mtime/etag
@@ -237,10 +239,9 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient
delete(entry.Extended, s3_constants.ExtLatestVersionETagKey)
delete(entry.Extended, s3_constants.ExtLatestVersionOwnerKey)
delete(entry.Extended, s3_constants.ExtLatestVersionIsDeleteMarker)
- _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
- Directory: string(parentPath),
- Entry: entry,
- })
+ // A version written since the audit read the directory has moved the
+ // pointer, and clearing it would hide that version.
+ _, err := updateEntryIfUnchanged(ctx, client, string(parentPath), entry, readEntry)
return err
}
diff --git a/weed/shell/command_s3_versions_audit_test.go b/weed/shell/command_s3_versions_audit_test.go
new file mode 100644
index 0000000000..b32c1051e4
--- /dev/null
+++ b/weed/shell/command_s3_versions_audit_test.go
@@ -0,0 +1,37 @@
+package shell
+
+import (
+ "context"
+ "testing"
+
+ "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+ s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+// The heal clears the pointer only while the directory still names the file
+// the audit found missing; a version written since then has moved it.
+func TestHealStrandedPointerIsConditional(t *testing.T) {
+ fake := &conditionalUpdateFiler{}
+ entry := &filer_pb.Entry{Name: "obj.versions", IsDirectory: true, Extended: map[string][]byte{
+ s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+ s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"),
+ "keep": []byte("x"),
+ }}
+ if err := healStrandedPointer(context.Background(), fake, "/buckets/b", entry); err != nil {
+ t.Fatalf("heal: %v", err)
+ }
+
+ clauses := fake.req.GetCondition().GetClauses()
+ if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL {
+ t.Fatalf("heal must be conditioned on the audited entry, got %v", fake.req.GetCondition())
+ }
+ if string(clauses[0].ExpectedEntry.Extended[s3_constants.ExtLatestVersionIdKey]) != "v1" {
+ t.Fatalf("expected entry must hold the pointer the audit saw, got %v", clauses[0].ExpectedEntry.Extended)
+ }
+ written := fake.req.Entry.Extended
+ _, idKept := written[s3_constants.ExtLatestVersionIdKey]
+ _, nameKept := written[s3_constants.ExtLatestVersionFileNameKey]
+ if idKept || nameKept || string(written["keep"]) != "x" {
+ t.Fatalf("written entry must clear only the pointer, got %v", written)
+ }
+}
diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go
index 67b0fa08e5..efabf1ceb1 100644
--- a/weed/shell/command_volume_fsck.go
+++ b/weed/shell/command_volume_fsck.go
@@ -105,6 +105,16 @@ func (c *commandVolumeFsck) Help() string {
walked directory by directory; there, avoid running with -reallyDeleteFromVolume
while files are being renamed or moved.
+ With -findMissingChunksInFiler, -volumeId may name a volume the master no
+ longer lists, to find the filer entries that still reference it. A volume is
+ also unlisted while its server is offline, so purging these entries with
+ -reallyDeleteFilerEntries needs -confirmMissingVolumeDeletion as well; pass
+ it only after checking that every expected volume server is online. A purge
+ removes the whole filer entry, including its chunks on other volumes.
+ A -collection filter cannot be checked against an unlisted volume. With a
+ filter, a purge only reports references to unlisted volumes, and it refuses
+ to run when -volumeId names one.
+
`
}
@@ -123,6 +133,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, "