diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 33ecf72c5..a6755dd4b 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -60,7 +60,8 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre VolumeId: req.VolumeId, }) if err != nil { - return fmt.Errorf("read volume status failed, %w", err) + glog.Warningf("failed to read source volume %d status before copy; skip record count validation: %v", req.VolumeId, err) + sourceVolumeStatus = nil } volFileInfoResp, err = client.ReadVolumeFileStatus(stream.Context(), @@ -71,18 +72,6 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return fmt.Errorf("read volume file status failed, %w", err) } - // Source is reachable and holds the volume: only now is it safe to drop - // an existing local replica before overwriting its files. - if hasExistingVolume { - glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) - // keep remote data: the inbound copy carries a .vif that may point at - // the same cloud-tier object the existing volume references. - if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true); delErr != nil { - return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) - } - glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) - } - diskType := volFileInfoResp.DiskType if req.DiskType != "" { diskType = req.DiskType @@ -96,11 +85,23 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre location := vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool { return location.DiskType == types.ToDiskType(diskType) && location.AvailableSpace.Load() > neededSpace - }) + }, needle.VolumeId(req.VolumeId)) if location == nil { return fmt.Errorf("%s %s", util.ErrVolumeNoSpaceLeft, types.ToDiskType(diskType).ReadableString()) } + // Source is reachable and a destination is reserved: only now is it + // safe to drop an existing local replica before overwriting its files. + if hasExistingVolume { + glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) + // keep remote data: the inbound copy carries a .vif that may point at + // the same cloud-tier object the existing volume references. + if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true); delErr != nil { + return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) + } + glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) + } + dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId)) indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) @@ -203,8 +204,8 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre VolumeId: req.VolumeId, }) if statusErr != nil { - glog.Warningf("failed to read source volume %d status after copy; skip record count validation: %v", req.VolumeId, statusErr) - sourceVolumeStatusAfterCopy = nil + err = fmt.Errorf("read source volume %d status after copy failed: %w", req.VolumeId, statusErr) + return err } return nil @@ -242,16 +243,14 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } shouldValidateCopyCounts := copyCountsStable(sourceVolumeStatus, sourceVolumeStatusAfterCopy) - if sourceVolumeStatusAfterCopy == nil { - glog.V(1).Infof("source volume %d status was unavailable after copy; skip record count validation", req.VolumeId) - } else if !shouldValidateCopyCounts { + if !shouldValidateCopyCounts { glog.V(1).Infof("source volume %d changed during copy; skip record count validation", req.VolumeId) } // Load and validate the volume before announcing it to the master. A failed // validation is unloaded by the store without ever making the replica // routable. - err = vs.store.MountVolumeWithValidator(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { + err = vs.store.MountVolume(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { if !shouldValidateCopyCounts { return nil } @@ -340,6 +339,9 @@ func checkCopyCounts(origin *volume_server_pb.VolumeStatusResponse, targetFileCo } func copyCountsStable(before, after *volume_server_pb.VolumeStatusResponse) bool { + // A writable source may receive writes or deletions while its files are + // copied. In that case the before/after counts do not describe one stable + // snapshot, so strict target-count validation would report a false error. return before != nil && after != nil && before.FileCount == after.FileCount && before.FileDeletedCount == after.FileDeletedCount diff --git a/weed/server/volume_grpc_copy_verify_test.go b/weed/server/volume_grpc_copy_verify_test.go index ef10832ab..2c3150f76 100644 --- a/weed/server/volume_grpc_copy_verify_test.go +++ b/weed/server/volume_grpc_copy_verify_test.go @@ -2,6 +2,11 @@ package weed_server import ( "context" + "errors" + "fmt" + "os" + "strings" + "sync/atomic" "testing" "google.golang.org/grpc" @@ -16,6 +21,191 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) +type volumeCopyStatusServer struct { + volume_server_pb.UnimplementedVolumeServerServer + delegate *VolumeServer + failStatusCall int32 + statusErr error + statusCalls atomic.Int32 +} + +func (s *volumeCopyStatusServer) VolumeStatus(ctx context.Context, req *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) { + if s.statusCalls.Add(1) == s.failStatusCall { + return nil, s.statusErr + } + return s.delegate.VolumeStatus(ctx, req) +} + +func (s *volumeCopyStatusServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) { + return s.delegate.ReadVolumeFileStatus(ctx, req) +} + +func (s *volumeCopyStatusServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) error { + return s.delegate.CopyFile(req, stream) +} + +func newVolumeCopyTestStore(t *testing.T, dir string) *storage.Store { + t.Helper() + store := storage.NewStore( + grpc.WithTransportCredentials(insecure.NewCredentials()), + "127.0.0.1", 0, 0, "", "test-store", + []string{dir}, []int32{10}, []util.MinFreeSpace{{}}, + dir, storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, [][]string{nil}, + 0, stats.DefaultDiskIOProbeConfig(), + ) + store.Locations[0].AvailableSpace.Store(^uint64(0)) + t.Cleanup(store.Close) + return store +} + +func runVolumeCopyWithStatusFailure(t *testing.T, failStatusCall int32) (error, *storage.Store) { + t.Helper() + const vid = needle.VolumeId(43) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: failStatusCall, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + return err, targetStore +} + +func TestVolumeCopyContinuesWhenInitialStatusUnavailable(t *testing.T) { + err, targetStore := runVolumeCopyWithStatusFailure(t, 1) + if err != nil { + t.Fatalf("VolumeCopy should continue when the initial status is unavailable: %v", err) + } + if targetStore.GetVolume(43) == nil { + t.Fatal("copied volume was not mounted") + } +} + +func TestVolumeCopyFailsWhenFinalStatusUnavailable(t *testing.T) { + err, targetStore := runVolumeCopyWithStatusFailure(t, 2) + if err == nil || !strings.Contains(err.Error(), "status after copy") { + t.Fatalf("VolumeCopy error = %v, want final status error", err) + } + if targetStore.GetVolume(43) != nil { + t.Fatal("volume was mounted after final status validation failed") + } + select { + case message := <-targetStore.NewVolumesChan: + t.Fatalf("volume was announced after final status validation failed: %+v", message) + default: + } + dataBaseFileName := storage.VolumeFileName(targetStore.Locations[0].Directory, "", 43) + indexBaseFileName := storage.VolumeFileName(targetStore.Locations[0].IdxDirectory, "", 43) + for _, fileName := range []string{ + dataBaseFileName + ".dat", + indexBaseFileName + ".idx", + dataBaseFileName + ".vif", + dataBaseFileName + ".note", + } { + if _, statErr := os.Stat(fileName); !os.IsNotExist(statErr) { + t.Fatalf("copy artifact %s remains after final status validation failed: %v", fileName, statErr) + } + } +} + +func TestVolumeCopyKeepsExistingReplicaWhenDestinationFull(t *testing.T) { + const vid = needle.VolumeId(44) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: 1, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := targetStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add target volume: %v", err) + } + targetStore.Locations[0].AvailableSpace.Store(0) + + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + if err == nil { + t.Fatal("VolumeCopy should fail when no destination location is available") + } + if targetStore.GetVolume(vid) == nil { + t.Fatal("existing replica was destroyed before a destination was reserved") + } +} + +func TestVolumeCopyReplacesReplicaAtSlotLimit(t *testing.T) { + const vid = needle.VolumeId(45) + + sourceStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := sourceStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add source volume: %v", err) + } + source := &volumeCopyStatusServer{ + delegate: &VolumeServer{store: sourceStore}, + failStatusCall: 1, + statusErr: errors.New("source volume status unavailable"), + } + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, source) + }) + + targetStore := newVolumeCopyTestStore(t, t.TempDir()) + if err := targetStore.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, + needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add target volume: %v", err) + } + targetStore.Locations[0].MaxVolumeCount = 1 + + target := &VolumeServer{ + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: fmt.Sprintf("127.0.0.1:%d.%d", port-10000, port), + }, &fakeVolumeCopyStream{}) + if err != nil { + t.Fatalf("VolumeCopy should replace the replica at the slot limit: %v", err) + } + if targetStore.GetVolume(vid) == nil { + t.Fatal("replaced volume was not mounted") + } +} + // fakeVolumeCopyStream is a no-op VolumeServer_VolumeCopyServer; VolumeCopy // errors out before sending anything in this test. type fakeVolumeCopyStream struct { diff --git a/weed/storage/store.go b/weed/storage/store.go index 7d891fedc..a8983bb36 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -285,8 +285,12 @@ func (s *Store) findVolume(vid needle.VolumeId) *Volume { } return nil } -func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool) (ret *DiskLocation) { +func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool, replaceVid ...needle.VolumeId) (ret *DiskLocation) { max := int32(0) + var replace needle.VolumeId + if len(replaceVid) > 0 { + replace = replaceVid[0] + } for _, location := range s.Locations { if filterFn != nil && !filterFn(location) { continue @@ -295,6 +299,11 @@ func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool) (re continue } currentFreeCount := location.MaxVolumeCount - int32(location.VolumesLen()) + if replace != 0 { + if _, found := location.FindVolume(replace); found { + currentFreeCount++ + } + } currentFreeCount *= erasure_coding.DataShardsCount currentFreeCount -= int32(location.EcShardCount()) currentFreeCount /= erasure_coding.DataShardsCount @@ -966,24 +975,23 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { return persistErr } -func (s *Store) MountVolume(i needle.VolumeId, collection *string) error { - return s.mountVolume(i, collection, nil) +// MountVolume loads a volume and announces it after all optional validators +// succeed. A validator failure unloads the volume before returning the error, +// so an invalid newly copied replica is never announced to the master. +func (s *Store) MountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { + return s.mountVolume(i, collection, validators...) } -// MountVolumeWithValidator loads a volume, validates it before announcing it -// to the master, and unloads it when validation fails. This keeps an invalid -// newly copied replica out of the master's routable volume set. -func (s *Store) MountVolumeWithValidator(i needle.VolumeId, collection *string, validator func(*Volume) error) error { - return s.mountVolume(i, collection, validator) -} - -func (s *Store) mountVolume(i needle.VolumeId, collection *string, validator func(*Volume) error) error { +func (s *Store) mountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { for diskId, location := range s.Locations { if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind, collection); found == true { glog.V(0).Infof("mount volume %d", i) v := s.findVolume(i) v.diskId = uint32(diskId) // Set disk ID when mounting - if validator != nil { + for _, validator := range validators { + if validator == nil { + continue + } if err := validator(v); err != nil { if unloadErr := location.UnloadVolume(i); unloadErr != nil { return fmt.Errorf("%w; failed to unload volume %d after validation error: %v", err, i, unloadErr) diff --git a/weed/storage/store_mount_validation_test.go b/weed/storage/store_mount_validation_test.go index cf6b16eb0..64a7d7e9a 100644 --- a/weed/storage/store_mount_validation_test.go +++ b/weed/storage/store_mount_validation_test.go @@ -13,7 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) -func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { +func TestMountVolumeValidatorAnnouncesOnlyAfterValidation(t *testing.T) { dir := t.TempDir() const vid = needle.VolumeId(17) @@ -29,7 +29,7 @@ func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { t.Cleanup(store.Close) validationErr := errors.New("copy counts differ") - err = store.MountVolumeWithValidator(vid, nil, func(*Volume) error { + err = store.MountVolume(vid, nil, func(*Volume) error { select { case <-store.NewVolumesChan: t.Fatal("volume was announced before validation completed")