diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index a33be5a42..dcb039621 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -50,11 +50,20 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre // send .dat file // confirm size and timestamp var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse + var sourceVolumeStatus *volume_server_pb.VolumeStatusResponse + var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string var hasRemoteDatFile bool err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { var err error - volFileInfoResp, err = client.ReadVolumeFileStatus(context.Background(), + sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{ + VolumeId: req.VolumeId, + }) + if err != nil { + return fmt.Errorf("read volume status failed, %w", err) + } + + volFileInfoResp, err = client.ReadVolumeFileStatus(stream.Context(), &volume_server_pb.ReadVolumeFileStatusRequest{ VolumeId: req.VolumeId, }) @@ -189,6 +198,15 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return fmt.Errorf("remove .note for volume %d: %w", req.VolumeId, noteErr) } + var statusErr error + sourceVolumeStatusAfterCopy, statusErr = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{ + 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 + } + return nil }) @@ -223,10 +241,24 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } } - // mount the volume - err = vs.store.MountVolume(needle.VolumeId(req.VolumeId)) + 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 { + 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), func(targetVolume *storage.Volume) error { + if !shouldValidateCopyCounts { + return nil + } + return checkCopyCounts(sourceVolumeStatusAfterCopy, targetVolume.FileCount(), targetVolume.DeletedCount()) + }) if err != nil { - return fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err) + return fmt.Errorf("failed to mount or validate volume %d: %w", req.VolumeId, err) } if err = stream.Send(&volume_server_pb.VolumeCopyResponse{ @@ -266,11 +298,8 @@ func (vs *VolumeServer) doCopyFileWithThrottler(client volume_server_pb.VolumeSe } -/* -* -only check the differ of the file size -todo: maybe should check the received count and deleted count of the volume -*/ +// checkCopyFiles verifies the copied file sizes. Record counts are checked +// after the target volume is mounted, when the target needle map is available. func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse, hasRemoteDatFile bool, idxFileName, datFileName string) error { stat, err := os.Stat(idxFileName) if err != nil { @@ -300,6 +329,22 @@ func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse return nil } +func checkCopyCounts(origin *volume_server_pb.VolumeStatusResponse, targetFileCount, targetDeletedCount uint64) error { + if origin.FileCount != targetFileCount { + return fmt.Errorf("target file count [%d] is not same as origin file count [%d]", targetFileCount, origin.FileCount) + } + if origin.FileDeletedCount != targetDeletedCount { + return fmt.Errorf("target deleted count [%d] is not same as origin deleted count [%d]", targetDeletedCount, origin.FileDeletedCount) + } + return nil +} + +func copyCountsStable(before, after *volume_server_pb.VolumeStatusResponse) bool { + return before != nil && after != nil && + before.FileCount == after.FileCount && + before.FileDeletedCount == after.FileDeletedCount +} + func findLastAppendAtNsFromCopiedFiles(idxFileName, datFileName string, version needle.Version) (uint64, error) { if version < needle.Version3 { return 0, nil diff --git a/weed/server/volume_grpc_copy_count_test.go b/weed/server/volume_grpc_copy_count_test.go new file mode 100644 index 000000000..5c3d01be2 --- /dev/null +++ b/weed/server/volume_grpc_copy_count_test.go @@ -0,0 +1,148 @@ +package weed_server + +import ( + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" +) + +func TestCheckCopyCounts(t *testing.T) { + tests := []struct { + name string + sourceFileCount uint64 + sourceDeleteCount uint64 + targetFileCount uint64 + targetDeleteCount uint64 + wantErr string + }{ + { + name: "empty volume counts match", + }, + { + name: "file and deleted counts match", + sourceFileCount: 10, + sourceDeleteCount: 3, + targetFileCount: 10, + targetDeleteCount: 3, + }, + { + name: "file count differs", + sourceFileCount: 10, + sourceDeleteCount: 3, + targetFileCount: 9, + targetDeleteCount: 3, + wantErr: "file count", + }, + { + name: "deleted count differs", + sourceFileCount: 10, + sourceDeleteCount: 3, + targetFileCount: 10, + targetDeleteCount: 2, + wantErr: "deleted count", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := checkCopyCounts( + &volume_server_pb.VolumeStatusResponse{ + FileCount: tt.sourceFileCount, + FileDeletedCount: tt.sourceDeleteCount, + }, + tt.targetFileCount, + tt.targetDeleteCount, + ) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("checkCopyCounts() error = %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("checkCopyCounts() error = %v, want substring %q", err, tt.wantErr) + } + }) + } +} + +func TestCopyCountsStable(t *testing.T) { + tests := []struct { + name string + before *volume_server_pb.VolumeStatusResponse + after *volume_server_pb.VolumeStatusResponse + want bool + }{ + { + name: "empty volume is stable", + before: &volume_server_pb.VolumeStatusResponse{}, + after: &volume_server_pb.VolumeStatusResponse{}, + want: true, + }, + { + name: "matching file and deleted counts are stable", + before: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + }, + after: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + }, + want: true, + }, + { + name: "read-only state changes without count changes are stable", + before: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + IsReadOnly: true, + }, + after: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + IsReadOnly: false, + }, + want: true, + }, + { + name: "changed file count is not stable", + before: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + }, + after: &volume_server_pb.VolumeStatusResponse{ + FileCount: 9, + FileDeletedCount: 3, + }, + }, + { + name: "changed deleted count is not stable", + before: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 3, + }, + after: &volume_server_pb.VolumeStatusResponse{ + FileCount: 10, + FileDeletedCount: 2, + }, + }, + { + name: "missing before status is not stable", + after: &volume_server_pb.VolumeStatusResponse{}, + }, + { + name: "missing after status is not stable", + before: &volume_server_pb.VolumeStatusResponse{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := copyCountsStable(tt.before, tt.after); got != tt.want { + t.Fatalf("copyCountsStable() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/weed/storage/store.go b/weed/storage/store.go index af005f3d2..3b483d802 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -967,11 +967,30 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { } func (s *Store) MountVolume(i needle.VolumeId) error { + return s.mountVolume(i, nil) +} + +// 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, validator func(*Volume) error) error { + return s.mountVolume(i, validator) +} + +func (s *Store) mountVolume(i needle.VolumeId, validator func(*Volume) error) error { for diskId, location := range s.Locations { if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind); 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 { + 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) + } + return err + } + } readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), diff --git a/weed/storage/store_mount_validation_test.go b/weed/storage/store_mount_validation_test.go new file mode 100644 index 000000000..d2de1369b --- /dev/null +++ b/weed/storage/store_mount_validation_test.go @@ -0,0 +1,48 @@ +package storage + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { + dir := t.TempDir() + const vid = needle.VolumeId(17) + + volume, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory, + &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + volume.Close() + + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, + stats.DefaultDiskIOProbeConfig()) + t.Cleanup(store.Close) + + validationErr := errors.New("copy counts differ") + err = store.MountVolumeWithValidator(vid, func(*Volume) error { + select { + case <-store.NewVolumesChan: + t.Fatal("volume was announced before validation completed") + default: + } + return validationErr + }) + + require.ErrorIs(t, err, validationErr) + require.Nil(t, store.GetVolume(vid)) + select { + case message := <-store.NewVolumesChan: + t.Fatalf("volume was announced after validation failed: %+v", message) + default: + } +}