mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-16 04:06:44 +00:00
s3: take bucket sizes from the master's summary (#10664)
* pb: ask the master what each collection holds Callers tracking usage were sent every volume in the cluster to add up themselves, which is the master's largest single allocation. * topology: summarise what each collection holds One pass over the topology, allocating per collection rather than per volume. Regular volumes count once each for logical totals and once per replica for physical, taken from the lookup index, which is already keyed by volume and so needs no set of seen ids. Ec shards are node-local so their sizes sum, while the file and delete counts describe the volume and resolve once every holder has been seen. Replicas of one volume disagree while a write is landing or a heartbeat is late. Walking a full listing took whichever replica the map iteration reached first, so the answer moved between runs; this takes the largest, which is stable and never reports usage below what some replica already holds. * s3: take bucket sizes from the master's summary The bucket size metrics pulled the whole volume list once a minute and added it up, which cost the master 184.6MB of allocation and 17.8MB on the wire for six numbers per collection. VolumeList over 550k volumes 184.6 MB allocated, 17.8 MB on the wire CollectionStatistics 176 bytes allocated, 47 bytes on the wire The aggregation moves to the master with it, so the cases the removed tests covered are now asserted against it directly. * topology: count the replica holding the most live data Quotas are enforced on size less deletions, and the replica with the biggest raw size can be the one that has deleted the most. Counting it reported a bucket smaller than it is and would leave one writable over its quota, which is the opposite of what picking the largest was meant to guarantee. * topology: cap a volume's deletions at what it holds Live usage is read as a collection's size less its deletions, so a volume reporting more deleted bytes than it has cancels live bytes belonging to other volumes in the same bucket and reports it smaller than it is. Replica selection already floored that volume's own live size at zero; the totals have to agree with it.
This commit is contained in:
Chris Lu
@@ -59,6 +59,8 @@ service Seaweed {
|
||||
}
|
||||
rpc VolumeGrow (VolumeGrowRequest) returns (VolumeGrowResponse) {
|
||||
}
|
||||
rpc CollectionStatistics (CollectionStatisticsRequest) returns (CollectionStatisticsResponse) {
|
||||
}
|
||||
}
|
||||
|
||||
//////////////////////////////////////////////////
|
||||
@@ -331,6 +333,26 @@ message CollectionListResponse {
|
||||
repeated Collection collections = 1;
|
||||
}
|
||||
|
||||
// Summarises what each collection holds, so a caller tracking usage does not
|
||||
// have to be sent every volume in the cluster to add it up itself.
|
||||
message CollectionStatisticsRequest {
|
||||
}
|
||||
message CollectionStatisticsResponse {
|
||||
repeated CollectionStatistics collections = 1;
|
||||
}
|
||||
message CollectionStatistics {
|
||||
string collection = 1;
|
||||
uint64 file_count = 2;
|
||||
uint64 delete_count = 3;
|
||||
uint64 deleted_byte_count = 4;
|
||||
// one copy of the data: a single replica of a regular volume, the data
|
||||
// shards of an ec volume
|
||||
uint64 size = 5;
|
||||
// what is on disk: every replica, and parity shards
|
||||
uint64 physical_size = 6;
|
||||
uint64 volume_count = 7;
|
||||
}
|
||||
|
||||
message CollectionDeleteRequest {
|
||||
string name = 1;
|
||||
}
|
||||
|
||||
@@ -59,6 +59,8 @@ service Seaweed {
|
||||
}
|
||||
rpc VolumeGrow (VolumeGrowRequest) returns (VolumeGrowResponse) {
|
||||
}
|
||||
rpc CollectionStatistics (CollectionStatisticsRequest) returns (CollectionStatisticsResponse) {
|
||||
}
|
||||
}
|
||||
|
||||
//////////////////////////////////////////////////
|
||||
@@ -331,6 +333,26 @@ message CollectionListResponse {
|
||||
repeated Collection collections = 1;
|
||||
}
|
||||
|
||||
// Summarises what each collection holds, so a caller tracking usage does not
|
||||
// have to be sent every volume in the cluster to add it up itself.
|
||||
message CollectionStatisticsRequest {
|
||||
}
|
||||
message CollectionStatisticsResponse {
|
||||
repeated CollectionStatistics collections = 1;
|
||||
}
|
||||
message CollectionStatistics {
|
||||
string collection = 1;
|
||||
uint64 file_count = 2;
|
||||
uint64 delete_count = 3;
|
||||
uint64 deleted_byte_count = 4;
|
||||
// one copy of the data: a single replica of a regular volume, the data
|
||||
// shards of an ec volume
|
||||
uint64 size = 5;
|
||||
// what is on disk: every replica, and parity shards
|
||||
uint64 physical_size = 6;
|
||||
uint64 volume_count = 7;
|
||||
}
|
||||
|
||||
message CollectionDeleteRequest {
|
||||
string name = 1;
|
||||
}
|
||||
|
||||
+474
-276
File diff suppressed because it is too large
Load Diff
@@ -44,6 +44,7 @@ const (
|
||||
Seaweed_RaftRemoveServer_FullMethodName = "/master_pb.Seaweed/RaftRemoveServer"
|
||||
Seaweed_RaftLeadershipTransfer_FullMethodName = "/master_pb.Seaweed/RaftLeadershipTransfer"
|
||||
Seaweed_VolumeGrow_FullMethodName = "/master_pb.Seaweed/VolumeGrow"
|
||||
Seaweed_CollectionStatistics_FullMethodName = "/master_pb.Seaweed/CollectionStatistics"
|
||||
)
|
||||
|
||||
// SeaweedClient is the client API for Seaweed service.
|
||||
@@ -75,6 +76,7 @@ type SeaweedClient interface {
|
||||
RaftRemoveServer(ctx context.Context, in *RaftRemoveServerRequest, opts ...grpc.CallOption) (*RaftRemoveServerResponse, error)
|
||||
RaftLeadershipTransfer(ctx context.Context, in *RaftLeadershipTransferRequest, opts ...grpc.CallOption) (*RaftLeadershipTransferResponse, error)
|
||||
VolumeGrow(ctx context.Context, in *VolumeGrowRequest, opts ...grpc.CallOption) (*VolumeGrowResponse, error)
|
||||
CollectionStatistics(ctx context.Context, in *CollectionStatisticsRequest, opts ...grpc.CallOption) (*CollectionStatisticsResponse, error)
|
||||
}
|
||||
|
||||
type seaweedClient struct {
|
||||
@@ -344,6 +346,16 @@ func (c *seaweedClient) VolumeGrow(ctx context.Context, in *VolumeGrowRequest, o
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *seaweedClient) CollectionStatistics(ctx context.Context, in *CollectionStatisticsRequest, opts ...grpc.CallOption) (*CollectionStatisticsResponse, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(CollectionStatisticsResponse)
|
||||
err := c.cc.Invoke(ctx, Seaweed_CollectionStatistics_FullMethodName, in, out, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SeaweedServer is the server API for Seaweed service.
|
||||
// All implementations must embed UnimplementedSeaweedServer
|
||||
// for forward compatibility.
|
||||
@@ -373,6 +385,7 @@ type SeaweedServer interface {
|
||||
RaftRemoveServer(context.Context, *RaftRemoveServerRequest) (*RaftRemoveServerResponse, error)
|
||||
RaftLeadershipTransfer(context.Context, *RaftLeadershipTransferRequest) (*RaftLeadershipTransferResponse, error)
|
||||
VolumeGrow(context.Context, *VolumeGrowRequest) (*VolumeGrowResponse, error)
|
||||
CollectionStatistics(context.Context, *CollectionStatisticsRequest) (*CollectionStatisticsResponse, error)
|
||||
mustEmbedUnimplementedSeaweedServer()
|
||||
}
|
||||
|
||||
@@ -458,6 +471,9 @@ func (UnimplementedSeaweedServer) RaftLeadershipTransfer(context.Context, *RaftL
|
||||
func (UnimplementedSeaweedServer) VolumeGrow(context.Context, *VolumeGrowRequest) (*VolumeGrowResponse, error) {
|
||||
return nil, status.Error(codes.Unimplemented, "method VolumeGrow not implemented")
|
||||
}
|
||||
func (UnimplementedSeaweedServer) CollectionStatistics(context.Context, *CollectionStatisticsRequest) (*CollectionStatisticsResponse, error) {
|
||||
return nil, status.Error(codes.Unimplemented, "method CollectionStatistics not implemented")
|
||||
}
|
||||
func (UnimplementedSeaweedServer) mustEmbedUnimplementedSeaweedServer() {}
|
||||
func (UnimplementedSeaweedServer) testEmbeddedByValue() {}
|
||||
|
||||
@@ -896,6 +912,24 @@ func _Seaweed_VolumeGrow_Handler(srv interface{}, ctx context.Context, dec func(
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _Seaweed_CollectionStatistics_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(CollectionStatisticsRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if interceptor == nil {
|
||||
return srv.(SeaweedServer).CollectionStatistics(ctx, in)
|
||||
}
|
||||
info := &grpc.UnaryServerInfo{
|
||||
Server: srv,
|
||||
FullMethod: Seaweed_CollectionStatistics_FullMethodName,
|
||||
}
|
||||
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||
return srv.(SeaweedServer).CollectionStatistics(ctx, req.(*CollectionStatisticsRequest))
|
||||
}
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
// Seaweed_ServiceDesc is the grpc.ServiceDesc for Seaweed service.
|
||||
// It's only intended for direct use with grpc.RegisterService,
|
||||
// and not to be introspected or modified (even as a copy)
|
||||
@@ -991,6 +1025,10 @@ var Seaweed_ServiceDesc = grpc.ServiceDesc{
|
||||
MethodName: "VolumeGrow",
|
||||
Handler: _Seaweed_VolumeGrow_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "CollectionStatistics",
|
||||
Handler: _Seaweed_CollectionStatistics_Handler,
|
||||
},
|
||||
},
|
||||
Streams: []grpc.StreamDesc{
|
||||
{
|
||||
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/stats"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -46,12 +45,6 @@ func (c *CollectionInfo) LogicalSize() float64 {
|
||||
return c.Size - c.DeletedByteCount
|
||||
}
|
||||
|
||||
// volumeKey uniquely identifies a volume for deduplication
|
||||
type volumeKey struct {
|
||||
collection string
|
||||
volumeId uint32
|
||||
}
|
||||
|
||||
// startBucketSizeMetricsLoop periodically collects bucket size metrics and updates Prometheus gauges.
|
||||
// Uses a distributed lock to ensure only one S3 instance collects metrics at a time.
|
||||
// Should be called as a goroutine; stops when the provided context is cancelled.
|
||||
@@ -185,18 +178,28 @@ func (s3a *S3ApiServer) collectCollectionInfoFromMaster(ctx context.Context) (ma
|
||||
masterMap[string(master)] = master
|
||||
}
|
||||
|
||||
// Connect to any available master and get volume list with topology
|
||||
// Ask the master to summarise. Adding this up here instead would mean
|
||||
// being sent every volume in the cluster once a minute.
|
||||
collectionInfos := make(map[string]*CollectionInfo)
|
||||
|
||||
err := pb.WithOneOfGrpcMasterClients(false, masterMap, s3a.option.GrpcDialOption, func(client master_pb.SeaweedClient) error {
|
||||
resp, err := client.VolumeList(ctx, &master_pb.VolumeListRequest{})
|
||||
resp, err := client.CollectionStatistics(ctx, &master_pb.CollectionStatisticsRequest{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get volume list: %w", err)
|
||||
return fmt.Errorf("failed to get collection statistics: %w", err)
|
||||
}
|
||||
if resp == nil || resp.TopologyInfo == nil {
|
||||
return fmt.Errorf("empty topology info from master")
|
||||
if resp == nil {
|
||||
return fmt.Errorf("empty collection statistics from master")
|
||||
}
|
||||
for _, c := range resp.Collections {
|
||||
collectionInfos[c.Collection] = &CollectionInfo{
|
||||
FileCount: float64(c.FileCount),
|
||||
DeleteCount: float64(c.DeleteCount),
|
||||
DeletedByteCount: float64(c.DeletedByteCount),
|
||||
Size: float64(c.Size),
|
||||
PhysicalSize: float64(c.PhysicalSize),
|
||||
VolumeCount: int(c.VolumeCount),
|
||||
}
|
||||
}
|
||||
collectCollectionInfoFromTopology(resp.TopologyInfo, collectionInfos)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
@@ -256,105 +259,3 @@ func (s3a *S3ApiServer) listBuckets(ctx context.Context) ([]*filer_pb.Entry, err
|
||||
|
||||
return buckets, err
|
||||
}
|
||||
|
||||
// ecVolumeAgg accumulates per-volume EC counts across the shard holders.
|
||||
// fileCount is volume-wide (every holder sees the same .ecx) so we take the
|
||||
// max across reporters to avoid a slow node with a not-yet-loaded .ecx
|
||||
// pinning the aggregate at 0. deleteCount is node-local to each .ecj
|
||||
// deletion journal, so it's summed across reporters.
|
||||
type ecVolumeAgg struct {
|
||||
collection string
|
||||
fileCount uint64
|
||||
deleteCount uint64
|
||||
}
|
||||
|
||||
// collectCollectionInfoFromTopology extracts collection info from topology.
|
||||
// Deduplicates by volume ID to correctly handle missing replicas.
|
||||
// Unlike dividing by copyCount (which would give wrong results if replicas are missing),
|
||||
// we track seen volume IDs and only count each volume once for logical size/count.
|
||||
// EC-encoded volumes are folded in via per-shard aggregation: every shard is
|
||||
// node-local (not a replica), so shard sizes are summed across nodes; the
|
||||
// per-volume file/delete counts carried on each shard message are deduped
|
||||
// via max/sum so the aggregate doesn't double-count or drop after a volume
|
||||
// is converted from regular to erasure coding.
|
||||
func collectCollectionInfoFromTopology(t *master_pb.TopologyInfo, collectionInfos map[string]*CollectionInfo) {
|
||||
// Track which volumes we've already seen to deduplicate by volume ID
|
||||
seenVolumes := make(map[volumeKey]bool)
|
||||
ecVolumes := make(map[volumeKey]*ecVolumeAgg)
|
||||
|
||||
for _, dc := range t.DataCenterInfos {
|
||||
for _, r := range dc.RackInfos {
|
||||
for _, dn := range r.DataNodeInfos {
|
||||
for _, diskInfo := range dn.DiskInfos {
|
||||
for _, vi := range diskInfo.VolumeInfos {
|
||||
c := vi.Collection
|
||||
cif, found := collectionInfos[c]
|
||||
if !found {
|
||||
cif = &CollectionInfo{}
|
||||
collectionInfos[c] = cif
|
||||
}
|
||||
|
||||
// Always add to physical size (all replicas)
|
||||
cif.PhysicalSize += float64(vi.Size)
|
||||
|
||||
// Check if we've already counted this volume for logical stats
|
||||
key := volumeKey{collection: c, volumeId: vi.Id}
|
||||
if seenVolumes[key] {
|
||||
// Already counted this volume, skip logical stats
|
||||
continue
|
||||
}
|
||||
seenVolumes[key] = true
|
||||
|
||||
// First time seeing this volume - add to logical stats
|
||||
cif.Size += float64(vi.Size)
|
||||
cif.FileCount += float64(vi.FileCount)
|
||||
cif.DeleteCount += float64(vi.DeleteCount)
|
||||
cif.DeletedByteCount += float64(vi.DeletedByteCount)
|
||||
cif.VolumeCount++
|
||||
}
|
||||
|
||||
for _, esi := range diskInfo.EcShardInfos {
|
||||
c := esi.Collection
|
||||
cif, found := collectionInfos[c]
|
||||
if !found {
|
||||
cif = &CollectionInfo{}
|
||||
collectionInfos[c] = cif
|
||||
}
|
||||
|
||||
// EC shards are node-local (no replication), so both
|
||||
// physical and logical shard sizes sum across nodes
|
||||
// without any dedupe. Logical size excludes parity
|
||||
// shards; physical size includes them. Upstream OSS
|
||||
// uses the fixed 10+4 ratio (dataShards=0 → default);
|
||||
// forks with per-volume ratio metadata can pass the
|
||||
// configured value here.
|
||||
cif.PhysicalSize += float64(erasure_coding.EcShardsTotalSize(esi))
|
||||
cif.Size += float64(erasure_coding.EcShardsDataSize(esi, 0))
|
||||
|
||||
key := volumeKey{collection: c, volumeId: esi.Id}
|
||||
agg, ok := ecVolumes[key]
|
||||
if !ok {
|
||||
agg = &ecVolumeAgg{collection: c}
|
||||
ecVolumes[key] = agg
|
||||
cif.VolumeCount++
|
||||
}
|
||||
if esi.FileCount > agg.fileCount {
|
||||
agg.fileCount = esi.FileCount
|
||||
}
|
||||
agg.deleteCount += esi.DeleteCount
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Fold deduped EC file/delete counts into each collection's totals.
|
||||
for _, agg := range ecVolumes {
|
||||
cif := collectionInfos[agg.collection]
|
||||
if cif == nil {
|
||||
continue
|
||||
}
|
||||
cif.FileCount += float64(agg.fileCount)
|
||||
cif.DeleteCount += float64(agg.deleteCount)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,219 +2,8 @@ package s3api
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
)
|
||||
|
||||
// TestCollectCollectionInfoFromTopologyEC verifies that EC-encoded volumes
|
||||
// contribute to per-collection logical/physical size, file count, and volume
|
||||
// count. Before this fix, encoding a volume to EC caused the bucket size
|
||||
// metrics exported to Prometheus to drop to zero for that volume.
|
||||
//
|
||||
// Layout: one 10+4 EC volume in collection "crm-docs-storage", 14 shards of
|
||||
// 1000 bytes each split across two nodes.
|
||||
// - nodeA holds data shards 0..6 (7 * 1000 = 7000)
|
||||
// - nodeB holds data shards 7..9 (3 * 1000 = 3000) and parity 10..13 (4 * 1000 = 4000)
|
||||
//
|
||||
// Expected:
|
||||
// - PhysicalSize = 14 * 1000 = 14000
|
||||
// - Size (logical, data shards) = 10 * 1000 = 10000
|
||||
// - FileCount = 100 total - (2 + 3) local deletes = 95 is NOT what we check; the
|
||||
// collector reports raw file_count and delete_count as separate gauges, so
|
||||
// we assert FileCount = 100 (max across reporters) and DeleteCount = 5 (sum).
|
||||
// - VolumeCount = 1 (one unique EC volume)
|
||||
func TestCollectCollectionInfoFromTopologyEC(t *testing.T) {
|
||||
nodeA := &master_pb.DataNodeInfo{
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{
|
||||
"disk1": {
|
||||
EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{
|
||||
Id: 42,
|
||||
Collection: "crm-docs-storage",
|
||||
EcIndexBits: (1 << 0) | (1 << 1) | (1 << 2) | (1 << 3) | (1 << 4) | (1 << 5) | (1 << 6),
|
||||
ShardSizes: []int64{1000, 1000, 1000, 1000, 1000, 1000, 1000},
|
||||
FileCount: 100,
|
||||
DeleteCount: 2,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
nodeB := &master_pb.DataNodeInfo{
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{
|
||||
"disk1": {
|
||||
EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{
|
||||
Id: 42,
|
||||
Collection: "crm-docs-storage",
|
||||
EcIndexBits: (1 << 7) | (1 << 8) | (1 << 9) | (1 << 10) | (1 << 11) | (1 << 12) | (1 << 13),
|
||||
ShardSizes: []int64{1000, 1000, 1000, 1000, 1000, 1000, 1000},
|
||||
FileCount: 100,
|
||||
DeleteCount: 3,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
topo := &master_pb.TopologyInfo{
|
||||
DataCenterInfos: []*master_pb.DataCenterInfo{
|
||||
{
|
||||
RackInfos: []*master_pb.RackInfo{
|
||||
{
|
||||
DataNodeInfos: []*master_pb.DataNodeInfo{nodeA, nodeB},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
got := make(map[string]*CollectionInfo)
|
||||
collectCollectionInfoFromTopology(topo, got)
|
||||
|
||||
info, ok := got["crm-docs-storage"]
|
||||
if !ok {
|
||||
t.Fatalf("expected collection crm-docs-storage, got: %v", got)
|
||||
}
|
||||
if info.PhysicalSize != 14000 {
|
||||
t.Errorf("PhysicalSize: got %.0f, want 14000", info.PhysicalSize)
|
||||
}
|
||||
if info.Size != 10000 {
|
||||
t.Errorf("Size (logical): got %.0f, want 10000", info.Size)
|
||||
}
|
||||
if info.FileCount != 100 {
|
||||
t.Errorf("FileCount: got %.0f, want 100 (max across reporters)", info.FileCount)
|
||||
}
|
||||
if info.DeleteCount != 5 {
|
||||
t.Errorf("DeleteCount: got %.0f, want 5 (sum across reporters)", info.DeleteCount)
|
||||
}
|
||||
if info.VolumeCount != 1 {
|
||||
t.Errorf("VolumeCount: got %d, want 1", info.VolumeCount)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCollectCollectionInfoFromTopologyMixed verifies that regular and EC
|
||||
// volumes accumulate under the same collection without one clobbering the
|
||||
// other, which is the state during an in-progress EC conversion.
|
||||
func TestCollectCollectionInfoFromTopologyMixed(t *testing.T) {
|
||||
node := &master_pb.DataNodeInfo{
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{
|
||||
"disk1": {
|
||||
VolumeInfos: []*master_pb.VolumeInformationMessage{
|
||||
{
|
||||
Id: 1,
|
||||
Collection: "bucket-mix",
|
||||
Size: 5000,
|
||||
FileCount: 50,
|
||||
DeleteCount: 1,
|
||||
DeletedByteCount: 100,
|
||||
},
|
||||
},
|
||||
EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{
|
||||
Id: 2,
|
||||
Collection: "bucket-mix",
|
||||
EcIndexBits: (1 << 0) | (1 << 1) | (1 << 10), // 2 data + 1 parity
|
||||
ShardSizes: []int64{3000, 3000, 3000},
|
||||
FileCount: 80,
|
||||
DeleteCount: 4,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
topo := &master_pb.TopologyInfo{
|
||||
DataCenterInfos: []*master_pb.DataCenterInfo{
|
||||
{
|
||||
RackInfos: []*master_pb.RackInfo{
|
||||
{
|
||||
DataNodeInfos: []*master_pb.DataNodeInfo{node},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
got := make(map[string]*CollectionInfo)
|
||||
collectCollectionInfoFromTopology(topo, got)
|
||||
|
||||
info, ok := got["bucket-mix"]
|
||||
if !ok {
|
||||
t.Fatalf("expected collection bucket-mix, got: %v", got)
|
||||
}
|
||||
// Regular volume: 5000 physical + logical. EC shards: 9000 physical,
|
||||
// 6000 logical (data shards 0 and 1).
|
||||
if info.PhysicalSize != 5000+9000 {
|
||||
t.Errorf("PhysicalSize: got %.0f, want 14000", info.PhysicalSize)
|
||||
}
|
||||
if info.Size != 5000+6000 {
|
||||
t.Errorf("Size: got %.0f, want 11000", info.Size)
|
||||
}
|
||||
// LogicalSize drops the 100 bytes of un-vacuumed garbage on the regular
|
||||
// volume; EC shards carry no DeletedByteCount here.
|
||||
if info.LogicalSize() != 11000-100 {
|
||||
t.Errorf("LogicalSize: got %.0f, want 10900", info.LogicalSize())
|
||||
}
|
||||
if info.FileCount != 50+80 {
|
||||
t.Errorf("FileCount: got %.0f, want 130", info.FileCount)
|
||||
}
|
||||
if info.DeleteCount != 1+4 {
|
||||
t.Errorf("DeleteCount: got %.0f, want 5", info.DeleteCount)
|
||||
}
|
||||
if info.VolumeCount != 2 {
|
||||
t.Errorf("VolumeCount: got %d, want 2", info.VolumeCount)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCollectCollectionInfoFromTopologyECFileCountMaxDedupe verifies that a
|
||||
// slow shard holder reporting file_count=0 (because it has not yet finished
|
||||
// loading .ecx) does not pin the per-volume FileCount at 0.
|
||||
func TestCollectCollectionInfoFromTopologyECFileCountMaxDedupe(t *testing.T) {
|
||||
makeNode := func(bits uint32, sizes []int64, fileCount uint64) *master_pb.DataNodeInfo {
|
||||
return &master_pb.DataNodeInfo{
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{
|
||||
"disk1": {
|
||||
EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{
|
||||
Id: 11,
|
||||
Collection: "bucket-b",
|
||||
EcIndexBits: bits,
|
||||
ShardSizes: sizes,
|
||||
FileCount: fileCount,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
topo := &master_pb.TopologyInfo{
|
||||
DataCenterInfos: []*master_pb.DataCenterInfo{
|
||||
{
|
||||
RackInfos: []*master_pb.RackInfo{
|
||||
{
|
||||
DataNodeInfos: []*master_pb.DataNodeInfo{
|
||||
makeNode((1<<0)|(1<<1)|(1<<2)|(1<<3)|(1<<4)|(1<<5)|(1<<6), []int64{1, 1, 1, 1, 1, 1, 1}, 0),
|
||||
makeNode((1<<7)|(1<<8)|(1<<9)|(1<<10)|(1<<11)|(1<<12)|(1<<13), []int64{1, 1, 1, 1, 1, 1, 1}, 6),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
got := make(map[string]*CollectionInfo)
|
||||
collectCollectionInfoFromTopology(topo, got)
|
||||
info, ok := got["bucket-b"]
|
||||
if !ok {
|
||||
t.Fatalf("expected collection bucket-b, got: %v", got)
|
||||
}
|
||||
if info.FileCount != 6 {
|
||||
t.Errorf("FileCount: got %.0f, want 6 (max across reporters)", info.FileCount)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCollectionInfoLogicalSize verifies logical size excludes un-vacuumed
|
||||
// garbage and never goes negative. Quota enforcement runs on this value so a
|
||||
// bucket full of tombstones is not flipped read-only while its live data is
|
||||
|
||||
@@ -198,6 +198,31 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// CollectionStatistics summarises every collection, so callers tracking usage
|
||||
// do not pull the whole volume list to add it up themselves.
|
||||
func (ms *MasterServer) CollectionStatistics(ctx context.Context, req *master_pb.CollectionStatisticsRequest) (*master_pb.CollectionStatisticsResponse, error) {
|
||||
if !ms.Topo.IsLeader() {
|
||||
return nil, raft.NotLeaderError
|
||||
}
|
||||
|
||||
stats := ms.Topo.CollectionStatistics()
|
||||
resp := &master_pb.CollectionStatisticsResponse{
|
||||
Collections: make([]*master_pb.CollectionStatistics, 0, len(stats)),
|
||||
}
|
||||
for _, s := range stats {
|
||||
resp.Collections = append(resp.Collections, &master_pb.CollectionStatistics{
|
||||
Collection: s.Collection,
|
||||
FileCount: s.FileCount,
|
||||
DeleteCount: s.DeleteCount,
|
||||
DeletedByteCount: s.DeletedByteCount,
|
||||
Size: s.Size,
|
||||
PhysicalSize: s.PhysicalSize,
|
||||
VolumeCount: s.VolumeCount,
|
||||
})
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.StatisticsRequest) (*master_pb.StatisticsResponse, error) {
|
||||
|
||||
if !ms.Topo.IsLeader() {
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
package topology
|
||||
|
||||
import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
)
|
||||
|
||||
// CollectionStatistics is what a collection holds, summarised so that callers
|
||||
// tracking usage do not have to be sent every volume in the cluster to add it
|
||||
// up themselves.
|
||||
type CollectionStatistics struct {
|
||||
Collection string
|
||||
FileCount uint64
|
||||
DeleteCount uint64
|
||||
DeletedByteCount uint64
|
||||
// Size counts one copy of the data: a single replica of a regular volume,
|
||||
// the data shards of an ec volume.
|
||||
Size uint64
|
||||
// PhysicalSize counts what is on disk: every replica, and parity shards.
|
||||
PhysicalSize uint64
|
||||
VolumeCount uint64
|
||||
}
|
||||
|
||||
type ecStatsKey struct {
|
||||
collection string
|
||||
volumeId needle.VolumeId
|
||||
}
|
||||
|
||||
// ecFileCounts holds the per-volume counts that can only be resolved once every
|
||||
// shard holder has been seen.
|
||||
type ecFileCounts struct {
|
||||
collection string
|
||||
fileCount uint64
|
||||
deleteCount uint64
|
||||
}
|
||||
|
||||
// CollectionStatistics summarises every collection in one pass over the
|
||||
// topology, allocating per collection rather than per volume.
|
||||
func (t *Topology) CollectionStatistics() []*CollectionStatistics {
|
||||
byCollection := make(map[string]*CollectionStatistics)
|
||||
statsFor := func(collection string) *CollectionStatistics {
|
||||
stats, found := byCollection[collection]
|
||||
if !found {
|
||||
stats = &CollectionStatistics{Collection: collection}
|
||||
byCollection[collection] = stats
|
||||
}
|
||||
return stats
|
||||
}
|
||||
|
||||
// Regular volumes are counted once each for logical totals and once per
|
||||
// replica for physical, which the lookup index gives without a set of seen
|
||||
// ids: it is already keyed by volume.
|
||||
for _, c := range t.collectionMap.Items() {
|
||||
collection := c.(*Collection)
|
||||
for _, vl := range collection.GetAllVolumeLayouts() {
|
||||
vl.accessLock.RLock()
|
||||
for vid, locations := range vl.vid2location {
|
||||
stats := statsFor(collection.Name)
|
||||
// Replicas of one volume can disagree while a write is landing
|
||||
// or a heartbeat is late. Count the one holding the most live
|
||||
// data, so the answer does not depend on which replica is
|
||||
// looked at first and usage is never reported lower than some
|
||||
// replica already holds. Quotas are enforced on size less
|
||||
// deletions, so that is what has to be the largest -- a replica
|
||||
// with the biggest raw size can be the one that has deleted the
|
||||
// most, and picking it would leave an over-quota bucket
|
||||
// writable.
|
||||
var largest storage.VolumeInfo
|
||||
var largestLive uint64
|
||||
found := false
|
||||
for _, dn := range locations.list {
|
||||
v, err := dn.GetVolumesById(vid)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
stats.PhysicalSize += v.Size
|
||||
live := v.Size
|
||||
if v.DeletedByteCount < live {
|
||||
live -= v.DeletedByteCount
|
||||
} else {
|
||||
live = 0
|
||||
}
|
||||
if !found || live > largestLive {
|
||||
largest, largestLive, found = v, live, true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
stats.Size += largest.Size
|
||||
stats.FileCount += uint64(largest.FileCount)
|
||||
stats.DeleteCount += uint64(largest.DeleteCount)
|
||||
// Never more deletions than the volume holds. Live usage is
|
||||
// read as the collection's size less its deletions, so a volume
|
||||
// reporting more deleted bytes than it has would cancel live
|
||||
// bytes belonging to other volumes and report the bucket
|
||||
// smaller than it is.
|
||||
stats.DeletedByteCount += largest.Size - largestLive
|
||||
stats.VolumeCount++
|
||||
}
|
||||
vl.accessLock.RUnlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Ec shards are node-local rather than replicated, so their sizes sum
|
||||
// across holders. The file and delete counts describe the volume rather
|
||||
// than the shard, so they resolve once every holder has been seen.
|
||||
perEcVolume := make(map[ecStatsKey]*ecFileCounts)
|
||||
for _, dcNode := range t.Children() {
|
||||
for _, rackNode := range dcNode.(*DataCenter).Children() {
|
||||
for _, dnNode := range rackNode.(*Rack).Children() {
|
||||
for _, ecInfo := range dnNode.(*DataNode).GetEcShards() {
|
||||
message := ecInfo.ToVolumeEcShardInformationMessage()
|
||||
stats := statsFor(ecInfo.Collection)
|
||||
stats.PhysicalSize += uint64(erasure_coding.EcShardsTotalSize(message))
|
||||
stats.Size += uint64(erasure_coding.EcShardsDataSize(message, 0))
|
||||
|
||||
key := ecStatsKey{collection: ecInfo.Collection, volumeId: ecInfo.VolumeId}
|
||||
counts, found := perEcVolume[key]
|
||||
if !found {
|
||||
counts = &ecFileCounts{collection: ecInfo.Collection}
|
||||
perEcVolume[key] = counts
|
||||
stats.VolumeCount++
|
||||
}
|
||||
if message.FileCount > counts.fileCount {
|
||||
counts.fileCount = message.FileCount
|
||||
}
|
||||
counts.deleteCount += message.DeleteCount
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, counts := range perEcVolume {
|
||||
stats := byCollection[counts.collection]
|
||||
if stats == nil {
|
||||
continue
|
||||
}
|
||||
stats.FileCount += counts.fileCount
|
||||
stats.DeleteCount += counts.deleteCount
|
||||
}
|
||||
|
||||
ret := make([]*CollectionStatistics, 0, len(byCollection))
|
||||
for _, stats := range byCollection {
|
||||
ret = append(ret, stats)
|
||||
}
|
||||
return ret
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
package topology
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
)
|
||||
|
||||
// referenceCollectionStatistics is what callers computed for themselves from a
|
||||
// full topology listing, kept here so the summary can be held to producing the
|
||||
// same numbers. Quota enforcement reads these.
|
||||
func referenceCollectionStatistics(t *master_pb.TopologyInfo) map[string]*CollectionStatistics {
|
||||
type volumeKey struct {
|
||||
collection string
|
||||
volumeId uint32
|
||||
}
|
||||
out := map[string]*CollectionStatistics{}
|
||||
seen := map[volumeKey]bool{}
|
||||
ecCounts := map[volumeKey]*ecFileCounts{}
|
||||
statsFor := func(c string) *CollectionStatistics {
|
||||
s, ok := out[c]
|
||||
if !ok {
|
||||
s = &CollectionStatistics{Collection: c}
|
||||
out[c] = s
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
for _, dc := range t.DataCenterInfos {
|
||||
for _, r := range dc.RackInfos {
|
||||
for _, dn := range r.DataNodeInfos {
|
||||
for _, diskInfo := range dn.DiskInfos {
|
||||
for _, vi := range diskInfo.VolumeInfos {
|
||||
s := statsFor(vi.Collection)
|
||||
s.PhysicalSize += vi.Size
|
||||
key := volumeKey{vi.Collection, vi.Id}
|
||||
if seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
s.Size += vi.Size
|
||||
s.FileCount += vi.FileCount
|
||||
s.DeleteCount += vi.DeleteCount
|
||||
s.DeletedByteCount += vi.DeletedByteCount
|
||||
s.VolumeCount++
|
||||
}
|
||||
for _, esi := range diskInfo.EcShardInfos {
|
||||
s := statsFor(esi.Collection)
|
||||
s.PhysicalSize += uint64(erasure_coding.EcShardsTotalSize(esi))
|
||||
s.Size += uint64(erasure_coding.EcShardsDataSize(esi, 0))
|
||||
key := volumeKey{esi.Collection, esi.Id}
|
||||
agg, ok := ecCounts[key]
|
||||
if !ok {
|
||||
agg = &ecFileCounts{collection: esi.Collection}
|
||||
ecCounts[key] = agg
|
||||
s.VolumeCount++
|
||||
}
|
||||
if esi.FileCount > agg.fileCount {
|
||||
agg.fileCount = esi.FileCount
|
||||
}
|
||||
agg.deleteCount += esi.DeleteCount
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, agg := range ecCounts {
|
||||
if s := out[agg.collection]; s != nil {
|
||||
s.FileCount += agg.fileCount
|
||||
s.DeleteCount += agg.deleteCount
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func statsTopology(t *testing.T) *Topology {
|
||||
t.Helper()
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
|
||||
nodes := make([]*DataNode, 3)
|
||||
for i := range nodes {
|
||||
nodes[i] = rack.GetOrCreateDataNode(
|
||||
"10.0.0."+string(rune('1'+i)), 8080, 18080, "", "", map[string]uint32{"": 1000, "ssd": 1000})
|
||||
}
|
||||
|
||||
// Two collections, replicated volumes, one volume on a second disk type.
|
||||
// Replicas agree here; disagreement is covered on its own, where the
|
||||
// listing this is compared against is itself order dependent.
|
||||
report := func(dn *DataNode, msgs ...*master_pb.VolumeInformationMessage) {
|
||||
topo.SyncDataNodeRegistration(msgs, dn)
|
||||
}
|
||||
vol := func(id uint32, collection string, size uint64, diskType string) *master_pb.VolumeInformationMessage {
|
||||
return &master_pb.VolumeInformationMessage{
|
||||
Id: id, Collection: collection, Size: size, FileCount: uint64(id) * 10,
|
||||
DeleteCount: uint64(id), DeletedByteCount: uint64(id) * 100,
|
||||
Version: 3, ReplicaPlacement: 1, DiskType: diskType,
|
||||
}
|
||||
}
|
||||
report(nodes[0], vol(1, "bucket-a", 1000, ""), vol(2, "bucket-a", 2000, ""), vol(5, "bucket-b", 700, "ssd"))
|
||||
report(nodes[1], vol(1, "bucket-a", 1000, ""), vol(3, "bucket-b", 3000, ""))
|
||||
report(nodes[2], vol(2, "bucket-a", 2000, ""), vol(3, "bucket-b", 3000, ""))
|
||||
|
||||
// Ec shards for a third collection, spread over two nodes, with the file
|
||||
// count reported differently by each holder.
|
||||
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 9, Collection: "bucket-c", EcIndexBits: 0x1f, ShardSizes: []int64{10, 20, 30, 40, 50}, FileCount: 40, DeleteCount: 2},
|
||||
}, nodes[0])
|
||||
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 9, Collection: "bucket-c", EcIndexBits: 0x3e0, ShardSizes: []int64{60, 70, 80, 90, 100}, FileCount: 44, DeleteCount: 3},
|
||||
}, nodes[1])
|
||||
return topo
|
||||
}
|
||||
|
||||
// The summary replaces callers adding up a full topology listing, so it has to
|
||||
// produce what that produced: these numbers enforce bucket quotas.
|
||||
func TestCollectionStatisticsMatchesAFullListing(t *testing.T) {
|
||||
topo := statsTopology(t)
|
||||
|
||||
want := referenceCollectionStatistics(topo.ToTopologyInfo())
|
||||
got := map[string]*CollectionStatistics{}
|
||||
for _, s := range topo.CollectionStatistics() {
|
||||
got[s.Collection] = s
|
||||
}
|
||||
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("summarised %d collections, the listing had %d: %v vs %v", len(got), len(want), got, want)
|
||||
}
|
||||
for name, expected := range want {
|
||||
actual, found := got[name]
|
||||
if !found {
|
||||
t.Errorf("collection %s is missing from the summary", name)
|
||||
continue
|
||||
}
|
||||
if *actual != *expected {
|
||||
t.Errorf("collection %s:\n summary %+v\n listing %+v", name, *actual, *expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Replicas disagree while a write is landing or a heartbeat is late. The full
|
||||
// listing walked the topology in map order and took whichever replica it
|
||||
// reached first, so its answer was not stable; the summary takes the largest.
|
||||
func TestCollectionStatisticsPicksTheLargestReplica(t *testing.T) {
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
|
||||
small := rack.GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "a", map[string]uint32{"": 100})
|
||||
large := rack.GetOrCreateDataNode("10.0.0.2", 8080, 18080, "", "b", map[string]uint32{"": 100})
|
||||
|
||||
behind := &master_pb.VolumeInformationMessage{
|
||||
Id: 1, Collection: "bucket-a", Size: 1000, FileCount: 10, Version: 3, ReplicaPlacement: 1,
|
||||
}
|
||||
ahead := &master_pb.VolumeInformationMessage{
|
||||
Id: 1, Collection: "bucket-a", Size: 4000, FileCount: 40, Version: 3, ReplicaPlacement: 1,
|
||||
}
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{behind}, small)
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ahead}, large)
|
||||
|
||||
for i := 0; i < 8; i++ {
|
||||
stats := topo.CollectionStatistics()
|
||||
if len(stats) != 1 {
|
||||
t.Fatalf("expected one collection, got %d", len(stats))
|
||||
}
|
||||
if stats[0].Size != 4000 || stats[0].FileCount != 40 {
|
||||
t.Fatalf("run %d reported size %d files %d, want the larger replica's 4000 and 40",
|
||||
i, stats[0].Size, stats[0].FileCount)
|
||||
}
|
||||
if stats[0].PhysicalSize != 5000 {
|
||||
t.Fatalf("run %d reported physical size %d, want both replicas summed", i, stats[0].PhysicalSize)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func statsByCollection(topo *Topology) map[string]*CollectionStatistics {
|
||||
out := map[string]*CollectionStatistics{}
|
||||
for _, s := range topo.CollectionStatistics() {
|
||||
out[s.Collection] = s
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// The .ecx and .ecj travel with the shards, so a holder still loading them
|
||||
// reports zero and must not pin the volume's count down.
|
||||
func TestCollectionStatisticsTakesTheLargestEcFileCount(t *testing.T) {
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
|
||||
loading := rack.GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "a", map[string]uint32{"": 100})
|
||||
loaded := rack.GetOrCreateDataNode("10.0.0.2", 8080, 18080, "", "b", map[string]uint32{"": 100})
|
||||
|
||||
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 11, Collection: "bucket-b", EcIndexBits: 0x7f, ShardSizes: []int64{1, 1, 1, 1, 1, 1, 1}, FileCount: 0},
|
||||
}, loading)
|
||||
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 11, Collection: "bucket-b", EcIndexBits: 0x3f80, ShardSizes: []int64{1, 1, 1, 1, 1, 1, 1}, FileCount: 6},
|
||||
}, loaded)
|
||||
|
||||
stats := statsByCollection(topo)["bucket-b"]
|
||||
if stats == nil {
|
||||
t.Fatal("expected bucket-b to be reported")
|
||||
}
|
||||
if stats.FileCount != 6 {
|
||||
t.Errorf("file count %d, want the largest any holder reported (6)", stats.FileCount)
|
||||
}
|
||||
if stats.VolumeCount != 1 {
|
||||
t.Errorf("volume count %d, want one volume however many holders report shards", stats.VolumeCount)
|
||||
}
|
||||
// 14 shards of 1 byte each; only the 10 data shards count as logical.
|
||||
if stats.PhysicalSize != 14 {
|
||||
t.Errorf("physical size %d, want every shard counted (14)", stats.PhysicalSize)
|
||||
}
|
||||
if stats.Size != 10 {
|
||||
t.Errorf("size %d, want the data shards only (10)", stats.Size)
|
||||
}
|
||||
}
|
||||
|
||||
// A collection with both kinds has to have them added together.
|
||||
func TestCollectionStatisticsCountsRegularAndEcTogether(t *testing.T) {
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
|
||||
dn := rack.GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "a", map[string]uint32{"": 100})
|
||||
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
|
||||
{Id: 1, Collection: "mixed", Size: 500, FileCount: 5, DeleteCount: 1, DeletedByteCount: 50, Version: 3},
|
||||
}, dn)
|
||||
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 2, Collection: "mixed", EcIndexBits: 0x3fff,
|
||||
ShardSizes: []int64{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, FileCount: 7, DeleteCount: 2},
|
||||
}, dn)
|
||||
|
||||
stats := statsByCollection(topo)["mixed"]
|
||||
if stats == nil {
|
||||
t.Fatal("expected the mixed collection to be reported")
|
||||
}
|
||||
if stats.VolumeCount != 2 {
|
||||
t.Errorf("volume count %d, want the regular and the ec volume (2)", stats.VolumeCount)
|
||||
}
|
||||
if stats.FileCount != 12 {
|
||||
t.Errorf("file count %d, want 5 regular plus 7 ec", stats.FileCount)
|
||||
}
|
||||
if stats.DeleteCount != 3 {
|
||||
t.Errorf("delete count %d, want 1 regular plus 2 ec", stats.DeleteCount)
|
||||
}
|
||||
if stats.Size != 510 {
|
||||
t.Errorf("size %d, want 500 regular plus 10 data shards", stats.Size)
|
||||
}
|
||||
if stats.PhysicalSize != 514 {
|
||||
t.Errorf("physical size %d, want 500 regular plus all 14 shards", stats.PhysicalSize)
|
||||
}
|
||||
}
|
||||
|
||||
// Quotas are enforced on size less deletions, so the replica holding the most
|
||||
// live data is the one to count. The replica with the biggest raw size can be
|
||||
// the one that has deleted the most, and counting that one would report a
|
||||
// bucket smaller than it is and leave it writable over its quota.
|
||||
func TestCollectionStatisticsPicksTheReplicaHoldingTheMostLiveData(t *testing.T) {
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
|
||||
bigMostlyDeleted := rack.GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "a", map[string]uint32{"": 100})
|
||||
smallerButLive := rack.GetOrCreateDataNode("10.0.0.2", 8080, 18080, "", "b", map[string]uint32{"": 100})
|
||||
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
|
||||
Id: 1, Collection: "bucket-a", Size: 1000, DeletedByteCount: 900,
|
||||
Version: 3, ReplicaPlacement: 1,
|
||||
}}, bigMostlyDeleted)
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
|
||||
Id: 1, Collection: "bucket-a", Size: 900, DeletedByteCount: 0,
|
||||
Version: 3, ReplicaPlacement: 1,
|
||||
}}, smallerButLive)
|
||||
|
||||
stats := statsByCollection(topo)["bucket-a"]
|
||||
if stats == nil {
|
||||
t.Fatal("expected bucket-a to be reported")
|
||||
}
|
||||
live := stats.Size - stats.DeletedByteCount
|
||||
if live != 900 {
|
||||
t.Errorf("reported %d bytes live (size %d less %d deleted), want the 900 one replica holds",
|
||||
live, stats.Size, stats.DeletedByteCount)
|
||||
}
|
||||
if stats.PhysicalSize != 1900 {
|
||||
t.Errorf("physical size %d, want both replicas summed", stats.PhysicalSize)
|
||||
}
|
||||
}
|
||||
|
||||
// Live usage is read as the collection's size less its deletions, so a volume
|
||||
// reporting more deleted bytes than it holds must not cancel live bytes
|
||||
// belonging to other volumes in the same bucket.
|
||||
func TestCollectionStatisticsDeletionsNeverExceedTheVolume(t *testing.T) {
|
||||
topo := NewTopology("stats", nil, 32*1024*1024*1024, 5, false)
|
||||
dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
|
||||
GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "a", map[string]uint32{"": 100})
|
||||
|
||||
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
|
||||
// More deleted bytes than the volume holds, which a compaction can
|
||||
// leave behind, and a healthy volume beside it.
|
||||
{Id: 1, Collection: "bucket-a", Size: 100, DeletedByteCount: 500, Version: 3},
|
||||
{Id: 2, Collection: "bucket-a", Size: 1000, DeletedByteCount: 0, Version: 3},
|
||||
}, dn)
|
||||
|
||||
stats := statsByCollection(topo)["bucket-a"]
|
||||
if stats == nil {
|
||||
t.Fatal("expected bucket-a to be reported")
|
||||
}
|
||||
if stats.DeletedByteCount > stats.Size {
|
||||
t.Errorf("deletions %d exceed size %d, so live usage reads as zero", stats.DeletedByteCount, stats.Size)
|
||||
}
|
||||
if live := stats.Size - stats.DeletedByteCount; live != 1000 {
|
||||
t.Errorf("reported %d bytes live, want the 1000 the second volume holds", live)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user