mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-09 21:38:19 +01:00
getting ec volume deletions
This commit is contained in:
chrislu committed
1 parent
553a229fd3
commit
97d58e77c6
9 files changed
+1094
-660
No files matched your search
@@ -915,26 +915,61 @@ func (s *AdminServer) getVolumeHealthFromServer(server string, volumeID uint32)
|
||||
var healthInfo *EcVolumeHealthInfo
|
||||
|
||||
err := s.WithVolumeServerClient(pb.ServerAddress(server), func(client volume_server_pb.VolumeServerClient) error {
|
||||
var collection string = "" // Default collection name
|
||||
var totalSize uint64 = 0
|
||||
var fileCount uint64 = 0
|
||||
|
||||
// Try to get volume file status (which may include original volume metrics)
|
||||
// This will fail for EC-only volumes, so we handle that gracefully
|
||||
resp, err := client.ReadVolumeFileStatus(context.Background(), &volume_server_pb.ReadVolumeFileStatusRequest{
|
||||
VolumeId: volumeID,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
glog.V(2).Infof("ReadVolumeFileStatus failed for EC volume %d on server %s (expected for EC-only volumes): %v", volumeID, server, err)
|
||||
// For EC-only volumes, we don't have original volume metrics, but we can still get deletion info
|
||||
} else if resp.VolumeInfo != nil {
|
||||
// Extract metrics from regular volume info if available
|
||||
totalSize = uint64(resp.VolumeInfo.DatFileSize)
|
||||
fileCount = resp.FileCount
|
||||
collection = resp.Collection
|
||||
}
|
||||
|
||||
// Extract health metrics from volume info
|
||||
if resp.VolumeInfo != nil {
|
||||
totalSize := uint64(resp.VolumeInfo.DatFileSize)
|
||||
// Always try to get EC deletion information using the new gRPC endpoint
|
||||
deletionResp, deletionErr := client.VolumeEcDeletionInfo(context.Background(), &volume_server_pb.VolumeEcDeletionInfoRequest{
|
||||
VolumeId: volumeID,
|
||||
Collection: collection,
|
||||
Generation: 0, // Use default generation for backward compatibility
|
||||
})
|
||||
|
||||
if deletionErr != nil {
|
||||
glog.V(1).Infof("Failed to get EC deletion info for volume %d on server %s: %v", volumeID, server, deletionErr)
|
||||
// If we have some info from ReadVolumeFileStatus, still create healthInfo with that
|
||||
if totalSize > 0 {
|
||||
healthInfo = &EcVolumeHealthInfo{
|
||||
TotalSize: totalSize,
|
||||
DeletedByteCount: 0, // EC volumes don't track deletions in VolumeInfo
|
||||
FileCount: resp.FileCount,
|
||||
DeleteCount: 0, // Not available in current API
|
||||
DeletedByteCount: 0,
|
||||
FileCount: fileCount,
|
||||
DeleteCount: 0,
|
||||
GarbageRatio: 0.0,
|
||||
}
|
||||
}
|
||||
} else if deletionResp != nil {
|
||||
// Create health info with deletion data
|
||||
healthInfo = &EcVolumeHealthInfo{
|
||||
TotalSize: totalSize, // May be 0 for EC-only volumes
|
||||
DeletedByteCount: deletionResp.DeletedBytes,
|
||||
FileCount: fileCount,
|
||||
DeleteCount: deletionResp.DeletedCount,
|
||||
GarbageRatio: 0.0,
|
||||
}
|
||||
|
||||
// Calculate garbage ratio if we have total size
|
||||
if healthInfo.TotalSize > 0 {
|
||||
healthInfo.GarbageRatio = float64(healthInfo.DeletedByteCount) / float64(healthInfo.TotalSize)
|
||||
}
|
||||
|
||||
glog.V(1).Infof("EC volume %d on server %s: %d deleted bytes, %d deleted needles, total size: %d bytes",
|
||||
volumeID, server, healthInfo.DeletedByteCount, healthInfo.DeleteCount, healthInfo.TotalSize)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -6,9 +6,14 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/operation"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
"github.com/seaweedfs/seaweedfs/weed/worker/types"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
)
|
||||
|
||||
// NewMaintenanceScanner creates a new maintenance scanner
|
||||
@@ -477,35 +482,28 @@ func (ms *MaintenanceScanner) collectECVolumeDelationsFromAllServers(volumeId ui
|
||||
}
|
||||
|
||||
// Merge deletion information, avoiding double counting
|
||||
newNeedles := 0
|
||||
for needle := range serverDeletedNeedles {
|
||||
if !deletedNeedles[needle] {
|
||||
deletedNeedles[needle] = true
|
||||
// We can't get exact size per needle without more complex analysis,
|
||||
// so we'll use the server's reported total as a conservative estimate
|
||||
// This could be enhanced with proper .ecj parsing
|
||||
newNeedles++
|
||||
}
|
||||
}
|
||||
|
||||
// For now, sum the deleted bytes from all servers
|
||||
// Note: This might double-count if the same needle is deleted across shards,
|
||||
// but it provides a reasonable upper bound estimate
|
||||
totalDeletedBytes += serverDeletedBytes
|
||||
|
||||
glog.V(3).Infof("Server %s reported %d deleted bytes for EC volume %d", server, serverDeletedBytes, volumeId)
|
||||
}
|
||||
|
||||
// Apply conservative adjustment to account for potential double counting
|
||||
// Since deletions are tracked per shard but affect the whole needle,
|
||||
// we should not simply sum all deleted bytes from all servers
|
||||
if len(servers) > 1 && totalDeletedBytes > 0 {
|
||||
// Conservative approach: assume some overlap and reduce the total
|
||||
adjustmentFactor := float64(len(deletedNeedles)) / float64(len(servers))
|
||||
if adjustmentFactor < 1.0 {
|
||||
totalDeletedBytes = int64(float64(totalDeletedBytes) * adjustmentFactor)
|
||||
glog.V(3).Infof("Applied conservative adjustment factor %.2f to EC volume %d deleted bytes", adjustmentFactor, volumeId)
|
||||
// Only add bytes for needles that are actually new to avoid double counting
|
||||
// Assume bytes are evenly distributed per needle for deduplication
|
||||
if len(serverDeletedNeedles) > 0 && serverDeletedBytes > 0 {
|
||||
avgBytesPerNeedle := float64(serverDeletedBytes) / float64(len(serverDeletedNeedles))
|
||||
totalDeletedBytes += int64(float64(newNeedles) * avgBytesPerNeedle)
|
||||
}
|
||||
|
||||
glog.V(3).Infof("Server %s reported %d deleted bytes, %d needles (%d new) for EC volume %d",
|
||||
server, serverDeletedBytes, len(serverDeletedNeedles), newNeedles, volumeId)
|
||||
}
|
||||
|
||||
glog.V(2).Infof("EC volume %d total: %d unique deleted needles, %d estimated deleted bytes from %d servers",
|
||||
volumeId, len(deletedNeedles), totalDeletedBytes, len(servers))
|
||||
|
||||
return totalDeletedBytes, nil
|
||||
}
|
||||
|
||||
@@ -557,17 +555,46 @@ func (ms *MaintenanceScanner) getServerECVolumeDeletions(volumeId uint32, collec
|
||||
// return totalDeleted, deletedNeedleMap, nil
|
||||
// })
|
||||
|
||||
// For now, implement a conservative heuristic approach
|
||||
// Use the new VolumeEcDeletionInfo gRPC endpoint to get accurate deletion data
|
||||
var deletedBytes int64 = 0
|
||||
deletedNeedles := make(map[string]bool)
|
||||
|
||||
// Very conservative estimate to avoid false positives
|
||||
// This will be replaced with proper .ecj/.ecx analysis
|
||||
conservativeEstimate := int64(1024) // 1KB conservative estimate per server
|
||||
glog.V(0).Infof("Making gRPC call to server %s for volume %d collection %s", server, volumeId, collection)
|
||||
|
||||
glog.V(4).Infof("Applied conservative deletion estimate for EC volume %d on server %s: %d bytes (heuristic mode)",
|
||||
volumeId, server, conservativeEstimate)
|
||||
err := operation.WithVolumeServerClient(false, pb.ServerAddress(server),
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()), func(client volume_server_pb.VolumeServerClient) error {
|
||||
glog.V(0).Infof("Connected to volume server %s, calling VolumeEcDeletionInfo", server)
|
||||
resp, err := client.VolumeEcDeletionInfo(context.Background(), &volume_server_pb.VolumeEcDeletionInfoRequest{
|
||||
VolumeId: volumeId,
|
||||
Collection: collection,
|
||||
Generation: 0, // Use default generation for backward compatibility
|
||||
})
|
||||
if err != nil {
|
||||
glog.V(0).Infof("VolumeEcDeletionInfo call failed for server %s: %v", server, err)
|
||||
return err
|
||||
}
|
||||
|
||||
return conservativeEstimate, deletedNeedles, nil
|
||||
deletedBytes = int64(resp.DeletedBytes)
|
||||
|
||||
// Convert deleted needle IDs to map for duplicate tracking
|
||||
for _, needleId := range resp.DeletedNeedleIds {
|
||||
deletedNeedles[fmt.Sprintf("%d", needleId)] = true
|
||||
}
|
||||
|
||||
glog.V(0).Infof("Got EC deletion info for volume %d on server %s: %d bytes, %d needles",
|
||||
volumeId, server, deletedBytes, len(resp.DeletedNeedleIds))
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
glog.V(0).Infof("Failed to get EC deletion info for volume %d on server %s, using conservative estimate: %v", volumeId, server, err)
|
||||
// Fallback to conservative estimate if gRPC call fails
|
||||
deletedBytes = int64(1024) // 1KB conservative estimate
|
||||
}
|
||||
|
||||
glog.V(0).Infof("Returning from getServerECVolumeDeletions: %d bytes, %d needles", deletedBytes, len(deletedNeedles))
|
||||
return deletedBytes, deletedNeedles, nil
|
||||
}
|
||||
|
||||
// convertToTaskMetrics converts existing volume metrics to task system format
|
||||
|
||||
@@ -152,64 +152,102 @@ templ EcVolumeDetails(data dash.EcVolumeDetailsData) {
|
||||
</div>
|
||||
<div class="card-body">
|
||||
<!-- Storage Metrics -->
|
||||
if data.TotalSize > 0 {
|
||||
if data.TotalSize > 0 || data.DeletedByteCount > 0 || data.DeleteCount > 0 {
|
||||
<div class="row mb-3">
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold text-success">
|
||||
{bytesToHumanReadableUint64(data.TotalSize - data.DeletedByteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Active Bytes</small>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold text-danger">
|
||||
{bytesToHumanReadableUint64(data.DeletedByteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Deleted Bytes</small>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- File Metrics -->
|
||||
<div class="row mb-3">
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h5 mb-0 text-primary">
|
||||
{fmt.Sprintf("%d", data.FileCount - data.DeleteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Active Files</small>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h5 mb-0 text-warning">
|
||||
{fmt.Sprintf("%d", data.DeleteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Deleted Files</small>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Garbage Ratio for EC Vacuum -->
|
||||
<div class="row mb-3">
|
||||
<div class="col-12">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold" style={"color: " + getGarbageRatioColor(data.GarbageRatio)}>
|
||||
{fmt.Sprintf("%.1f%%", data.GarbageRatio * 100)}
|
||||
</div>
|
||||
<small class="text-muted">Garbage Ratio</small>
|
||||
if data.GarbageRatio >= 0.3 {
|
||||
<div class="mt-1">
|
||||
<span class="badge bg-warning">
|
||||
<i class="fas fa-broom me-1"></i>EC Vacuum Candidate
|
||||
</span>
|
||||
if data.TotalSize > 0 {
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold text-success">
|
||||
{bytesToHumanReadableUint64(data.TotalSize - data.DeletedByteCount)}
|
||||
</div>
|
||||
}
|
||||
<small class="text-muted">Active Bytes</small>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold text-danger">
|
||||
{bytesToHumanReadableUint64(data.DeletedByteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Deleted Bytes</small>
|
||||
</div>
|
||||
</div>
|
||||
} else {
|
||||
<!-- EC-only volume - show deletion info differently -->
|
||||
<div class="col-12">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold text-warning">
|
||||
{fmt.Sprintf("%d", data.DeleteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Deleted Needles (EC-only volume)</small>
|
||||
if data.DeleteCount > 0 {
|
||||
<div class="mt-1">
|
||||
<small class="text-info">
|
||||
<i class="fas fa-info-circle me-1"></i>
|
||||
Deletion info from .ecj files
|
||||
</small>
|
||||
</div>
|
||||
}
|
||||
</div>
|
||||
</div>
|
||||
}
|
||||
</div>
|
||||
|
||||
<!-- File Metrics - only show for regular volumes -->
|
||||
if data.TotalSize > 0 {
|
||||
<div class="row mb-3">
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h5 mb-0 text-primary">
|
||||
{fmt.Sprintf("%d", data.FileCount - data.DeleteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Active Files</small>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-6">
|
||||
<div class="text-center">
|
||||
<div class="h5 mb-0 text-warning">
|
||||
{fmt.Sprintf("%d", data.DeleteCount)}
|
||||
</div>
|
||||
<small class="text-muted">Deleted Files</small>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Garbage Ratio for EC Vacuum -->
|
||||
<div class="row mb-3">
|
||||
<div class="col-12">
|
||||
<div class="text-center">
|
||||
<div class="h4 mb-0 font-weight-bold" style={"color: " + getGarbageRatioColor(data.GarbageRatio)}>
|
||||
{fmt.Sprintf("%.1f%%", data.GarbageRatio * 100)}
|
||||
</div>
|
||||
<small class="text-muted">Garbage Ratio</small>
|
||||
if data.GarbageRatio >= 0.3 {
|
||||
<div class="mt-1">
|
||||
<span class="badge bg-warning">
|
||||
<i class="fas fa-broom me-1"></i>EC Vacuum Candidate
|
||||
</span>
|
||||
</div>
|
||||
}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
} else if data.DeleteCount > 0 {
|
||||
<!-- For EC-only volumes, show a note about EC vacuum eligibility -->
|
||||
<div class="row mb-3">
|
||||
<div class="col-12">
|
||||
<div class="text-center">
|
||||
<div class="mt-2">
|
||||
<span class="badge bg-info">
|
||||
<i class="fas fa-broom me-1"></i>EC Vacuum Eligible
|
||||
</span>
|
||||
<div class="mt-1">
|
||||
<small class="text-muted">Volume has {fmt.Sprintf("%d", data.DeleteCount)} deleted needles</small>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
}
|
||||
} else {
|
||||
<div class="text-center text-muted py-3">
|
||||
<i class="fas fa-info-circle mb-2"></i>
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -91,6 +91,8 @@ service VolumeServer {
|
||||
}
|
||||
rpc VolumeEcShardsInfo (VolumeEcShardsInfoRequest) returns (VolumeEcShardsInfoResponse) {
|
||||
}
|
||||
rpc VolumeEcDeletionInfo (VolumeEcDeletionInfoRequest) returns (VolumeEcDeletionInfoResponse) {
|
||||
}
|
||||
|
||||
// tiered storage
|
||||
rpc VolumeTierMoveDatToRemote (VolumeTierMoveDatToRemoteRequest) returns (stream VolumeTierMoveDatToRemoteResponse) {
|
||||
@@ -478,6 +480,17 @@ message EcShardInfo {
|
||||
string collection = 3;
|
||||
}
|
||||
|
||||
message VolumeEcDeletionInfoRequest {
|
||||
uint32 volume_id = 1;
|
||||
string collection = 2;
|
||||
uint32 generation = 3; // generation of EC volume, defaults to 0
|
||||
}
|
||||
message VolumeEcDeletionInfoResponse {
|
||||
uint64 deleted_bytes = 1;
|
||||
uint64 deleted_count = 2;
|
||||
repeated uint64 deleted_needle_ids = 3; // list of deleted needle IDs for debugging
|
||||
}
|
||||
|
||||
message ReadVolumeFileStatusRequest {
|
||||
uint32 volume_id = 1;
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -55,6 +55,7 @@ const (
|
||||
VolumeServer_VolumeEcBlobDelete_FullMethodName = "/volume_server_pb.VolumeServer/VolumeEcBlobDelete"
|
||||
VolumeServer_VolumeEcShardsToVolume_FullMethodName = "/volume_server_pb.VolumeServer/VolumeEcShardsToVolume"
|
||||
VolumeServer_VolumeEcShardsInfo_FullMethodName = "/volume_server_pb.VolumeServer/VolumeEcShardsInfo"
|
||||
VolumeServer_VolumeEcDeletionInfo_FullMethodName = "/volume_server_pb.VolumeServer/VolumeEcDeletionInfo"
|
||||
VolumeServer_VolumeTierMoveDatToRemote_FullMethodName = "/volume_server_pb.VolumeServer/VolumeTierMoveDatToRemote"
|
||||
VolumeServer_VolumeTierMoveDatFromRemote_FullMethodName = "/volume_server_pb.VolumeServer/VolumeTierMoveDatFromRemote"
|
||||
VolumeServer_VolumeServerStatus_FullMethodName = "/volume_server_pb.VolumeServer/VolumeServerStatus"
|
||||
@@ -108,6 +109,7 @@ type VolumeServerClient interface {
|
||||
VolumeEcBlobDelete(ctx context.Context, in *VolumeEcBlobDeleteRequest, opts ...grpc.CallOption) (*VolumeEcBlobDeleteResponse, error)
|
||||
VolumeEcShardsToVolume(ctx context.Context, in *VolumeEcShardsToVolumeRequest, opts ...grpc.CallOption) (*VolumeEcShardsToVolumeResponse, error)
|
||||
VolumeEcShardsInfo(ctx context.Context, in *VolumeEcShardsInfoRequest, opts ...grpc.CallOption) (*VolumeEcShardsInfoResponse, error)
|
||||
VolumeEcDeletionInfo(ctx context.Context, in *VolumeEcDeletionInfoRequest, opts ...grpc.CallOption) (*VolumeEcDeletionInfoResponse, error)
|
||||
// tiered storage
|
||||
VolumeTierMoveDatToRemote(ctx context.Context, in *VolumeTierMoveDatToRemoteRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[VolumeTierMoveDatToRemoteResponse], error)
|
||||
VolumeTierMoveDatFromRemote(ctx context.Context, in *VolumeTierMoveDatFromRemoteRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[VolumeTierMoveDatFromRemoteResponse], error)
|
||||
@@ -555,6 +557,16 @@ func (c *volumeServerClient) VolumeEcShardsInfo(ctx context.Context, in *VolumeE
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *volumeServerClient) VolumeEcDeletionInfo(ctx context.Context, in *VolumeEcDeletionInfoRequest, opts ...grpc.CallOption) (*VolumeEcDeletionInfoResponse, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(VolumeEcDeletionInfoResponse)
|
||||
err := c.cc.Invoke(ctx, VolumeServer_VolumeEcDeletionInfo_FullMethodName, in, out, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *volumeServerClient) VolumeTierMoveDatToRemote(ctx context.Context, in *VolumeTierMoveDatToRemoteRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[VolumeTierMoveDatToRemoteResponse], error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
stream, err := c.cc.NewStream(ctx, &VolumeServer_ServiceDesc.Streams[8], VolumeServer_VolumeTierMoveDatToRemote_FullMethodName, cOpts...)
|
||||
@@ -705,6 +717,7 @@ type VolumeServerServer interface {
|
||||
VolumeEcBlobDelete(context.Context, *VolumeEcBlobDeleteRequest) (*VolumeEcBlobDeleteResponse, error)
|
||||
VolumeEcShardsToVolume(context.Context, *VolumeEcShardsToVolumeRequest) (*VolumeEcShardsToVolumeResponse, error)
|
||||
VolumeEcShardsInfo(context.Context, *VolumeEcShardsInfoRequest) (*VolumeEcShardsInfoResponse, error)
|
||||
VolumeEcDeletionInfo(context.Context, *VolumeEcDeletionInfoRequest) (*VolumeEcDeletionInfoResponse, error)
|
||||
// tiered storage
|
||||
VolumeTierMoveDatToRemote(*VolumeTierMoveDatToRemoteRequest, grpc.ServerStreamingServer[VolumeTierMoveDatToRemoteResponse]) error
|
||||
VolumeTierMoveDatFromRemote(*VolumeTierMoveDatFromRemoteRequest, grpc.ServerStreamingServer[VolumeTierMoveDatFromRemoteResponse]) error
|
||||
@@ -834,6 +847,9 @@ func (UnimplementedVolumeServerServer) VolumeEcShardsToVolume(context.Context, *
|
||||
func (UnimplementedVolumeServerServer) VolumeEcShardsInfo(context.Context, *VolumeEcShardsInfoRequest) (*VolumeEcShardsInfoResponse, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsInfo not implemented")
|
||||
}
|
||||
func (UnimplementedVolumeServerServer) VolumeEcDeletionInfo(context.Context, *VolumeEcDeletionInfoRequest) (*VolumeEcDeletionInfoResponse, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method VolumeEcDeletionInfo not implemented")
|
||||
}
|
||||
func (UnimplementedVolumeServerServer) VolumeTierMoveDatToRemote(*VolumeTierMoveDatToRemoteRequest, grpc.ServerStreamingServer[VolumeTierMoveDatToRemoteResponse]) error {
|
||||
return status.Errorf(codes.Unimplemented, "method VolumeTierMoveDatToRemote not implemented")
|
||||
}
|
||||
@@ -1467,6 +1483,24 @@ func _VolumeServer_VolumeEcShardsInfo_Handler(srv interface{}, ctx context.Conte
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _VolumeServer_VolumeEcDeletionInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(VolumeEcDeletionInfoRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if interceptor == nil {
|
||||
return srv.(VolumeServerServer).VolumeEcDeletionInfo(ctx, in)
|
||||
}
|
||||
info := &grpc.UnaryServerInfo{
|
||||
Server: srv,
|
||||
FullMethod: VolumeServer_VolumeEcDeletionInfo_FullMethodName,
|
||||
}
|
||||
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||
return srv.(VolumeServerServer).VolumeEcDeletionInfo(ctx, req.(*VolumeEcDeletionInfoRequest))
|
||||
}
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _VolumeServer_VolumeTierMoveDatToRemote_Handler(srv interface{}, stream grpc.ServerStream) error {
|
||||
m := new(VolumeTierMoveDatToRemoteRequest)
|
||||
if err := stream.RecvMsg(m); err != nil {
|
||||
@@ -1709,6 +1743,10 @@ var VolumeServer_ServiceDesc = grpc.ServiceDesc{
|
||||
MethodName: "VolumeEcShardsInfo",
|
||||
Handler: _VolumeServer_VolumeEcShardsInfo_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "VolumeEcDeletionInfo",
|
||||
Handler: _VolumeServer_VolumeEcDeletionInfo_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "VolumeServerStatus",
|
||||
Handler: _VolumeServer_VolumeServerStatus_Handler,
|
||||
|
||||
@@ -490,6 +490,84 @@ func (vs *VolumeServer) VolumeEcBlobDelete(ctx context.Context, req *volume_serv
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// VolumeEcDeletionInfo gets deletion information for an EC volume by reading .ecj and .ecx files
|
||||
func (vs *VolumeServer) VolumeEcDeletionInfo(ctx context.Context, req *volume_server_pb.VolumeEcDeletionInfoRequest) (*volume_server_pb.VolumeEcDeletionInfoResponse, error) {
|
||||
glog.V(0).Infof("VolumeEcDeletionInfo: volume=%d, collection='%s', generation=%d", req.VolumeId, req.Collection, req.Generation)
|
||||
|
||||
resp := &volume_server_pb.VolumeEcDeletionInfoResponse{}
|
||||
|
||||
// Find the EC volume
|
||||
ecVolume, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
|
||||
if !found {
|
||||
return nil, fmt.Errorf("EC volume %d not found", req.VolumeId)
|
||||
}
|
||||
|
||||
// Validate generation if specified
|
||||
if req.Generation != 0 && req.Generation != ecVolume.Generation {
|
||||
glog.V(1).Infof("Generation mismatch for volume %d: requested %d, found %d", req.VolumeId, req.Generation, ecVolume.Generation)
|
||||
return nil, fmt.Errorf("EC volume %d generation mismatch: requested %d, found %d",
|
||||
req.VolumeId, req.Generation, ecVolume.Generation)
|
||||
}
|
||||
|
||||
// Use generation-aware filenames
|
||||
indexBaseFileName := ecVolume.IndexBaseFileName()
|
||||
glog.V(0).Infof("Volume %d: using indexBaseFileName='%s'", req.VolumeId, indexBaseFileName)
|
||||
|
||||
// Get total deleted bytes and needle IDs using existing EC functions
|
||||
var deletedBytes uint64 = 0
|
||||
var deletedCount uint64 = 0
|
||||
var deletedNeedleIds []uint64
|
||||
|
||||
// Get the total EC volume size for average needle size estimation
|
||||
totalVolumeSize := ecVolume.Size()
|
||||
glog.V(0).Infof("Volume %d: total size=%d bytes", req.VolumeId, totalVolumeSize)
|
||||
|
||||
// Read all deleted needle IDs from .ecj file
|
||||
err := erasure_coding.IterateEcjFile(indexBaseFileName, func(needleId types.NeedleId) error {
|
||||
deletedCount++
|
||||
deletedNeedleIds = append(deletedNeedleIds, uint64(needleId))
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read EC journal file for volume %d: %v", req.VolumeId, err)
|
||||
}
|
||||
|
||||
glog.V(0).Infof("Volume %d: found %d deleted needles, total volume size: %d bytes", req.VolumeId, deletedCount, totalVolumeSize)
|
||||
|
||||
// Estimate deleted bytes based on volume size and needle count
|
||||
// For EC volumes, use proportional estimation since individual needle sizes may not be available
|
||||
if deletedCount > 0 && totalVolumeSize > 0 {
|
||||
// Assume average needle size based on total data shards vs all shards ratio
|
||||
// EC volumes store original data across data shards, so estimate based on that
|
||||
dataShardSize := totalVolumeSize * uint64(erasure_coding.DataShardsCount) / uint64(erasure_coding.TotalShardsCount)
|
||||
|
||||
// Rough estimation: assume 1KB average per needle (conservative)
|
||||
// This can be improved with better heuristics in the future
|
||||
estimatedBytesPerNeedle := uint64(1024) // 1KB average
|
||||
if dataShardSize > 0 {
|
||||
// If we have data shard info, use more sophisticated estimation
|
||||
estimatedBytesPerNeedle = dataShardSize / 100 // Assume ~100 needles per data shard on average
|
||||
if estimatedBytesPerNeedle < 512 {
|
||||
estimatedBytesPerNeedle = 512 // Minimum 512 bytes per needle
|
||||
}
|
||||
}
|
||||
|
||||
deletedBytes = deletedCount * estimatedBytesPerNeedle
|
||||
glog.V(1).Infof("EC volume %d: estimated %d deleted bytes from %d needles (avg %d bytes/needle)",
|
||||
req.VolumeId, deletedBytes, deletedCount, estimatedBytesPerNeedle)
|
||||
}
|
||||
|
||||
resp.DeletedBytes = deletedBytes
|
||||
resp.DeletedCount = deletedCount
|
||||
resp.DeletedNeedleIds = deletedNeedleIds
|
||||
|
||||
glog.V(1).Infof("EC volume %d deletion info: %d deleted needles, %d deleted bytes",
|
||||
req.VolumeId, deletedCount, deletedBytes)
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// VolumeEcShardsToVolume generates the .idx, .dat files from .ecx, .ecj and .ec01 ~ .ec14 files
|
||||
func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_server_pb.VolumeEcShardsToVolumeRequest) (*volume_server_pb.VolumeEcShardsToVolumeResponse, error) {
|
||||
|
||||
|
||||
@@ -150,6 +150,12 @@ func iterateEcjFile(baseFileName string, processNeedleFn func(key types.NeedleId
|
||||
|
||||
}
|
||||
|
||||
// IterateEcjFile iterates through deleted needle IDs in an EC journal file
|
||||
// This is the public interface for reading .ecj files
|
||||
func IterateEcjFile(baseFileName string, processNeedleFn func(key types.NeedleId) error) error {
|
||||
return iterateEcjFile(baseFileName, processNeedleFn)
|
||||
}
|
||||
|
||||
// WriteDatFile generates .dat from .ec00 ~ .ec09 files
|
||||
func WriteDatFile(baseFileName string, datFileSize int64, shardFileNames []string) error {
|
||||
|
||||
|
||||
Reference in new issue
Block a user