diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index ca1c0114c..324cbf5bb 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -250,9 +250,11 @@ message CreateEntryRequest { repeated int32 signatures = 5; bool skip_check_parent_directory = 6; // Optional precondition evaluated against the current entry atomically with - // the write, under the filer's per-path lock. The caller must route the - // key's writes to this entry's owner filer for the check to be authoritative. + // the write, under the filer's per-path lock. WriteCondition condition = 7; + // Set on a create a filer forwarded to the entry's ring owner, so the owner + // applies it locally instead of forwarding again. Clients leave it unset. + bool is_moved = 8; } // WriteCondition is the precondition the filer evaluates against the existing @@ -561,6 +563,9 @@ message AssignVolumeResponse { string error = 8; Location location = 9; repeated Location replicas = 10; + // fsync is the storage rule's fsync decision for the assigned path, so the + // client can carry it onto the volume server upload request. + bool fsync = 11; } message LookupVolumeRequest { @@ -576,7 +581,6 @@ message Location { string public_url = 2; uint32 grpc_port = 3; string data_center = 4; - bool data_in_remote = 5; } message LookupVolumeResponse { map locations_map = 1; diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index dbd18dfaf..fd440bbae 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -250,9 +250,11 @@ message CreateEntryRequest { repeated int32 signatures = 5; bool skip_check_parent_directory = 6; // Optional precondition evaluated against the current entry atomically with - // the write, under the filer's per-path lock. The caller must route the - // key's writes to this entry's owner filer for the check to be authoritative. + // the write, under the filer's per-path lock. WriteCondition condition = 7; + // Set on a create a filer forwarded to the entry's ring owner, so the owner + // applies it locally instead of forwarding again. Clients leave it unset. + bool is_moved = 8; } // WriteCondition is the precondition the filer evaluates against the existing diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 14e532d49..46bacb313 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v7.35.1 +// protoc v7.35.0 // source: filer.proto package filer_pb @@ -1374,9 +1374,11 @@ type CreateEntryRequest struct { Signatures []int32 `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"` SkipCheckParentDirectory bool `protobuf:"varint,6,opt,name=skip_check_parent_directory,json=skipCheckParentDirectory,proto3" json:"skip_check_parent_directory,omitempty"` // Optional precondition evaluated against the current entry atomically with - // the write, under the filer's per-path lock. The caller must route the - // key's writes to this entry's owner filer for the check to be authoritative. - Condition *WriteCondition `protobuf:"bytes,7,opt,name=condition,proto3" json:"condition,omitempty"` + // the write, under the filer's per-path lock. + Condition *WriteCondition `protobuf:"bytes,7,opt,name=condition,proto3" json:"condition,omitempty"` + // Set on a create a filer forwarded to the entry's ring owner, so the owner + // applies it locally instead of forwarding again. Clients leave it unset. + IsMoved bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1460,6 +1462,13 @@ func (x *CreateEntryRequest) GetCondition() *WriteCondition { return nil } +func (x *CreateEntryRequest) GetIsMoved() bool { + if x != nil { + return x.IsMoved + } + return false +} + // WriteCondition is the precondition the filer evaluates against the existing // entry before writing, under the per-path lock. A failed condition returns // FilerError PRECONDITION_FAILED. The client maps request semantics (e.g. RFC @@ -7075,7 +7084,7 @@ const file_filer_proto_rawDesc = "" + "\bctime_ns\x18\x14 \x01(\x05R\actimeNs\x12\x1b\n" + "\tcrtime_ns\x18\x15 \x01(\x05R\bcrtimeNs\x12\x14\n" + "\x05atime\x18\x16 \x01(\x03R\x05atime\x12\x19\n" + - "\batime_ns\x18\x17 \x01(\x05R\aatimeNs\"\xba\x02\n" + + "\batime_ns\x18\x17 \x01(\x05R\aatimeNs\"\xd5\x02\n" + "\x12CreateEntryRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" + "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x15\n" + @@ -7085,7 +7094,8 @@ const file_filer_proto_rawDesc = "" + "signatures\x18\x05 \x03(\x05R\n" + "signatures\x12=\n" + "\x1bskip_check_parent_directory\x18\x06 \x01(\bR\x18skipCheckParentDirectory\x126\n" + - "\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\"\xbc\x04\n" + + "\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" + + "\bis_moved\x18\b \x01(\bR\aisMoved\"\xbc\x04\n" + "\x0eWriteCondition\x129\n" + "\aclauses\x18\x01 \x03(\v2\x1f.filer_pb.WriteCondition.ClauseR\aclauses\x1a\x91\x02\n" + "\x06Clause\x121\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 6d944c6b2..0b7b4d153 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1014,6 +1014,16 @@ func (m *CreateEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.IsMoved { + i-- + if m.IsMoved { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x40 + } if m.Condition != nil { size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i]) if err != nil { @@ -6687,6 +6697,9 @@ func (m *CreateEntryRequest) SizeVT() (n int) { l = m.Condition.SizeVT() n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + if m.IsMoved { + n += 2 + } n += len(m.unknownFields) return n } @@ -11746,6 +11759,26 @@ func (m *CreateEntryRequest) UnmarshalVT(dAtA []byte) error { return err } iNdEx = postIndex + case 8: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field IsMoved", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.IsMoved = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index b6ed2d411..b86d82475 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -200,6 +200,36 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr resp = &filer_pb.CreateEntryResponse{} + // An exclusive or conditional create is a read-then-write that the per-path + // lock below only makes atomic on this filer, while the store's insert is an + // upsert. Route it to the entry's ring owner so one filer's lock arbitrates + // every creator cluster-wide; is_moved bounds this to one hop. Plain creates + // are upserts either way and stay local. + if !req.IsMoved && (req.OExcl || conditionIsSet(req.Condition)) { + fullpath := util.NewFullPath(req.Directory, req.Entry.Name) + // Held apart from the named resp, which the local path below writes into: + // a failed forward must not leave it nil. + var ownerResp *filer_pb.CreateEntryResponse + handled, forwardErr := fs.forwardToWriteOwner(ctx, entryRouteKey(fullpath), func(owner pb.ServerAddress) error { + glog.V(2).InfofCtx(ctx, "CreateEntry %s: forwarding to owner %s", fullpath, owner) + req.IsMoved = true + return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + forwarded, e := client.CreateEntry(ctx, req) + if e != nil { + return e + } + ownerResp = forwarded + return nil + }) + }) + if handled { + if forwardErr != nil { + return &filer_pb.CreateEntryResponse{}, forwardErr + } + return ownerResp, nil + } + } + chunks, garbage, err2 := fs.cleanupChunks(ctx, util.Join(req.Directory, req.Entry.Name), nil, req.Entry) if err2 != nil { return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry cleanupChunks %s %s: %v", req.Directory, req.Entry.Name, err2) diff --git a/weed/server/filer_grpc_server_create_route_test.go b/weed/server/filer_grpc_server_create_route_test.go new file mode 100644 index 000000000..f5f4dc99b --- /dev/null +++ b/weed/server/filer_grpc_server_create_route_test.go @@ -0,0 +1,175 @@ +package weed_server + +import ( + "context" + "fmt" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// createRouteServer builds a filer whose ring holds two unreachable peers, so a +// forwarded create fails at the dial and is distinguishable from a local apply. +func createRouteServer(t *testing.T) (*FilerServer, *renameTestStore) { + t.Helper() + const self = pb.ServerAddress("127.0.0.1:18888") + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + f.DirBucketsPath = "/buckets" + f.Dlm = lock_manager.NewDistributedLockManager(self) + f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self, "127.0.0.1:18889"}, 1) + return &FilerServer{ + filer: f, + option: &FilerOption{Host: self}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + entryLockTable: util.NewLockTable[util.FullPath](), + }, store +} + +// peerOwnedName returns a child name under /test whose entry the ring assigns to +// the peer, so a routed create must leave this filer. +func peerOwnedName(t *testing.T, fs *FilerServer) string { + t.Helper() + for i := 0; i < 4000; i++ { + name := fmt.Sprintf("obj-%d", i) + key := entryRouteKey(util.NewFullPath("/test", name)) + if fs.filer.Dlm.LockRing.GetPrimary(key) != fs.option.Host { + return name + } + } + t.Skip("no name owned by the peer") + return "" +} + +func createReq(name string, oExcl bool) *filer_pb.CreateEntryRequest { + return &filer_pb.CreateEntryRequest{ + Directory: "/test", + OExcl: oExcl, + SkipCheckParentDirectory: true, + Entry: &filer_pb.Entry{ + Name: name, + Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1}, + }, + } +} + +// An exclusive create for a peer-owned path must leave this filer rather than +// take the local lock, whose arbitration would not bind the other filers. +func TestCreateEntryExclusiveLeavesNonOwner(t *testing.T) { + fs, store := createRouteServer(t) + name := peerOwnedName(t, fs) + + _, err := fs.CreateEntry(context.Background(), createReq(name, true)) + if err == nil { + t.Fatal("exclusive create on a non-owner must not be applied locally") + } + if _, found := store.entries[string(util.NewFullPath("/test", name))]; found { + t.Fatal("forwarded create must not also write locally") + } +} + +// Plain creates are upserts whichever filer applies them, so routing them would +// buy nothing and cost a hop. +func TestCreateEntryPlainStaysLocal(t *testing.T) { + fs, store := createRouteServer(t) + name := peerOwnedName(t, fs) + + resp, err := fs.CreateEntry(context.Background(), createReq(name, false)) + if err != nil || resp.Error != "" { + t.Fatalf("plain create must be applied locally, got err=%v resp=%v", err, resp.Error) + } + if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found { + t.Fatal("plain create must reach the local store") + } +} + +// is_moved bounds forwarding to one hop: the owner applies it even if its own +// ring view says someone else owns the key. +func TestCreateEntryMovedAppliesLocally(t *testing.T) { + fs, store := createRouteServer(t) + name := peerOwnedName(t, fs) + + req := createReq(name, true) + req.IsMoved = true + resp, err := fs.CreateEntry(context.Background(), req) + if err != nil || resp.Error != "" { + t.Fatalf("forwarded create must be applied locally, got err=%v resp=%v", err, resp.Error) + } + if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found { + t.Fatal("forwarded create must reach the local store") + } +} + +// An exclusive create this filer owns is applied locally, under its per-path lock. +func TestCreateEntryExclusiveAppliedByOwner(t *testing.T) { + fs, store := createRouteServer(t) + var name string + for i := 0; i < 4000 && name == ""; i++ { + candidate := fmt.Sprintf("obj-%d", i) + key := entryRouteKey(util.NewFullPath("/test", candidate)) + if fs.filer.Dlm.LockRing.GetPrimary(key) == fs.option.Host { + name = candidate + } + } + if name == "" { + t.Skip("no name owned by this filer") + } + + resp, err := fs.CreateEntry(context.Background(), createReq(name, true)) + if err != nil || resp.Error != "" { + t.Fatalf("owner must apply the create, got err=%v resp=%v", err, resp.Error) + } + if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found { + t.Fatal("owner's create must reach the local store") + } +} + +// A key still inside the cooling-off window belongs to its prior owner. If that +// filer cannot be reached the create fails: it may be partitioned rather than +// down, and applying here would be the second serialization point on the key. +func TestCreateEntryFailsWhenPriorOwnerIsUnreachable(t *testing.T) { + const self = pb.ServerAddress("127.0.0.1:28888") + const peer = pb.ServerAddress("127.0.0.1:28889") + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + f.DirBucketsPath = "/buckets" + f.Dlm = lock_manager.NewDistributedLockManager(self) + // Dropping the peer moves every key it owned to this filer, with the peer as + // their prior owner — and nothing is listening on it. + f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self, peer}, 1) + f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self}, 2) + fs := &FilerServer{ + filer: f, + option: &FilerOption{Host: self}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + entryLockTable: util.NewLockTable[util.FullPath](), + } + + var name string + for i := 0; i < 4000 && name == ""; i++ { + candidate := fmt.Sprintf("obj-%d", i) + if fs.filer.Dlm.LockRing.PriorOwner(entryRouteKey(util.NewFullPath("/test", candidate))) == peer { + name = candidate + } + } + if name == "" { + t.Skip("no key whose prior owner is the departed peer") + } + + resp, err := fs.CreateEntry(context.Background(), createReq(name, true)) + if err == nil { + t.Fatalf("unreachable prior owner must fail the create, got resp=%v", resp) + } + if resp == nil { + t.Fatal("a failed create must still answer with a response, not nil") + } + if _, found := store.entries[string(util.NewFullPath("/test", name))]; found { + t.Fatal("a create routed away must not also write locally") + } +} diff --git a/weed/server/filer_grpc_server_route.go b/weed/server/filer_grpc_server_route.go index b52d3844b..d60254bfb 100644 --- a/weed/server/filer_grpc_server_route.go +++ b/weed/server/filer_grpc_server_route.go @@ -5,6 +5,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util" ) // writeOwner returns the filer that serializes writes to key, or "" when this @@ -46,3 +48,10 @@ func (fs *FilerServer) forwardToWriteOwner(ctx context.Context, key string, send } return true, nil } + +// entryRouteKey is the ring key for an entry's writes. It shares the S3 +// gateway's namespace so an object's ObjectTransaction and its CreateEntry +// resolve to the same owner, and land on that filer's one per-path lock. +func entryRouteKey(fullpath util.FullPath) string { + return s3_constants.ObjectWriteRouteKeyPrefix + string(fullpath) +}