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 {