From fee3fcb55a61fa082b1da3e662c9e02fd46ea85e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 27 Jul 2026 14:28:29 -0700 Subject: [PATCH] mount: report data sizes to df with -df.logical (#10459) df on a mount shows the space the cluster gives up to the data: every replica of a regular volume, every shard of an ec one. That is the honest answer for capacity planning, but it is not the question a user asks when they want to know how much of their data is stored. Add -df.logical. The master reports the logical sizes alongside the raw ones: one replica per regular volume, the data shards of each ec volume counted once. Free space is divided by the copies the requested replication makes, so used plus available stays the amount of data the mount can still write, and it comes off the cluster-wide usage rather than one collection's, since capacity is cluster-wide too. Statistics through a filer resolves an unset replication to the filer's default rather than the master's, matching where the writes it is sizing for actually land. The flag governs the quota check too, so a mount has one notion of how much it is using. A filer that predates the new fields sends zeros, and the mount keeps reporting the raw sizes. --- other/java/client/src/main/proto/filer.proto | 3 + weed/command/mount.go | 2 + weed/command/mount_std.go | 1 + weed/mount/weedfs.go | 4 + weed/mount/weedfs_quota.go | 17 ++- weed/mount/weedfs_stats.go | 18 ++- weed/mount/weedfs_stats_test.go | 31 +++++ weed/pb/filer.proto | 3 + weed/pb/filer_pb/filer.pb.go | 35 +++-- weed/pb/filer_pb/filer_vtproto.pb.go | 54 ++++++++ weed/pb/master.proto | 5 + weed/pb/master_pb/master.pb.go | 37 ++++-- weed/server/filer_grpc_server_admin.go | 21 ++- .../filer_grpc_server_statistics_test.go | 20 +++ .../master_grpc_server_statistics_test.go | 124 ++++++++++++++++++ weed/server/master_grpc_server_volume.go | 42 +++++- weed/storage/erasure_coding/ec_volume_info.go | 9 ++ weed/topology/topology.go | 2 + weed/topology/topology_ec.go | 22 +++- weed/topology/topology_stats_test.go | 16 ++- weed/topology/volume_layout.go | 6 +- 21 files changed, 429 insertions(+), 43 deletions(-) create mode 100644 weed/mount/weedfs_stats_test.go create mode 100644 weed/server/filer_grpc_server_statistics_test.go create mode 100644 weed/server/master_grpc_server_statistics_test.go diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 359f9f819..1eb76c7b9 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -604,6 +604,9 @@ message StatisticsResponse { uint64 total_size = 4; uint64 used_size = 5; uint64 file_count = 6; + // sizes counting one copy of the data, as reported by the master + uint64 logical_total_size = 7; + uint64 logical_used_size = 8; } message PingRequest { diff --git a/weed/command/mount.go b/weed/command/mount.go index d400deeeb..5ae83c481 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -12,6 +12,7 @@ type MountOptions struct { dirAutoCreate *bool collection *string collectionQuota *int + logicalDiskUsage *bool replication *string diskType *string ttlSec *int @@ -99,6 +100,7 @@ func init() { mountOptions.dirAutoCreate = cmdMount.Flag.Bool("dirAutoCreate", false, "auto create the directory to mount to") mountOptions.collection = cmdMount.Flag.String("collection", "", "collection to create the files") mountOptions.collectionQuota = cmdMount.Flag.Int("collectionQuotaMB", 0, "quota for the collection") + mountOptions.logicalDiskUsage = cmdMount.Flag.Bool("df.logical", false, "report data sizes to df and the quota, instead of the space they occupy with replicas and ec parity") mountOptions.replication = cmdMount.Flag.String("replication", "", "replication(e.g. 000, 001) to create to files. If empty, let filer decide.") mountOptions.diskType = cmdMount.Flag.String("disk", "", "[hdd|ssd|] hard drive or solid state drive or any tag") mountOptions.ttlSec = cmdMount.Flag.Int("ttl", 0, "file ttl in seconds") diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go index 7a3fb21c2..b7a981421 100644 --- a/weed/command/mount_std.go +++ b/weed/command/mount_std.go @@ -349,6 +349,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { CacheMetaTTlSec: *option.cacheMetaTtlSec, DataCenter: *option.dataCenter, Quota: int64(*option.collectionQuota) * 1024 * 1024, + LogicalDiskUsage: *option.logicalDiskUsage, MountUid: uid, MountGid: gid, MountMode: mountMode, diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 4b10fabaa..0dedcae49 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -61,6 +61,10 @@ type Option struct { DisableXAttr bool IsMacOs bool + // LogicalDiskUsage reports data sizes rather than the space they occupy, + // for both df and the quota. See WFS.diskSizes. + LogicalDiskUsage bool + MountUid uint32 MountGid uint32 MountMode os.FileMode diff --git a/weed/mount/weedfs_quota.go b/weed/mount/weedfs_quota.go index df35fa61e..c09ddbaf2 100644 --- a/weed/mount/weedfs_quota.go +++ b/weed/mount/weedfs_quota.go @@ -64,8 +64,8 @@ func (wfs *WFS) IsOverQuotaWithUncommitted() bool { } // Check if uncommitted writes would exceed quota uncommitted := atomic.LoadInt64(&uncommittedBytes) - usedSize := int64(wfs.stats.UsedSize) - return (usedSize + uncommitted) > wfs.option.Quota + _, usedSize := wfs.diskSizes() + return (int64(usedSize) + uncommitted) > wfs.option.Quota } func (wfs *WFS) loopCheckQuota() { @@ -98,9 +98,9 @@ func (wfs *WFS) getQuotaCheckInterval() time.Duration { return defaultQuotaCheckInterval } - usedSize := int64(wfs.stats.UsedSize) + _, usedSize := wfs.diskSizes() uncommitted := atomic.LoadInt64(&uncommittedBytes) - totalUsed := usedSize + uncommitted + totalUsed := int64(usedSize) + uncommitted // If we're at 90% or more of quota, check more frequently if float64(totalUsed) >= float64(wfs.option.Quota)*quotaWarningThreshold { @@ -129,15 +129,18 @@ func (wfs *WFS) checkQuotaOnce() { // Update the stats cache with latest filer data wfs.stats.UsedSize = resp.UsedSize wfs.stats.TotalSize = resp.TotalSize + wfs.stats.LogicalUsedSize = resp.LogicalUsedSize + wfs.stats.LogicalTotalSize = resp.LogicalTotalSize // Reset uncommitted counter since we now have fresh data from filer wfs.ResetUncommittedBytes() - isOverQuota := int64(resp.UsedSize) > wfs.option.Quota + _, usedSize := wfs.diskSizes() + isOverQuota := int64(usedSize) > wfs.option.Quota if isOverQuota && !wfs.IsOverQuota { - glog.Warningf("Quota Exceeded! quota:%d used:%d", wfs.option.Quota, resp.UsedSize) + glog.Warningf("Quota Exceeded! quota:%d used:%d", wfs.option.Quota, usedSize) } else if !isOverQuota && wfs.IsOverQuota { - glog.Warningf("Within quota limit! quota:%d used:%d", wfs.option.Quota, resp.UsedSize) + glog.Warningf("Within quota limit! quota:%d used:%d", wfs.option.Quota, usedSize) } wfs.IsOverQuota = isOverQuota diff --git a/weed/mount/weedfs_stats.go b/weed/mount/weedfs_stats.go index 767cf2503..152e0b1e8 100644 --- a/weed/mount/weedfs_stats.go +++ b/weed/mount/weedfs_stats.go @@ -18,6 +18,19 @@ type statsCache struct { lastChecked int64 // unix time in seconds } +// diskSizes reports the sizes df and the quota work from. By default that is +// the space the cluster gives up to the data, counting every replica and every +// EC shard. Under -df.logical it is the data itself, with the free space +// converted to how much more of it the mount's replication setting allows. +func (wfs *WFS) diskSizes() (totalSize, usedSize uint64) { + // a filer older than the logical sizes sends zeros, so keep reporting the + // raw ones rather than an empty filesystem + if wfs.option.LogicalDiskUsage && wfs.stats.LogicalTotalSize > 0 { + return wfs.stats.LogicalTotalSize, wfs.stats.LogicalUsedSize + } + return wfs.stats.TotalSize, wfs.stats.UsedSize +} + func (wfs *WFS) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.StatfsOut) (code fuse.Status) { // glog.V(4).Infof("reading fs stats") @@ -43,6 +56,8 @@ func (wfs *WFS) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.Stat wfs.stats.TotalSize = resp.TotalSize wfs.stats.UsedSize = resp.UsedSize + wfs.stats.LogicalTotalSize = resp.LogicalTotalSize + wfs.stats.LogicalUsedSize = resp.LogicalUsedSize wfs.stats.FileCount = resp.FileCount wfs.stats.lastChecked = time.Now().Unix() @@ -54,8 +69,7 @@ func (wfs *WFS) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.Stat } } - totalDiskSize := wfs.stats.TotalSize - usedDiskSize := wfs.stats.UsedSize + totalDiskSize, usedDiskSize := wfs.diskSizes() actualFileCount := wfs.stats.FileCount if wfs.option.Quota > 0 && totalDiskSize > uint64(wfs.option.Quota) { diff --git a/weed/mount/weedfs_stats_test.go b/weed/mount/weedfs_stats_test.go new file mode 100644 index 000000000..6b00ec8ae --- /dev/null +++ b/weed/mount/weedfs_stats_test.go @@ -0,0 +1,31 @@ +package mount + +import "testing" + +func TestDiskSizes(t *testing.T) { + wfs := &WFS{option: &Option{}} + wfs.stats.TotalSize = 3000 + wfs.stats.UsedSize = 2000 + wfs.stats.LogicalTotalSize = 1500 + wfs.stats.LogicalUsedSize = 1000 + + total, used := wfs.diskSizes() + if total != 3000 || used != 2000 { + t.Errorf("raw sizes: got %d/%d, want 3000/2000", total, used) + } + + wfs.option.LogicalDiskUsage = true + total, used = wfs.diskSizes() + if total != 1500 || used != 1000 { + t.Errorf("logical sizes: got %d/%d, want 1500/1000", total, used) + } + + // a filer that does not report logical sizes must not read as an empty + // filesystem + wfs.stats.LogicalTotalSize = 0 + wfs.stats.LogicalUsedSize = 0 + total, used = wfs.diskSizes() + if total != 3000 || used != 2000 { + t.Errorf("sizes from a filer without logical support: got %d/%d, want 3000/2000", total, used) + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 359f9f819..1eb76c7b9 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -604,6 +604,9 @@ message StatisticsResponse { uint64 total_size = 4; uint64 used_size = 5; uint64 file_count = 6; + // sizes counting one copy of the data, as reported by the master + uint64 logical_total_size = 7; + uint64 logical_used_size = 8; } message PingRequest { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index ec4dcc375..d89004db6 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -3807,12 +3807,15 @@ func (x *StatisticsRequest) GetDiskType() string { } type StatisticsResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - TotalSize uint64 `protobuf:"varint,4,opt,name=total_size,json=totalSize,proto3" json:"total_size,omitempty"` - UsedSize uint64 `protobuf:"varint,5,opt,name=used_size,json=usedSize,proto3" json:"used_size,omitempty"` - FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + TotalSize uint64 `protobuf:"varint,4,opt,name=total_size,json=totalSize,proto3" json:"total_size,omitempty"` + UsedSize uint64 `protobuf:"varint,5,opt,name=used_size,json=usedSize,proto3" json:"used_size,omitempty"` + FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"` + // sizes counting one copy of the data, as reported by the master + LogicalTotalSize uint64 `protobuf:"varint,7,opt,name=logical_total_size,json=logicalTotalSize,proto3" json:"logical_total_size,omitempty"` + LogicalUsedSize uint64 `protobuf:"varint,8,opt,name=logical_used_size,json=logicalUsedSize,proto3" json:"logical_used_size,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *StatisticsResponse) Reset() { @@ -3866,6 +3869,20 @@ func (x *StatisticsResponse) GetFileCount() uint64 { return 0 } +func (x *StatisticsResponse) GetLogicalTotalSize() uint64 { + if x != nil { + return x.LogicalTotalSize + } + return 0 +} + +func (x *StatisticsResponse) GetLogicalUsedSize() uint64 { + if x != nil { + return x.LogicalUsedSize + } + return 0 +} + type PingRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Target string `protobuf:"bytes,1,opt,name=target,proto3" json:"target,omitempty"` // default to ping itself @@ -7269,13 +7286,15 @@ const file_filer_proto_rawDesc = "" + "collection\x18\x02 \x01(\tR\n" + "collection\x12\x10\n" + "\x03ttl\x18\x03 \x01(\tR\x03ttl\x12\x1b\n" + - "\tdisk_type\x18\x04 \x01(\tR\bdiskType\"o\n" + + "\tdisk_type\x18\x04 \x01(\tR\bdiskType\"\xc9\x01\n" + "\x12StatisticsResponse\x12\x1d\n" + "\n" + "total_size\x18\x04 \x01(\x04R\ttotalSize\x12\x1b\n" + "\tused_size\x18\x05 \x01(\x04R\busedSize\x12\x1d\n" + "\n" + - "file_count\x18\x06 \x01(\x04R\tfileCount\"F\n" + + "file_count\x18\x06 \x01(\x04R\tfileCount\x12,\n" + + "\x12logical_total_size\x18\a \x01(\x04R\x10logicalTotalSize\x12*\n" + + "\x11logical_used_size\x18\b \x01(\x04R\x0flogicalUsedSize\"F\n" + "\vPingRequest\x12\x16\n" + "\x06target\x18\x01 \x01(\tR\x06target\x12\x1f\n" + "\vtarget_type\x18\x02 \x01(\tR\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 93995c539..26272e1b0 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -3502,6 +3502,16 @@ func (m *StatisticsResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.LogicalUsedSize != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.LogicalUsedSize)) + i-- + dAtA[i] = 0x40 + } + if m.LogicalTotalSize != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.LogicalTotalSize)) + i-- + dAtA[i] = 0x38 + } if m.FileCount != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.FileCount)) i-- @@ -7591,6 +7601,12 @@ func (m *StatisticsResponse) SizeVT() (n int) { if m.FileCount != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.FileCount)) } + if m.LogicalTotalSize != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.LogicalTotalSize)) + } + if m.LogicalUsedSize != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.LogicalUsedSize)) + } n += len(m.unknownFields) return n } @@ -18140,6 +18156,44 @@ func (m *StatisticsResponse) UnmarshalVT(dAtA []byte) error { break } } + case 7: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LogicalTotalSize", wireType) + } + m.LogicalTotalSize = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.LogicalTotalSize |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 8: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LogicalUsedSize", wireType) + } + m.LogicalUsedSize = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.LogicalUsedSize |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/pb/master.proto b/weed/pb/master.proto index 0fbfe198a..e53d6676a 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -294,6 +294,11 @@ message StatisticsResponse { uint64 total_size = 4; uint64 used_size = 5; uint64 file_count = 6; + // sizes counting one copy of the data: a single replica of a regular volume, + // the data shards of an ec volume. logical_total_size scales the free space + // by the copies the requested replication makes. + uint64 logical_total_size = 7; + uint64 logical_used_size = 8; } // diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 955d2397f..56bea92a4 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/weed/pb/master_pb/master.pb.go @@ -1864,12 +1864,17 @@ func (x *StatisticsRequest) GetDiskType() string { } type StatisticsResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - TotalSize uint64 `protobuf:"varint,4,opt,name=total_size,json=totalSize,proto3" json:"total_size,omitempty"` - UsedSize uint64 `protobuf:"varint,5,opt,name=used_size,json=usedSize,proto3" json:"used_size,omitempty"` - FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + TotalSize uint64 `protobuf:"varint,4,opt,name=total_size,json=totalSize,proto3" json:"total_size,omitempty"` + UsedSize uint64 `protobuf:"varint,5,opt,name=used_size,json=usedSize,proto3" json:"used_size,omitempty"` + FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"` + // sizes counting one copy of the data: a single replica of a regular volume, + // the data shards of an ec volume. logical_total_size scales the free space + // by the copies the requested replication makes. + LogicalTotalSize uint64 `protobuf:"varint,7,opt,name=logical_total_size,json=logicalTotalSize,proto3" json:"logical_total_size,omitempty"` + LogicalUsedSize uint64 `protobuf:"varint,8,opt,name=logical_used_size,json=logicalUsedSize,proto3" json:"logical_used_size,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *StatisticsResponse) Reset() { @@ -1923,6 +1928,20 @@ func (x *StatisticsResponse) GetFileCount() uint64 { return 0 } +func (x *StatisticsResponse) GetLogicalTotalSize() uint64 { + if x != nil { + return x.LogicalTotalSize + } + return 0 +} + +func (x *StatisticsResponse) GetLogicalUsedSize() uint64 { + if x != nil { + return x.LogicalUsedSize + } + return 0 +} + // collection related type Collection struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4763,13 +4782,15 @@ const file_master_proto_rawDesc = "" + "collection\x18\x02 \x01(\tR\n" + "collection\x12\x10\n" + "\x03ttl\x18\x03 \x01(\tR\x03ttl\x12\x1b\n" + - "\tdisk_type\x18\x04 \x01(\tR\bdiskType\"o\n" + + "\tdisk_type\x18\x04 \x01(\tR\bdiskType\"\xc9\x01\n" + "\x12StatisticsResponse\x12\x1d\n" + "\n" + "total_size\x18\x04 \x01(\x04R\ttotalSize\x12\x1b\n" + "\tused_size\x18\x05 \x01(\x04R\busedSize\x12\x1d\n" + "\n" + - "file_count\x18\x06 \x01(\x04R\tfileCount\" \n" + + "file_count\x18\x06 \x01(\x04R\tfileCount\x12,\n" + + "\x12logical_total_size\x18\a \x01(\x04R\x10logicalTotalSize\x12*\n" + + "\x11logical_used_size\x18\b \x01(\x04R\x0flogicalUsedSize\" \n" + "\n" + "Collection\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\"{\n" + diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go index 9c9b86c31..0cef9e0c4 100644 --- a/weed/server/filer_grpc_server_admin.go +++ b/weed/server/filer_grpc_server_admin.go @@ -23,7 +23,7 @@ func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsR err = fs.filer.MasterClient.WithClient(false, func(masterClient master_pb.SeaweedClient) error { grpcResponse, grpcErr := masterClient.Statistics(context.Background(), &master_pb.StatisticsRequest{ - Replication: req.Replication, + Replication: fs.statisticsReplication(req.Replication), Collection: req.Collection, Ttl: req.Ttl, DiskType: req.DiskType, @@ -41,12 +41,25 @@ func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsR } return &filer_pb.StatisticsResponse{ - TotalSize: output.TotalSize, - UsedSize: output.UsedSize, - FileCount: output.FileCount, + TotalSize: output.TotalSize, + UsedSize: output.UsedSize, + FileCount: output.FileCount, + LogicalTotalSize: output.LogicalTotalSize, + LogicalUsedSize: output.LogicalUsedSize, }, nil } +// statisticsReplication names the replication the caller's writes will use, so +// the master sizes free space by the right number of copies. Writes through +// this filer follow its default, not the master's, so fill that in when the +// request leaves the choice open. +func (fs *FilerServer) statisticsReplication(requested string) string { + if requested != "" { + return requested + } + return fs.option.DefaultReplication +} + // isKnownPingTarget reports whether target is a peer the filer has learned // about from its master subscription (other filers, volume servers) or from // its own master list. Restricting Ping prevents the RPC from being used as diff --git a/weed/server/filer_grpc_server_statistics_test.go b/weed/server/filer_grpc_server_statistics_test.go new file mode 100644 index 000000000..3eb0a447d --- /dev/null +++ b/weed/server/filer_grpc_server_statistics_test.go @@ -0,0 +1,20 @@ +package weed_server + +import "testing" + +func TestStatisticsReplication(t *testing.T) { + fs := &FilerServer{option: &FilerOption{DefaultReplication: "001"}} + + if got := fs.statisticsReplication("020"); got != "020" { + t.Errorf("requested replication: got %q, want %q", got, "020") + } + if got := fs.statisticsReplication(""); got != "001" { + t.Errorf("empty replication: got %q, want the filer default %q", got, "001") + } + + // a filer without its own default leaves the choice to the master + fs.option.DefaultReplication = "" + if got := fs.statisticsReplication(""); got != "" { + t.Errorf("empty replication without a filer default: got %q, want %q", got, "") + } +} diff --git a/weed/server/master_grpc_server_statistics_test.go b/weed/server/master_grpc_server_statistics_test.go new file mode 100644 index 000000000..6a8ba1daf --- /dev/null +++ b/weed/server/master_grpc_server_statistics_test.go @@ -0,0 +1,124 @@ +package weed_server + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +// ecShardSizes returns count shard sizes of size bytes each. +func ecShardSizes(count int, size int64) []int64 { + sizes := make([]int64, count) + for i := range sizes { + sizes[i] = size + } + return sizes +} + +// newStatisticsMaster returns a leader master over four nodes of 10 volume +// slots at 1MB each. Collection c1 holds a 1000 byte volume replicated to two +// nodes and an EC volume of 14 shards at 100 bytes; collection c2 holds a +// single 5000 byte volume. +func newStatisticsMaster(t *testing.T) *MasterServer { + t.Helper() + + ms := newLeaderMaster() + ms.option.VolumeSizeLimitMB = 1 + + rack := ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1") + maxVolumeCounts := map[string]uint32{"": 10} + newNode := func(host string) *topology.DataNode { + dn := rack.GetOrCreateDataNode(host, 34534, 0, host, "", maxVolumeCounts) + // VolumeLocationList.Stats only counts nodes connected for over a minute + dn.LastSeen = time.Now().Unix() - 61 + return dn + } + + replicatedVolume := &master_pb.VolumeInformationMessage{ + Id: 1, Size: 1000, Collection: "c1", FileCount: 10, + ReplicaPlacement: 1, // 001: one copy in the same rack + Version: uint32(needle.GetCurrentVersion()), + } + for _, host := range []string{"127.0.0.1", "127.0.0.2"} { + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{replicatedVolume}, newNode(host)) + } + + otherCollectionNode := newNode("127.0.0.3") + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + {Id: 3, Size: 5000, Collection: "c2", FileCount: 50, Version: uint32(needle.GetCurrentVersion())}, + }, otherCollectionNode) + + // EC shards are not replicated here, so the volume sits on one node + ms.Topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + {Id: 2, Collection: "c1", EcIndexBits: 1<<14 - 1, ShardSizes: ecShardSizes(14, 100), FileCount: 7}, + }, newNode("127.0.0.4")) + + return ms +} + +func TestStatisticsLogicalSizes(t *testing.T) { + ms := newStatisticsMaster(t) + + resp, err := ms.Statistics(context.Background(), &master_pb.StatisticsRequest{ + Collection: "c1", + Replication: "001", + }) + if err != nil { + t.Fatalf("Statistics: %v", err) + } + + // 1000 bytes on each of two replicas, plus every shard of the EC volume + if got, want := resp.UsedSize, uint64(2*1000+14*100); got != want { + t.Errorf("used size: got %d, want %d", got, want) + } + // one replica, and the 10 data shards of the EC volume + if got, want := resp.LogicalUsedSize, uint64(1000+10*100); got != want { + t.Errorf("logical used size: got %d, want %d", got, want) + } + + // four nodes of 10 slots at 1MB each + totalSize := uint64(40 * 1024 * 1024) + if resp.TotalSize != totalSize { + t.Fatalf("total size: got %d, want %d", resp.TotalSize, totalSize) + } + // what is left over is what c2 has not taken either, and 001 writes two + // copies of it + clusterUsedSize := resp.UsedSize + 5000 + want := resp.LogicalUsedSize + (totalSize-clusterUsedSize)/2 + if resp.LogicalTotalSize != want { + t.Errorf("logical total size: got %d, want %d", resp.LogicalTotalSize, want) + } +} + +func TestStatisticsReplicaCopyCount(t *testing.T) { + ms := newStatisticsMaster(t) + ms.option.DefaultReplicaPlacement = "010" + + for _, tc := range []struct { + name string + replication string + copies uint64 + }{ + {"requested", "002", 3}, + {"empty falls back to the master default", "", 2}, + {"unparsable falls back to the master default", "not-a-replication", 2}, + } { + t.Run(tc.name, func(t *testing.T) { + resp, err := ms.Statistics(context.Background(), &master_pb.StatisticsRequest{ + Replication: tc.replication, + }) + if err != nil { + t.Fatalf("Statistics: %v", err) + } + want := resp.LogicalUsedSize + (resp.TotalSize-resp.UsedSize)/tc.copies + if resp.LogicalTotalSize != want { + t.Errorf("logical total size: got %d, want %d for %d copies", + resp.LogicalTotalSize, want, tc.copies) + } + }) + } +} diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 9a4723439..df34ff1f9 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -205,19 +205,49 @@ func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.Statistic } // an empty collection means all collections, and a named collection covers - // all its layouts and EC volumes, so used size matches the topology-wide - // total size below + // all its layouts and EC volumes stats := ms.Topo.CollectionVolumeStats(req.Collection) - totalSize := ms.Topo.GetDiskUsages().GetMaxVolumeCount() * int64(ms.option.VolumeSizeLimitMB) * 1024 * 1024 + totalSize := uint64(ms.Topo.GetDiskUsages().GetMaxVolumeCount() * int64(ms.option.VolumeSizeLimitMB) * 1024 * 1024) + // capacity is cluster-wide, so what is left over is what every collection + // has not taken, not just the one asked about + clusterUsedSize := stats.UsedSize + if req.Collection != "" { + clusterUsedSize = ms.Topo.CollectionVolumeStats("").UsedSize + } + // and the free space holds that many copies fewer of whatever the caller writes + var freeSize uint64 + if totalSize > clusterUsedSize { + freeSize = (totalSize - clusterUsedSize) / uint64(ms.replicaCopyCount(req.Replication)) + } resp := &master_pb.StatisticsResponse{ - TotalSize: uint64(totalSize), - UsedSize: stats.UsedSize, - FileCount: stats.FileCount, + TotalSize: totalSize, + UsedSize: stats.UsedSize, + FileCount: stats.FileCount, + LogicalTotalSize: stats.LogicalUsedSize + freeSize, + LogicalUsedSize: stats.LogicalUsedSize, } return resp, nil } +// replicaCopyCount returns how many copies the given replication makes, falling +// back to the master default and then to a single copy. Unparsable input is +// reported rather than failing the call: statistics are informational. +func (ms *MasterServer) replicaCopyCount(replication string) int { + for _, s := range []string{replication, ms.option.DefaultReplicaPlacement} { + if s == "" { + continue + } + rp, err := super_block.NewReplicaPlacementFromString(s) + if err != nil { + glog.V(1).Infof("statistics replication %q: %v", s, err) + continue + } + return rp.GetCopyCount() + } + return 1 +} + func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) { if !ms.Topo.IsLeader() { diff --git a/weed/storage/erasure_coding/ec_volume_info.go b/weed/storage/erasure_coding/ec_volume_info.go index 0c6d18714..4d71d9012 100644 --- a/weed/storage/erasure_coding/ec_volume_info.go +++ b/weed/storage/erasure_coding/ec_volume_info.go @@ -18,6 +18,15 @@ type EcVolumeInfo struct { EncodeTsNs int64 // encode-run identity (unix nanos); one value per (volume, disk) } +// DataShardsOrDefault returns how many of this volume's shards hold data; shard +// ids below it are data, the rest are parity. Open-source SeaweedFS always uses +// the fixed 10+4 layout, so this returns DataShardsCount. It is a per-volume +// accessor, like EcShardsVolumeDataShards, so callers stay correct on builds +// that derive the ratio per volume. +func (ecInfo *EcVolumeInfo) DataShardsOrDefault() int { + return DataShardsCount +} + func (ecInfo *EcVolumeInfo) Minus(other *EcVolumeInfo) *EcVolumeInfo { return &EcVolumeInfo{ VolumeId: ecInfo.VolumeId, diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 1b30d5f69..e898ec7dc 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -449,6 +449,7 @@ func (t *Topology) CollectionVolumeStats(collectionName string) *VolumeLayoutSta stats := vl.Stats() ret.TotalSize += stats.TotalSize ret.UsedSize += stats.UsedSize + ret.LogicalUsedSize += stats.LogicalUsedSize ret.FileCount += stats.FileCount } } @@ -457,6 +458,7 @@ func (t *Topology) CollectionVolumeStats(collectionName string) *VolumeLayoutSta ecStats := t.CollectionEcVolumeStats(collectionName) ret.TotalSize += ecStats.TotalSize ret.UsedSize += ecStats.UsedSize + ret.LogicalUsedSize += ecStats.LogicalUsedSize ret.FileCount += ecStats.FileCount return ret } diff --git a/weed/topology/topology_ec.go b/weed/topology/topology_ec.go index 9621a1202..58f302515 100644 --- a/weed/topology/topology_ec.go +++ b/weed/topology/topology_ec.go @@ -172,18 +172,21 @@ func (t *Topology) LookupEcShards(vid needle.VolumeId) (locations *EcShardLocati return } -// ecVolumeCounts accumulates one EC volume's needle counts while they are -// collected from every node reporting its shards. +// ecVolumeCounts accumulates one EC volume's data size and needle counts while +// they are collected from every node reporting its shards. type ecVolumeCounts struct { - fileCount uint64 - deleteCount uint64 + countedShards erasure_coding.ShardBits + dataSize uint64 + fileCount uint64 + deleteCount uint64 } // CollectionEcVolumeStats sums the disk footprint and live needle count of the // EC volumes in one collection, or in every collection when collectionName is // empty. Every shard copy counts, parity included, the way a regular volume's // used size counts every replica; needle counts are per volume, again as a -// regular volume reports them. +// regular volume reports them. The logical size instead counts each volume's +// data shards once, leaving out parity and over-replicated copies. func (t *Topology) CollectionEcVolumeStats(collectionName string) *VolumeLayoutStats { ret := &VolumeLayoutStats{} perVolume := make(map[needle.VolumeId]*ecVolumeCounts) @@ -201,6 +204,14 @@ func (t *Topology) CollectionEcVolumeStats(collectionName string) *VolumeLayoutS counts = &ecVolumeCounts{} perVolume[ecInfo.VolumeId] = counts } + dataShards := ecInfo.DataShardsOrDefault() + for id := range erasure_coding.ShardBits(ecInfo.ShardsInfo.Bitmap()).All() { + if int(id) >= dataShards || counts.countedShards.Has(id) { + continue + } + counts.countedShards = counts.countedShards.Set(id) + counts.dataSize += uint64(ecInfo.ShardsInfo.Size(id)) + } // .ecx and .ecj are both volume-wide files that travel with // the shards, so take the largest count any holder reports // rather than summing: a node still loading .ecx reports 0 @@ -223,6 +234,7 @@ func (t *Topology) CollectionEcVolumeStats(collectionName string) *VolumeLayoutS // an EC volume is sealed, so it offers no room beyond what it holds ret.TotalSize = ret.UsedSize for _, counts := range perVolume { + ret.LogicalUsedSize += counts.dataSize if counts.fileCount > counts.deleteCount { ret.FileCount += counts.fileCount - counts.deleteCount } diff --git a/weed/topology/topology_stats_test.go b/weed/topology/topology_stats_test.go index ffe2cbd8f..e4bee9bd6 100644 --- a/weed/topology/topology_stats_test.go +++ b/weed/topology/topology_stats_test.go @@ -26,11 +26,17 @@ func TestCollectionVolumeStats(t *testing.T) { } topo.SyncDataNodeRegistration(volumeMessages, dn) + // volume 4 is replicated, so a second node holds a copy of it + replicaDn := rack.GetOrCreateDataNode("127.0.0.2", 34534, 0, "127.0.0.2", "", maxVolumeCounts) + topo.SyncDataNodeRegistration(volumeMessages[3:], replicaDn) + // VolumeLocationList.Stats only counts nodes connected for over a minute dn.LastSeen = time.Now().Unix() - 61 + replicaDn.LastSeen = dn.LastSeen allStats := topo.CollectionVolumeStats("") - assert(t, "all collections used size", int(allStats.UsedSize), 10000) + assert(t, "all collections used size", int(allStats.UsedSize), 14000) + assert(t, "all collections logical used size", int(allStats.LogicalUsedSize), 10000) assert(t, "all collections file count", int(allStats.FileCount), 100) c1Stats := topo.CollectionVolumeStats("c1") @@ -38,7 +44,9 @@ func TestCollectionVolumeStats(t *testing.T) { assert(t, "c1 file count", int(c1Stats.FileCount), 50) c2Stats := topo.CollectionVolumeStats("c2") - assert(t, "c2 used size", int(c2Stats.UsedSize), 4000) + // both copies of volume 4 count against the space, only one against the data + assert(t, "c2 used size", int(c2Stats.UsedSize), 8000) + assert(t, "c2 logical used size", int(c2Stats.LogicalUsedSize), 4000) missingStats := topo.CollectionVolumeStats("no-such-collection") assert(t, "missing collection used size", int(missingStats.UsedSize), 0) @@ -87,12 +95,16 @@ func TestCollectionVolumeStatsWithEcVolumes(t *testing.T) { // 100..700 on dn1, plus a second copy of shard 0 and 800..1400 on dn2 c1Stats := topo.CollectionVolumeStats("c1") assert(t, "c1 ec used size", int(c1Stats.UsedSize), 2800+7800) + // data shards 0..9 once each, so the second copy of shard 0 and the + // parity shards 10..13 are left out + assert(t, "c1 ec logical used size", int(c1Stats.LogicalUsedSize), 5500) // the shared journal counts once, not once per holder: 50 - 5 assert(t, "c1 ec file count", int(c1Stats.FileCount), 45) // volume 20 adds all 14 shards, each sized (id+1)*10 allStats := topo.CollectionVolumeStats("") assert(t, "all collections ec used size", int(allStats.UsedSize), 10600+1050) + assert(t, "all collections ec logical used size", int(allStats.LogicalUsedSize), 5500+550) assert(t, "all collections ec file count", int(allStats.FileCount), 45+7) if _, found := topo.FindCollection("c1"); found { diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index cab8510c7..71b2d4a52 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -140,7 +140,10 @@ type VolumeLayout struct { type VolumeLayoutStats struct { TotalSize uint64 UsedSize uint64 - FileCount uint64 + // LogicalUsedSize counts one copy of the data: a single replica of a + // regular volume, the data shards of an EC volume. + LogicalUsedSize uint64 + FileCount uint64 } func NewVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType, volumeSizeLimit uint64, replicationAsMin bool) *VolumeLayout { @@ -1009,6 +1012,7 @@ func (vl *VolumeLayout) Stats() *VolumeLayoutStats { size, fileCount := vll.Stats(vid, freshThreshold) ret.FileCount += uint64(fileCount) ret.UsedSize += size * uint64(vll.Length()) + ret.LogicalUsedSize += size if vl.readonlyVolumes.IsTrue(vid) { ret.TotalSize += size * uint64(vll.Length()) } else {