aboutsummaryrefslogtreecommitdiff
path: root/weed/server
diff options
context:
space:
mode:
Diffstat (limited to 'weed/server')
-rw-r--r--weed/server/volume_grpc_admin.go8
-rw-r--r--weed/server/volume_grpc_replicate.go155
-rw-r--r--weed/server/volume_grpc_sync.go20
-rw-r--r--weed/server/volume_grpc_vacuum.go22
4 files changed, 180 insertions, 25 deletions
diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go
index 429ca9b68..c924b7a62 100644
--- a/weed/server/volume_grpc_admin.go
+++ b/weed/server/volume_grpc_admin.go
@@ -29,7 +29,7 @@ func (vs *VolumeServer) AssignVolume(ctx context.Context, req *volume_server_pb.
resp := &volume_server_pb.AssignVolumeResponse{}
err := vs.store.AddVolume(
- storage.VolumeId(req.VolumdId),
+ storage.VolumeId(req.VolumeId),
req.Collection,
vs.needleMapKind,
req.Replication,
@@ -51,7 +51,7 @@ func (vs *VolumeServer) VolumeMount(ctx context.Context, req *volume_server_pb.V
resp := &volume_server_pb.VolumeMountResponse{}
- err := vs.store.MountVolume(storage.VolumeId(req.VolumdId))
+ err := vs.store.MountVolume(storage.VolumeId(req.VolumeId))
if err != nil {
glog.Errorf("volume mount %v: %v", req, err)
@@ -67,7 +67,7 @@ func (vs *VolumeServer) VolumeUnmount(ctx context.Context, req *volume_server_pb
resp := &volume_server_pb.VolumeUnmountResponse{}
- err := vs.store.UnmountVolume(storage.VolumeId(req.VolumdId))
+ err := vs.store.UnmountVolume(storage.VolumeId(req.VolumeId))
if err != nil {
glog.Errorf("volume unmount %v: %v", req, err)
@@ -83,7 +83,7 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.
resp := &volume_server_pb.VolumeDeleteResponse{}
- err := vs.store.DeleteVolume(storage.VolumeId(req.VolumdId))
+ err := vs.store.DeleteVolume(storage.VolumeId(req.VolumeId))
if err != nil {
glog.Errorf("volume delete %v: %v", req, err)
diff --git a/weed/server/volume_grpc_replicate.go b/weed/server/volume_grpc_replicate.go
new file mode 100644
index 000000000..20a85fd6f
--- /dev/null
+++ b/weed/server/volume_grpc_replicate.go
@@ -0,0 +1,155 @@
+package weed_server
+
+import (
+ "context"
+ "fmt"
+ "github.com/chrislusf/seaweedfs/weed/operation"
+ "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
+ "github.com/chrislusf/seaweedfs/weed/storage"
+ "io"
+ "os"
+)
+
+func (vs *VolumeServer) ReplicateVolume(ctx context.Context, req *volume_server_pb.ReplicateVolumeRequest) (*volume_server_pb.ReplicateVolumeResponse, error) {
+
+ v := vs.store.GetVolume(storage.VolumeId(req.VolumeId))
+ if v != nil {
+ // unmount the volume
+ err := vs.store.UnmountVolume(storage.VolumeId(req.VolumeId))
+ if err != nil {
+ return nil, fmt.Errorf("failed to unmount volume %d: %v", req.VolumeId, err)
+ }
+ }
+
+ location := vs.store.FindFreeLocation()
+ if location == nil {
+ return nil, fmt.Errorf("no space left")
+ }
+
+ volumeFileName := storage.VolumeFileName(req.Collection, location.Directory, int(req.VolumeId))
+
+ // the master will not start compaction for read-only volumes, so it is safe to just copy files directly
+ // copy .dat and .idx files
+ // read .idx .dat file size and timestamp
+ // send .idx file
+ // send .dat file
+ // confirm size and timestamp
+
+ err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+
+ // TODO read file sizes before copying
+ client.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{})
+
+ copyFileClient, err := client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
+ VolumeId: req.VolumeId,
+ IsIdxFile: true,
+ })
+
+ if err != nil {
+ return fmt.Errorf("failed to start copying volume %d idx file: %v", req.VolumeId, err)
+ }
+
+ err = writeToFile(copyFileClient, volumeFileName+".idx")
+ if err != nil {
+ return fmt.Errorf("failed to copy volume %d idx file: %v", req.VolumeId, err)
+ }
+
+ copyFileClient, err = client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
+ VolumeId: req.VolumeId,
+ IsDatFile: true,
+ })
+
+ if err != nil {
+ return fmt.Errorf("failed to start copying volume %d dat file: %v", req.VolumeId, err)
+ }
+
+ err = writeToFile(copyFileClient, volumeFileName+".dat")
+ if err != nil {
+ return fmt.Errorf("failed to copy volume %d dat file: %v", req.VolumeId, err)
+ }
+
+ return nil
+ })
+
+ if err != nil {
+ return nil, err
+ }
+
+ // TODO: check the timestamp and size
+
+ // mount the volume
+ err = vs.store.MountVolume(storage.VolumeId(req.VolumeId))
+ if err != nil {
+ return nil, fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
+ }
+
+ return &volume_server_pb.ReplicateVolumeResponse{}, err
+
+}
+
+func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName string) error {
+ println("writing to ", fileName)
+ dst, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
+ if err != nil {
+ return nil
+ }
+ defer dst.Close()
+
+ for {
+ resp, receiveErr := client.Recv()
+ if receiveErr == io.EOF {
+ break
+ }
+ if receiveErr != nil {
+ return fmt.Errorf("receiving %s: %v", fileName, receiveErr)
+ }
+ dst.Write(resp.FileContent)
+ }
+ return nil
+}
+
+func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) {
+ resp := &volume_server_pb.ReadVolumeFileStatusResponse{}
+ return resp, nil
+}
+
+func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) (error) {
+
+ v := vs.store.GetVolume(storage.VolumeId(req.VolumeId))
+ if v == nil {
+ return fmt.Errorf("not found volume id %d", req.VolumeId)
+ }
+
+ const BufferSize = 1024 * 16
+ var fileName = v.FileName()
+ if req.IsDatFile {
+ fileName += ".dat"
+ } else if req.IsIdxFile {
+ fileName += ".idx"
+ }
+ file, err := os.Open(fileName)
+ if err != nil {
+ return err
+ }
+ defer file.Close()
+
+ buffer := make([]byte, BufferSize)
+
+ for {
+ bytesread, err := file.Read(buffer)
+
+ if err != nil {
+ if err != io.EOF {
+ return err
+ }
+ break
+ }
+
+ stream.Send(&volume_server_pb.CopyFileResponse{
+ FileContent: buffer[:bytesread],
+ })
+
+ }
+
+ return nil
+}
diff --git a/weed/server/volume_grpc_sync.go b/weed/server/volume_grpc_sync.go
index 0114b38a4..971258689 100644
--- a/weed/server/volume_grpc_sync.go
+++ b/weed/server/volume_grpc_sync.go
@@ -12,14 +12,14 @@ import (
func (vs *VolumeServer) VolumeSyncStatus(ctx context.Context, req *volume_server_pb.VolumeSyncStatusRequest) (*volume_server_pb.VolumeSyncStatusResponse, error) {
- v := vs.store.GetVolume(storage.VolumeId(req.VolumdId))
+ v := vs.store.GetVolume(storage.VolumeId(req.VolumeId))
if v == nil {
- return nil, fmt.Errorf("not found volume id %d", req.VolumdId)
+ return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
}
resp := v.GetVolumeSyncStatus()
- glog.V(2).Infof("volume sync status %d", req.VolumdId)
+ glog.V(2).Infof("volume sync status %d", req.VolumeId)
return resp, nil
@@ -27,17 +27,17 @@ func (vs *VolumeServer) VolumeSyncStatus(ctx context.Context, req *volume_server
func (vs *VolumeServer) VolumeSyncIndex(req *volume_server_pb.VolumeSyncIndexRequest, stream volume_server_pb.VolumeServer_VolumeSyncIndexServer) error {
- v := vs.store.GetVolume(storage.VolumeId(req.VolumdId))
+ v := vs.store.GetVolume(storage.VolumeId(req.VolumeId))
if v == nil {
- return fmt.Errorf("not found volume id %d", req.VolumdId)
+ return fmt.Errorf("not found volume id %d", req.VolumeId)
}
content, err := v.IndexFileContent()
if err != nil {
- glog.Errorf("sync volume %d index: %v", req.VolumdId, err)
+ glog.Errorf("sync volume %d index: %v", req.VolumeId, err)
} else {
- glog.V(2).Infof("sync volume %d index", req.VolumdId)
+ glog.V(2).Infof("sync volume %d index", req.VolumeId)
}
const blockSizeLimit = 1024 * 1024 * 2
@@ -57,9 +57,9 @@ func (vs *VolumeServer) VolumeSyncIndex(req *volume_server_pb.VolumeSyncIndexReq
func (vs *VolumeServer) VolumeSyncData(req *volume_server_pb.VolumeSyncDataRequest, stream volume_server_pb.VolumeServer_VolumeSyncDataServer) error {
- v := vs.store.GetVolume(storage.VolumeId(req.VolumdId))
+ v := vs.store.GetVolume(storage.VolumeId(req.VolumeId))
if v == nil {
- return fmt.Errorf("not found volume id %d", req.VolumdId)
+ return fmt.Errorf("not found volume id %d", req.VolumeId)
}
if uint32(v.SuperBlock.CompactRevision) != req.Revision {
@@ -82,7 +82,7 @@ func (vs *VolumeServer) VolumeSyncData(req *volume_server_pb.VolumeSyncDataReque
}
if err != nil {
- glog.Errorf("sync volume %d data: %v", req.VolumdId, err)
+ glog.Errorf("sync volume %d data: %v", req.VolumeId, err)
}
const blockSizeLimit = 1024 * 1024 * 2
diff --git a/weed/server/volume_grpc_vacuum.go b/weed/server/volume_grpc_vacuum.go
index f0c87b582..d31b8f8e7 100644
--- a/weed/server/volume_grpc_vacuum.go
+++ b/weed/server/volume_grpc_vacuum.go
@@ -12,12 +12,12 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve
resp := &volume_server_pb.VacuumVolumeCheckResponse{}
- garbageRatio, err := vs.store.CheckCompactVolume(storage.VolumeId(req.VolumdId))
+ garbageRatio, err := vs.store.CheckCompactVolume(storage.VolumeId(req.VolumeId))
resp.GarbageRatio = garbageRatio
if err != nil {
- glog.V(3).Infof("check volume %d: %v", req.VolumdId, err)
+ glog.V(3).Infof("check volume %d: %v", req.VolumeId, err)
}
return resp, err
@@ -28,12 +28,12 @@ func (vs *VolumeServer) VacuumVolumeCompact(ctx context.Context, req *volume_ser
resp := &volume_server_pb.VacuumVolumeCompactResponse{}
- err := vs.store.CompactVolume(storage.VolumeId(req.VolumdId), req.Preallocate)
+ err := vs.store.CompactVolume(storage.VolumeId(req.VolumeId), req.Preallocate)
if err != nil {
- glog.Errorf("compact volume %d: %v", req.VolumdId, err)
+ glog.Errorf("compact volume %d: %v", req.VolumeId, err)
} else {
- glog.V(1).Infof("compact volume %d", req.VolumdId)
+ glog.V(1).Infof("compact volume %d", req.VolumeId)
}
return resp, err
@@ -44,12 +44,12 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv
resp := &volume_server_pb.VacuumVolumeCommitResponse{}
- err := vs.store.CommitCompactVolume(storage.VolumeId(req.VolumdId))
+ err := vs.store.CommitCompactVolume(storage.VolumeId(req.VolumeId))
if err != nil {
- glog.Errorf("commit volume %d: %v", req.VolumdId, err)
+ glog.Errorf("commit volume %d: %v", req.VolumeId, err)
} else {
- glog.V(1).Infof("commit volume %d", req.VolumdId)
+ glog.V(1).Infof("commit volume %d", req.VolumeId)
}
return resp, err
@@ -60,12 +60,12 @@ func (vs *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_ser
resp := &volume_server_pb.VacuumVolumeCleanupResponse{}
- err := vs.store.CommitCleanupVolume(storage.VolumeId(req.VolumdId))
+ err := vs.store.CommitCleanupVolume(storage.VolumeId(req.VolumeId))
if err != nil {
- glog.Errorf("cleanup volume %d: %v", req.VolumdId, err)
+ glog.Errorf("cleanup volume %d: %v", req.VolumeId, err)
} else {
- glog.V(1).Infof("cleanup volume %d", req.VolumdId)
+ glog.V(1).Infof("cleanup volume %d", req.VolumeId)
}
return resp, err
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 09:48:07 +0000