mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-17 12:05:52 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8a5a27cf69 | ||
|
|
7bb2eea466 |
@@ -571,6 +571,7 @@ message ReadVolumeFileStatusResponse {
|
||||
string disk_type = 9;
|
||||
VolumeInfo volume_info = 10;
|
||||
uint32 version = 11;
|
||||
uint64 expire_at_sec = 12; // TTL expiry counted from the last write; 0 when the volume never expires
|
||||
}
|
||||
|
||||
message DiskStatus {
|
||||
|
||||
@@ -2005,6 +2005,7 @@ impl VolumeServer for VolumeGrpcService {
|
||||
disk_type: store.locations[loc_idx].disk_type.to_string(),
|
||||
volume_info: Some(vol.volume_info().clone()),
|
||||
version: vol.version().0 as u32,
|
||||
expire_at_sec: vol.expire_at_sec(),
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
@@ -16,6 +16,18 @@ import (
|
||||
|
||||
func AllocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string) {
|
||||
t.Helper()
|
||||
allocateVolume(t, client, volumeID, collection, "")
|
||||
}
|
||||
|
||||
// AllocateVolumeWithTTL allocates a volume whose needles expire after ttl
|
||||
// (e.g. "10m").
|
||||
func AllocateVolumeWithTTL(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, ttl string) {
|
||||
t.Helper()
|
||||
allocateVolume(t, client, volumeID, collection, ttl)
|
||||
}
|
||||
|
||||
func allocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, ttl string) {
|
||||
t.Helper()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
@@ -24,6 +36,7 @@ func AllocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, vo
|
||||
VolumeId: volumeID,
|
||||
Collection: collection,
|
||||
Replication: "000",
|
||||
Ttl: ttl,
|
||||
Version: uint32(needle.GetCurrentVersion()),
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -573,6 +573,7 @@ message ReadVolumeFileStatusResponse {
|
||||
string disk_type = 9;
|
||||
VolumeInfo volume_info = 10;
|
||||
uint32 version = 11;
|
||||
uint64 expire_at_sec = 12; // TTL expiry counted from the last write; 0 when the volume never expires
|
||||
}
|
||||
|
||||
message DiskStatus {
|
||||
|
||||
@@ -4678,6 +4678,7 @@ type ReadVolumeFileStatusResponse struct {
|
||||
DiskType string `protobuf:"bytes,9,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
|
||||
VolumeInfo *VolumeInfo `protobuf:"bytes,10,opt,name=volume_info,json=volumeInfo,proto3" json:"volume_info,omitempty"`
|
||||
Version uint32 `protobuf:"varint,11,opt,name=version,proto3" json:"version,omitempty"`
|
||||
ExpireAtSec uint64 `protobuf:"varint,12,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // TTL expiry counted from the last write; 0 when the volume never expires
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -4789,6 +4790,13 @@ func (x *ReadVolumeFileStatusResponse) GetVersion() uint32 {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ReadVolumeFileStatusResponse) GetExpireAtSec() uint64 {
|
||||
if x != nil {
|
||||
return x.ExpireAtSec
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type DiskStatus struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Dir string `protobuf:"bytes,1,opt,name=dir,proto3" json:"dir,omitempty"`
|
||||
@@ -7601,7 +7609,7 @@ const file_volume_server_proto_rawDesc = "" +
|
||||
"collection\x12\x1b\n" +
|
||||
"\tvolume_id\x18\x04 \x01(\rR\bvolumeId\":\n" +
|
||||
"\x1bReadVolumeFileStatusRequest\x12\x1b\n" +
|
||||
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\xe3\x03\n" +
|
||||
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\x87\x04\n" +
|
||||
"\x1cReadVolumeFileStatusResponse\x12\x1b\n" +
|
||||
"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12;\n" +
|
||||
"\x1aidx_file_timestamp_seconds\x18\x02 \x01(\x04R\x17idxFileTimestampSeconds\x12\"\n" +
|
||||
@@ -7618,7 +7626,8 @@ const file_volume_server_proto_rawDesc = "" +
|
||||
"\vvolume_info\x18\n" +
|
||||
" \x01(\v2\x1c.volume_server_pb.VolumeInfoR\n" +
|
||||
"volumeInfo\x12\x18\n" +
|
||||
"\aversion\x18\v \x01(\rR\aversion\"\xd1\x01\n" +
|
||||
"\aversion\x18\v \x01(\rR\aversion\x12\"\n" +
|
||||
"\rexpire_at_sec\x18\f \x01(\x04R\vexpireAtSec\"\xd1\x01\n" +
|
||||
"\n" +
|
||||
"DiskStatus\x12\x10\n" +
|
||||
"\x03dir\x18\x01 \x01(\tR\x03dir\x12\x10\n" +
|
||||
|
||||
@@ -536,6 +536,7 @@ func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_se
|
||||
resp.DiskType = string(v.DiskType())
|
||||
resp.VolumeInfo = v.GetVolumeInfo()
|
||||
resp.Version = uint32(v.Version())
|
||||
resp.ExpireAtSec = v.ExpireAtSec()
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -56,6 +56,10 @@ type ErasureCodingTask struct {
|
||||
// re-delete and remove the now-EC .vif those servers share.
|
||||
emptyReplicasDeleted map[string]bool
|
||||
|
||||
// expireAtSec is the source volume's TTL expiry read with the file copy;
|
||||
// it goes into the .vif so the EC volume honors the collection's TTL.
|
||||
expireAtSec uint64
|
||||
|
||||
// encodedBlockSize is the shard block layout WriteEcFiles actually encoded
|
||||
// with, read back off the EC context. Every holder must report serving the
|
||||
// same one before the source volume may be deleted.
|
||||
@@ -512,6 +516,12 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read source volume file status: %w", err)
|
||||
}
|
||||
t.expireAtSec = fileStatus.GetExpireAtSec()
|
||||
if t.expireAtSec == 0 {
|
||||
// Volume servers that predate the top-level field still carry the
|
||||
// expiry in the nested .vif once SaveVolumeInfo has stamped it.
|
||||
t.expireAtSec = fileStatus.GetVolumeInfo().GetExpireAtSec()
|
||||
}
|
||||
|
||||
t.GetLogger().WithFields(map[string]interface{}{
|
||||
"volume_id": t.volumeID,
|
||||
@@ -749,6 +759,7 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
|
||||
volumeInfo := &volume_server_pb.VolumeInfo{
|
||||
Version: uint32(needle.GetCurrentVersion()),
|
||||
EcShardConfig: ecShardConfig,
|
||||
ExpireAtSec: t.expireAtSec,
|
||||
}
|
||||
// The decoder resolves the shard block layout from the encode-time .dat
|
||||
// size; without it, decoding falls back to inferring the layout from the
|
||||
|
||||
@@ -168,6 +168,7 @@ func TestGenerateEcShardsLocallyStampsEncodeIdentity(t *testing.T) {
|
||||
require.Greater(t, vi.EcShardConfig.DataShards, uint32(0))
|
||||
require.Greater(t, vi.EcShardConfig.ParityShards, uint32(0))
|
||||
require.NotZero(t, vi.EcShardConfig.EncodeTsNs, "worker-encoded .vif must carry an encode identity")
|
||||
require.Zero(t, vi.ExpireAtSec, "a source with no TTL must not gain an expiry")
|
||||
}
|
||||
|
||||
// A fenced run must stamp the admin-issued generation (t.encodeTsNs) verbatim into
|
||||
@@ -218,6 +219,56 @@ func TestGenerateEcShardsLocallyUsesAdminGeneration(t *testing.T) {
|
||||
require.Equal(t, adminGeneration, vi.EcShardConfig.EncodeTsNs, "fenced run must stamp the admin generation verbatim, not a local timestamp")
|
||||
}
|
||||
|
||||
// The worker-written .vif must carry the source volume's TTL expiry, or the EC
|
||||
// volume never becomes eligible for deletion while the shell ec.encode path does.
|
||||
func TestGenerateEcShardsLocallyCarriesExpiry(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test in short mode")
|
||||
}
|
||||
|
||||
clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
|
||||
conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
|
||||
defer conn.Close()
|
||||
|
||||
const volumeID = uint32(958)
|
||||
framework.AllocateVolumeWithTTL(t, grpcClient, volumeID, "", "10m")
|
||||
|
||||
httpClient := framework.NewHTTPClient()
|
||||
fid := framework.NewFileID(volumeID, 4001, 0x6666FFFF)
|
||||
resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("payload-for-ec-expiry"))
|
||||
_ = framework.ReadAllAndClose(t, resp)
|
||||
require.Equal(t, http.StatusCreated, resp.StatusCode)
|
||||
|
||||
task := NewErasureCodingTask(
|
||||
"ec-carry-expiry",
|
||||
clusterHarness.VolumeServerAddress(),
|
||||
volumeID,
|
||||
"",
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
require.NoError(t, task.markReplicasReadonly(ctx))
|
||||
|
||||
fileStatus, err := task.readSourceVolumeFileStatus(ctx)
|
||||
require.NoError(t, err)
|
||||
want := fileStatus.GetExpireAtSec()
|
||||
require.NotZero(t, want, "a TTL'd source must report a nonzero expiry")
|
||||
|
||||
localFiles, err := task.copyVolumeFilesToWorker(ctx, t.TempDir())
|
||||
require.NoError(t, err)
|
||||
|
||||
shardFiles, err := task.generateEcShardsLocally(localFiles, t.TempDir())
|
||||
require.NoError(t, err)
|
||||
|
||||
vi, _, found, err := volume_info.MaybeLoadVolumeInfo(shardFiles["vif"])
|
||||
require.NoError(t, err)
|
||||
require.True(t, found)
|
||||
require.Equal(t, want, vi.ExpireAtSec, "worker .vif must carry the source volume's TTL expiry")
|
||||
}
|
||||
|
||||
func compactVolumeOnce(t *testing.T, grpcClient volume_server_pb.VolumeServerClient, volumeID uint32) {
|
||||
t.Helper()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user