diff --git a/go/weed/weed_server/master_server_handlers_admin.go b/go/weed/weed_server/master_server_handlers_admin.go index 72962517d..b5eecc55b 100644 --- a/go/weed/weed_server/master_server_handlers_admin.go +++ b/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 } + diff --git a/go/weed/weed_server/volume_server.go b/go/weed/weed_server/volume_server.go index 3480ad09b..c1f5acb5a 100644 --- a/go/weed/weed_server/volume_server.go +++ b/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)) diff --git a/go/weed/weed_server/volume_server_handlers_admin.go b/go/weed/weed_server/volume_server_handlers_admin.go index 80aeb3f1d..15dccfb24 100644 --- a/go/weed/weed_server/volume_server_handlers_admin.go +++ b/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) +} diff --git a/go/weed/weed_server/volume_server_handlers_replicate.go b/go/weed/weed_server/volume_server_handlers_replicate.go deleted file mode 100644 index cc421dc88..000000000 --- a/go/weed/weed_server/volume_server_handlers_replicate.go +++ /dev/null @@ -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) -} diff --git a/go/weed/weed_server/volume_server_handlers_sync.go b/go/weed/weed_server/volume_server_handlers_sync.go index c650e5f53..859bff563 100644 --- a/go/weed/weed_server/volume_server_handlers_sync.go +++ b/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() +}