Volume server add `/admin/setting` interface, support set volume replication and readonly

go/weed/weed_server/master_server_handlers_admin.go

@@ -191,7 +191,8 @@ func (ms *MasterServer) getVolumeGrowOption(r *http.Request) (*topology.VolumeGr
 //only proxy to each volume server
 func (ms *MasterServer) setReplicaHandler(w http.ResponseWriter, r *http.Request) {
 	r.ParseForm()
-	if _, e := storage.NewReplicaPlacementFromString(r.FormValue("replication")); e != nil {
+	replicationValue := r.FormValue("replication")
+	if _, e := storage.NewReplicaPlacementFromString(replicationValue); e != nil {
 		writeJsonError(w, r, http.StatusBadRequest, e)
 		return
 	}
@@ -200,14 +201,27 @@ func (ms *MasterServer) setReplicaHandler(w http.ResponseWriter, r *http.Request
 		writeJsonError(w, r, http.StatusBadRequest, errors.New("No available agrs found."))
 		return
 	}
-	result := make(map[string]interface{})
-	forms := r.Form
+	result := ms.batchSetVolumeOption("replication", replicationValue, r.Form["volume"], r.Form["collection"])
+	writeJson(w, r, http.StatusOK, result)
+}
+
+func (ms *MasterServer) batchSetVolumeOption(settingKey, settingValue string, volumes, collections []string)(result map[string]interface{}){
+	forms := url.Values{}
+	forms.Set("key", settingKey)
+	forms.Set("value", settingValue)
+	if len(volumes) == 0 && len(collections) == 0 {
+		forms.Set("all", "true")
+	}else{
+		forms["volume"] = volumes
+		forms["collection"] = collections
+	}
+
 	var wg sync.WaitGroup
 	ms.Topo.WalkDataNode(func(dn *topology.DataNode) (e error) {
 		wg.Add(1)
 		go func(server string, values url.Values) {
 			defer wg.Done()
-			jsonBlob, e := util.Post("http://"+server+"/admin/set_replica", values)
+			jsonBlob, e := util.Post("http://"+server+"/admin/setting", values)
 			if e != nil {
 				result[server] = map[string]interface{}{
 					"error": e.Error() + " " + string(jsonBlob),
@@ -225,5 +239,6 @@ func (ms *MasterServer) setReplicaHandler(w http.ResponseWriter, r *http.Request
 		return nil
 	})
 	wg.Wait()
-	writeJson(w, r, http.StatusOK, result)
+	return
 }
+

go/weed/weed_server/volume_server.go

@@ -53,12 +53,12 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	adminMux.HandleFunc("/admin/vacuum/check", vs.guard.WhiteList(vs.vacuumVolumeCheckHandler))
 	adminMux.HandleFunc("/admin/vacuum/compact", vs.guard.WhiteList(vs.vacuumVolumeCompactHandler))
 	adminMux.HandleFunc("/admin/vacuum/commit", vs.guard.WhiteList(vs.vacuumVolumeCommitHandler))
+	adminMux.HandleFunc("/admin/setting", vs.guard.WhiteList(vs.setVolumeOptionHandler))
 	adminMux.HandleFunc("/admin/delete_collection", vs.guard.WhiteList(vs.deleteCollectionHandler))
 	adminMux.HandleFunc("/admin/sync/status", vs.guard.WhiteList(vs.getVolumeSyncStatusHandler))
 	adminMux.HandleFunc("/admin/sync/index", vs.guard.WhiteList(vs.getVolumeIndexContentHandler))
 	adminMux.HandleFunc("/admin/sync/data", vs.guard.WhiteList(vs.getVolumeDataContentHandler))
 	adminMux.HandleFunc("/admin/sync/vol_data", vs.guard.WhiteList(vs.getVolumeCleanDataHandler))
-	adminMux.HandleFunc("/admin/set_replica", vs.guard.WhiteList(vs.setVolumeReplicaHandler))
 	adminMux.HandleFunc("/stats/counter", vs.guard.WhiteList(statsCounterHandler))
 	adminMux.HandleFunc("/stats/memory", vs.guard.WhiteList(statsMemoryHandler))
 	adminMux.HandleFunc("/stats/disk", vs.guard.WhiteList(vs.statsDiskHandler))

go/weed/weed_server/volume_server_handlers_admin.go

@@ -1,14 +1,23 @@
 package weed_server
 
 import (
+	"errors"
 	"net/http"
 	"path/filepath"
+	"strconv"
+	"strings"
 
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/stats"
+	"github.com/chrislusf/seaweedfs/go/storage"
 	"github.com/chrislusf/seaweedfs/go/util"
 )
 
+type VolumeOptError struct {
+	Volume string `json:"volume"`
+	Err    string `json:"err"`
+}
+
 func (vs *VolumeServer) statusHandler(w http.ResponseWriter, r *http.Request) {
 	m := make(map[string]interface{})
 	m["Version"] = util.VERSION
@@ -48,3 +57,80 @@ func (vs *VolumeServer) statsDiskHandler(w http.ResponseWriter, r *http.Request)
 	m["DiskStatuses"] = ds
 	writeJsonQuiet(w, r, http.StatusOK, m)
 }
+
+func (vs *VolumeServer) setVolumeOptionHandler(w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	errs := []VolumeOptError{}
+	var (
+		setter storage.VolumeWalker
+	)
+
+	key := r.FormValue("key")
+	value := r.FormValue("value")
+	if key == "readonly" {
+		isReadOnly, e := strconv.ParseBool(value)
+		if e != nil {
+			writeJsonError(w, r, http.StatusBadRequest, e)
+			return
+		}
+		setter = func(v *storage.Volume) error {
+			if e := v.SetReadOnly(isReadOnly); e != nil {
+				errs = append(errs, VolumeOptError{
+					Volume: v.Id.String(),
+					Err:    e.Error(),
+				})
+			}
+			return nil
+		}
+	} else if key == "replication" {
+		replica, e := storage.NewReplicaPlacementFromString(r.FormValue(value))
+		if e != nil {
+			writeJsonError(w, r, http.StatusBadRequest, e)
+			return
+		}
+		setter = func(v *storage.Volume) error {
+			if e := v.SetReplica(replica); e != nil {
+				errs = append(errs, VolumeOptError{
+					Volume: v.Id.String(),
+					Err:    e.Error(),
+				})
+			}
+			return nil
+		}
+	} else {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("Unkonw setting: "+key))
+		return
+	}
+
+	all, _ := strconv.ParseBool(r.FormValue("all"))
+	if all {
+		vs.store.WalkVolume(setter)
+	} else {
+		volumesSet := make(map[string]bool)
+		for _, volume := range r.Form["volume"] {
+			volumesSet[strings.TrimSpace(volume)] = true
+		}
+		collectionsSet := make(map[string]bool)
+		for _, c := range r.Form["collection"] {
+			collectionsSet[strings.TrimSpace(c)] = true
+		}
+		if len(collectionsSet) > 0 || len(volumesSet) > 0 {
+			vs.store.WalkVolume(func(v *storage.Volume) (e error) {
+				if !collectionsSet[v.Collection] && !volumesSet[v.Id.String()] {
+					return nil
+				}
+				setter(v)
+				return nil
+			})
+		}
+
+	}
+
+	result := make(map[string]interface{})
+	if len(errs) > 0 {
+		result["error"] = "set volume replica error."
+		result["errors"] = errs
+	}
+
+	writeJson(w, r, http.StatusAccepted, result)
+}

go/weed/weed_server/volume_server_handlers_replicate.go

@@ -1,130 +0,0 @@
-package weed_server
-
-import (
-	"fmt"
-	"io"
-	"net/http"
-	"strconv"
-
-	"github.com/chrislusf/seaweedfs/go/glog"
-	"github.com/chrislusf/seaweedfs/go/storage"
-	"github.com/pierrec/lz4"
-	"strings"
-)
-
-func (vs *VolumeServer) getVolumeCleanDataHandler(w http.ResponseWriter, r *http.Request) {
-	v, e := vs.getVolume("volume", r)
-	if v == nil {
-		http.Error(w, fmt.Sprintf("Not Found volume: %v", e), http.StatusBadRequest)
-		return
-	}
-	//set read only when replicating
-	v.SetReadOnly(true)
-	defer v.SetReadOnly(false)
-	cr, e := v.GetVolumeCleanReader()
-	if e != nil {
-		http.Error(w, fmt.Sprintf("Get volume clean reader: %v", e), http.StatusInternalServerError)
-		return
-	}
-	totalSize, e := cr.Size()
-	if e != nil {
-		http.Error(w, fmt.Sprintf("Get volume size: %v", e), http.StatusInternalServerError)
-		return
-	}
-	w.Header().Set("Accept-Ranges", "bytes")
-	w.Header().Set("Content-Disposition", fmt.Sprintf(`filename="%d.dat.lz4"`, v.Id))
-
-	rangeReq := r.Header.Get("Range")
-	if rangeReq == "" {
-		w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
-		w.Header().Set("Content-Encoding", "lz4")
-		lz4w := lz4.NewWriter(w)
-		if _, e = io.Copy(lz4w, cr); e != nil {
-			glog.V(4).Infoln("response write error:", e)
-		}
-		lz4w.Close()
-		return
-	}
-	ranges, e := parseRange(rangeReq, totalSize)
-	if e != nil {
-		http.Error(w, e.Error(), http.StatusRequestedRangeNotSatisfiable)
-		return
-	}
-	if len(ranges) != 1 {
-		http.Error(w, "Only support one range", http.StatusNotImplemented)
-		return
-	}
-	ra := ranges[0]
-	if _, e := cr.Seek(ra.start, 0); e != nil {
-		http.Error(w, e.Error(), http.StatusInternalServerError)
-		return
-	}
-	w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10))
-	w.Header().Set("Content-Range", ra.contentRange(totalSize))
-	w.Header().Set("Content-Encoding", "lz4")
-	w.WriteHeader(http.StatusPartialContent)
-	lz4w := lz4.NewWriter(w)
-	if _, e = io.CopyN(lz4w, cr, ra.length); e != nil {
-		glog.V(2).Infoln("response write error:", e)
-	}
-	lz4w.Close()
-}
-
-type VolumeOptError struct {
-	Volume string `json:"volume"`
-	Err    string `json:"err"`
-}
-
-func (vs *VolumeServer) setVolumeReplicaHandler(w http.ResponseWriter, r *http.Request) {
-	r.ParseForm()
-	replica, e := storage.NewReplicaPlacementFromString(r.FormValue("replication"))
-	if e != nil {
-		writeJsonError(w, r, http.StatusBadRequest, e)
-		return
-	}
-	errs := []VolumeOptError{}
-	all, _ := strconv.ParseBool(r.FormValue("all"))
-	if all {
-		vs.store.WalkVolume(func(v *storage.Volume) (e error) {
-			if e := v.SetReplica(replica); e != nil {
-				errs = append(errs, VolumeOptError{
-					Volume: v.Id.String(),
-					Err:    e.Error(),
-				})
-			}
-			return nil
-		})
-	} else {
-		volumesSet := make(map[string]bool)
-		for _, volume := range r.Form["volume"] {
-			volumesSet[strings.TrimSpace(volume)] = true
-		}
-		collectionsSet := make(map[string]bool)
-		for _, c := range r.Form["collection"] {
-			collectionsSet[strings.TrimSpace(c)] = true
-		}
-		if len(collectionsSet) > 0 || len(volumesSet) > 0 {
-			vs.store.WalkVolume(func(v *storage.Volume) (e error) {
-				if !collectionsSet[v.Collection] && !volumesSet[v.Id.String()] {
-					return nil
-				}
-				if e := v.SetReplica(replica); e != nil {
-					errs = append(errs, VolumeOptError{
-						Volume: v.Id.String(),
-						Err:    e.Error(),
-					})
-				}
-				return nil
-			})
-		}
-
-	}
-
-	result := make(map[string]interface{})
-	if len(errs) > 0 {
-		result["error"] = "set volume replica error."
-		result["errors"] = errs
-	}
-
-	writeJson(w, r, http.StatusAccepted, result)
-}

go/weed/weed_server/volume_server_handlers_sync.go

@@ -2,11 +2,14 @@ package weed_server
 
 import (
 	"fmt"
+	"io"
 	"net/http"
+	"strconv"
 
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/storage"
 	"github.com/chrislusf/seaweedfs/go/util"
+	"github.com/pierrec/lz4"
 )
 
 func (vs *VolumeServer) getVolumeSyncStatusHandler(w http.ResponseWriter, r *http.Request) {
@@ -84,3 +87,61 @@ func (vs *VolumeServer) getVolume(volumeParameterName string, r *http.Request) (
 	}
 	return v, nil
 }
+
+func (vs *VolumeServer) getVolumeCleanDataHandler(w http.ResponseWriter, r *http.Request) {
+	v, e := vs.getVolume("volume", r)
+	if v == nil {
+		http.Error(w, fmt.Sprintf("Not Found volume: %v", e), http.StatusBadRequest)
+		return
+	}
+	//set read only when replicating
+	v.SetReadOnly(true)
+	defer v.SetReadOnly(false)
+	cr, e := v.GetVolumeCleanReader()
+	if e != nil {
+		http.Error(w, fmt.Sprintf("Get volume clean reader: %v", e), http.StatusInternalServerError)
+		return
+	}
+	totalSize, e := cr.Size()
+	if e != nil {
+		http.Error(w, fmt.Sprintf("Get volume size: %v", e), http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Accept-Ranges", "bytes")
+	w.Header().Set("Content-Disposition", fmt.Sprintf(`filename="%d.dat.lz4"`, v.Id))
+
+	rangeReq := r.Header.Get("Range")
+	if rangeReq == "" {
+		w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
+		w.Header().Set("Content-Encoding", "lz4")
+		lz4w := lz4.NewWriter(w)
+		if _, e = io.Copy(lz4w, cr); e != nil {
+			glog.V(4).Infoln("response write error:", e)
+		}
+		lz4w.Close()
+		return
+	}
+	ranges, e := parseRange(rangeReq, totalSize)
+	if e != nil {
+		http.Error(w, e.Error(), http.StatusRequestedRangeNotSatisfiable)
+		return
+	}
+	if len(ranges) != 1 {
+		http.Error(w, "Only support one range", http.StatusNotImplemented)
+		return
+	}
+	ra := ranges[0]
+	if _, e := cr.Seek(ra.start, 0); e != nil {
+		http.Error(w, e.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10))
+	w.Header().Set("Content-Range", ra.contentRange(totalSize))
+	w.Header().Set("Content-Encoding", "lz4")
+	w.WriteHeader(http.StatusPartialContent)
+	lz4w := lz4.NewWriter(w)
+	if _, e = io.CopyN(lz4w, cr, ra.length); e != nil {
+		glog.V(2).Infoln("response write error:", e)
+	}
+	lz4w.Close()
+}