shell: add volume.configure.replication to change replication for a volume

fix https://github.com/chrislusf/seaweedfs/issues/1192

1 files changed, 37 insertions, 0 deletions

diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go
index c631d2535..43987b748 100644
--- a/weed/server/volume_grpc_admin.go
+++ b/weed/server/volume_grpc_admin.go
@@ -2,10 +2,12 @@ package weed_server
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/storage/super_block"
 )
 
 func (vs *VolumeServer) DeleteCollection(ctx context.Context, req *volume_server_pb.DeleteCollectionRequest) (*volume_server_pb.DeleteCollectionResponse, error) {
@@ -96,6 +98,41 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.
 
 }
 
+func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) {
+
+	resp := &volume_server_pb.VolumeConfigureResponse{}
+
+	// check replication format
+	if _, err := super_block.NewReplicaPlacementFromString(req.Replication); err != nil {
+		resp.Error = fmt.Sprintf("volume configure replication %v: %v", req, err)
+		return resp, nil
+	}
+
+	// unmount
+	if err := vs.store.UnmountVolume(needle.VolumeId(req.VolumeId)); err != nil {
+		glog.Errorf("volume configure unmount %v: %v", req, err)
+		resp.Error = fmt.Sprintf("volume configure unmount %v: %v", req, err)
+		return resp, nil
+	}
+
+	// modify the volume info file
+	if err := vs.store.ConfigureVolume(needle.VolumeId(req.VolumeId), req.Replication); err != nil {
+		glog.Errorf("volume configure %v: %v", req, err)
+		resp.Error = fmt.Sprintf("volume configure %v: %v", req, err)
+		return resp, nil
+	}
+
+	// mount
+	if err := vs.store.MountVolume(needle.VolumeId(req.VolumeId)); err != nil {
+		glog.Errorf("volume configure mount %v: %v", req, err)
+		resp.Error = fmt.Sprintf("volume configure mount %v: %v", req, err)
+		return resp, nil
+	}
+
+	return resp, nil
+
+}
+
 func (vs *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_server_pb.VolumeMarkReadonlyRequest) (*volume_server_pb.VolumeMarkReadonlyResponse, error) {
 
 	resp := &volume_server_pb.VolumeMarkReadonlyResponse{}