diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index bddc1ae9c..4fb258158 100644 --- a/weed/operation/assign_file_id.go +++ b/weed/operation/assign_file_id.go @@ -193,12 +193,20 @@ type StorageOption struct { MaxFileNameLength uint32 Fsync bool SaveInside bool + // EnforceLocation marks a location that came from a placement rule or an + // explicit request rather than a filer-level default: it must fail the + // assign when unsatisfiable instead of widening to anywhere. + EnforceLocation bool } func (so *StorageOption) TtlString() string { return needle.SecondsToTTL(so.TtlSeconds) } +// ToAssignRequests builds the assign request for this storage option. When the +// location came from a filer-level default rather than a placement rule or an +// explicit request, it also returns a location-stripped fallback for the case +// where the preferred location cannot serve the write at all. func (so *StorageOption) ToAssignRequests(count int) (ar *VolumeAssignRequest, altRequest *VolumeAssignRequest) { ar = &VolumeAssignRequest{ Count: uint64(count), @@ -211,16 +219,13 @@ func (so *StorageOption) ToAssignRequests(count int) (ar *VolumeAssignRequest, a DataNode: so.DataNode, WritableVolumeCount: so.VolumeGrowthCount, } - if so.DataCenter != "" || so.Rack != "" || so.DataNode != "" { + if !so.EnforceLocation && (so.DataCenter != "" || so.Rack != "" || so.DataNode != "") { altRequest = &VolumeAssignRequest{ Count: uint64(count), Replication: so.Replication, Collection: so.Collection, Ttl: so.TtlString(), DiskType: so.DiskType, - DataCenter: "", - Rack: "", - DataNode: "", WritableVolumeCount: so.VolumeGrowthCount, } } diff --git a/weed/operation/assign_file_id_test.go b/weed/operation/assign_file_id_test.go new file mode 100644 index 000000000..e14327d8e --- /dev/null +++ b/weed/operation/assign_file_id_test.go @@ -0,0 +1,42 @@ +package operation + +import "testing" + +// An enforced location on a storage option is a placement contract: the assign +// request must carry it, and no widened fallback may follow it. +func TestToAssignRequestsKeepsEnforcedLocation(t *testing.T) { + so := &StorageOption{ + Collection: "dump", + Rack: "l1000", + DataCenter: "dc1", + EnforceLocation: true, + } + ar, alt := so.ToAssignRequests(1) + if ar.DataCenter != "dc1" || ar.Rack != "l1000" { + t.Fatalf("expected the rule's location, got dc=%q rack=%q", ar.DataCenter, ar.Rack) + } + if alt != nil { + t.Fatalf("enforced location must not be widened, got alt dc=%q rack=%q", alt.DataCenter, alt.Rack) + } +} + +// A filer-level default location is a preference: the request still carries +// it, and a location-stripped fallback follows so a preferred but unservable +// location does not fail the write. +func TestToAssignRequestsWidensDefaultLocation(t *testing.T) { + so := &StorageOption{ + Collection: "dump", + Rack: "l1000", + DataCenter: "dc1", + } + ar, alt := so.ToAssignRequests(1) + if ar.DataCenter != "dc1" || ar.Rack != "l1000" { + t.Fatalf("expected the preferred location, got dc=%q rack=%q", ar.DataCenter, ar.Rack) + } + if alt == nil || alt.DataCenter != "" || alt.Rack != "" { + t.Fatalf("default location should keep a widened fallback, got %+v", alt) + } + if alt.Collection != "dump" || alt.Replication != ar.Replication { + t.Fatalf("fallback must preserve the rest of the option, got %+v", alt) + } +} diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index 387024a7d..9db11ec09 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -772,11 +772,11 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, var readPersistedLogErr error var readInMemoryLogErr error var isDone bool - var lastCheckedFlushTsNs int64 = -1 // Track the last local flush we read the disk under - var lastCheckedFlushLowTsNs int64 = -1 // Track the last peer flush low-watermark we read the disk under - var lastDiskReadTsNs int64 = -1 // Track the last read position we used for disk read - var lastDiskRefsStopTsNs int64 = -1 // Track the last chunk listing bound we read the disk under - var lastDiskPassAt time.Time // Paces re-probes for files no watermark signals + var lastCheckedFlushTsNs int64 = -1 // Track the last local flush we read the disk under + var lastCheckedFlushLowTsNs int64 = -1 // Track the last peer flush low-watermark we read the disk under + var lastDiskReadTsNs int64 = -1 // Track the last read position we used for disk read + var lastDiskRefsStopTsNs int64 = -1 // Track the last chunk listing bound we read the disk under + var lastDiskPassAt time.Time // Paces re-probes for files no watermark signals sentRefs := make(map[string]sentRefState) aggBuffer := fs.filer.MetaAggregator.MetaLogBuffer diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go index 4bb91c9d0..0864409c3 100644 --- a/weed/server/filer_server_handlers_write.go +++ b/weed/server/filer_server_handlers_write.go @@ -335,6 +335,11 @@ func (fs *FilerServer) detectStorageOption(ctx context.Context, requestURI, qCol Fsync: rule.Fsync, VolumeGrowthCount: rule.VolumeGrowthCount, MaxFileNameLength: maxFileNameLength, + // A location that survives from a request parameter, a placement + // overlay, or the path rule is a contract; only a bare filer default + // is a preference that may still be widened. + EnforceLocation: dataCenter != "" || overlayDataCenter != "" || rule.DataCenter != "" || + rack != "" || rule.Rack != "" || dataNode != "" || rule.DataNode != "", }, nil }