mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-11-29 14:31:30 +08:00
190 lines
6.7 KiB
Go
190 lines
6.7 KiB
Go
package weed_server
|
|
|
|
import (
|
|
"code.google.com/p/weed-fs/go/storage"
|
|
"code.google.com/p/weed-fs/go/util"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
)
|
|
|
|
func (ms *MasterServer) dirLookupHandler(w http.ResponseWriter, r *http.Request) {
|
|
vid := r.FormValue("volumeId")
|
|
collection := r.FormValue("collection") //optional, but can be faster if too many collections
|
|
commaSep := strings.Index(vid, ",")
|
|
if commaSep > 0 {
|
|
vid = vid[0:commaSep]
|
|
}
|
|
volumeId, err := storage.NewVolumeId(vid)
|
|
if err == nil {
|
|
machines := ms.topo.Lookup(collection, volumeId)
|
|
if machines != nil {
|
|
ret := []map[string]string{}
|
|
for _, dn := range machines {
|
|
ret = append(ret, map[string]string{"url": dn.Url(), "publicUrl": dn.PublicUrl})
|
|
}
|
|
writeJsonQuiet(w, r, map[string]interface{}{"locations": ret})
|
|
} else {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "volume id " + volumeId.String() + " not found. "})
|
|
}
|
|
} else {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "unknown volumeId format " + vid})
|
|
}
|
|
}
|
|
|
|
func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) {
|
|
c, e := strconv.Atoi(r.FormValue("count"))
|
|
if e != nil {
|
|
c = 1
|
|
}
|
|
replication := r.FormValue("replication")
|
|
if replication == "" {
|
|
replication = ms.defaultReplicaPlacement
|
|
}
|
|
collection := r.FormValue("collection")
|
|
dataCenter := r.FormValue("dataCenter")
|
|
replicaPlacement, err := storage.NewReplicaPlacementFromString(replication)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
|
|
if ms.topo.GetVolumeLayout(collection, replicaPlacement).GetActiveVolumeCount(dataCenter) <= 0 {
|
|
if ms.topo.FreeSpace() <= 0 {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "No free volumes left!"})
|
|
return
|
|
} else {
|
|
ms.vgLock.Lock()
|
|
defer ms.vgLock.Unlock()
|
|
if ms.topo.GetVolumeLayout(collection, replicaPlacement).GetActiveVolumeCount(dataCenter) <= 0 {
|
|
if _, err = ms.vg.AutomaticGrowByType(collection, replicaPlacement, dataCenter, ms.topo); err != nil {
|
|
writeJsonQuiet(w, r, map[string]string{"error": "Cannot grow volume group! " + err.Error()})
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
fid, count, dn, err := ms.topo.PickForWrite(collection, replicaPlacement, c, dataCenter)
|
|
if err == nil {
|
|
writeJsonQuiet(w, r, map[string]interface{}{"fid": fid, "url": dn.Url(), "publicUrl": dn.PublicUrl, "count": count})
|
|
} else {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
|
|
}
|
|
}
|
|
|
|
func (ms *MasterServer) collectionDeleteHandler(w http.ResponseWriter, r *http.Request) {
|
|
collection, ok := ms.topo.GetCollection(r.FormValue("collection"))
|
|
if !ok {
|
|
writeJsonQuiet(w, r, map[string]interface{}{"error": "collection " + r.FormValue("collection") + "does not exist!"})
|
|
return
|
|
}
|
|
for _, server := range collection.ListVolumeServers() {
|
|
_, err := util.Get("http://" + server.Ip + ":" + strconv.Itoa(server.Port) + "/admin/delete_collection?collection=" + r.FormValue("collection"))
|
|
if err != nil {
|
|
writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
}
|
|
ms.topo.DeleteCollection(r.FormValue("collection"))
|
|
}
|
|
|
|
func (ms *MasterServer) dirJoinHandler(w http.ResponseWriter, r *http.Request) {
|
|
init := r.FormValue("init") == "true"
|
|
ip := r.FormValue("ip")
|
|
if ip == "" {
|
|
ip = r.RemoteAddr[0:strings.Index(r.RemoteAddr, ":")]
|
|
}
|
|
port, _ := strconv.Atoi(r.FormValue("port"))
|
|
maxVolumeCount, _ := strconv.Atoi(r.FormValue("maxVolumeCount"))
|
|
s := r.RemoteAddr[0:strings.Index(r.RemoteAddr, ":")+1] + r.FormValue("port")
|
|
publicUrl := r.FormValue("publicUrl")
|
|
volumes := new([]storage.VolumeInfo)
|
|
if err := json.Unmarshal([]byte(r.FormValue("volumes")), volumes); err != nil {
|
|
writeJsonQuiet(w, r, map[string]string{"error": "Cannot unmarshal \"volumes\": " + err.Error()})
|
|
return
|
|
}
|
|
debug(s, "volumes", r.FormValue("volumes"))
|
|
ms.topo.RegisterVolumes(init, *volumes, ip, port, publicUrl, maxVolumeCount, r.FormValue("dataCenter"), r.FormValue("rack"))
|
|
m := make(map[string]interface{})
|
|
m["VolumeSizeLimit"] = uint64(ms.volumeSizeLimitMB) * 1024 * 1024
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func (ms *MasterServer) dirStatusHandler(w http.ResponseWriter, r *http.Request) {
|
|
m := make(map[string]interface{})
|
|
m["Version"] = ms.version
|
|
m["Topology"] = ms.topo.ToMap()
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Request) {
|
|
gcThreshold := r.FormValue("garbageThreshold")
|
|
if gcThreshold == "" {
|
|
gcThreshold = ms.garbageThreshold
|
|
}
|
|
debug("garbageThreshold =", gcThreshold)
|
|
ms.topo.Vacuum(gcThreshold)
|
|
ms.dirStatusHandler(w, r)
|
|
}
|
|
|
|
func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request) {
|
|
count := 0
|
|
replicaPlacement, err := storage.NewReplicaPlacementFromString(r.FormValue("replication"))
|
|
if err == nil {
|
|
if count, err = strconv.Atoi(r.FormValue("count")); err == nil {
|
|
if ms.topo.FreeSpace() < count*replicaPlacement.GetCopyCount() {
|
|
err = errors.New("Only " + strconv.Itoa(ms.topo.FreeSpace()) + " volumes left! Not enough for " + strconv.Itoa(count*replicaPlacement.GetCopyCount()))
|
|
} else {
|
|
count, err = ms.vg.GrowByCountAndType(count, r.FormValue("collection"), replicaPlacement, r.FormValue("dataCenter"), ms.topo)
|
|
}
|
|
} else {
|
|
err = errors.New("parameter count is not found")
|
|
}
|
|
}
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
|
|
} else {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]interface{}{"count": count})
|
|
}
|
|
}
|
|
|
|
func (ms *MasterServer) volumeStatusHandler(w http.ResponseWriter, r *http.Request) {
|
|
m := make(map[string]interface{})
|
|
m["Version"] = ms.version
|
|
m["Volumes"] = ms.topo.ToVolumeMap()
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func (ms *MasterServer) redirectHandler(w http.ResponseWriter, r *http.Request) {
|
|
vid, _, _, _, _ := parseURLPath(r.URL.Path)
|
|
volumeId, err := storage.NewVolumeId(vid)
|
|
if err != nil {
|
|
debug("parsing error:", err, r.URL.Path)
|
|
return
|
|
}
|
|
machines := ms.topo.Lookup("", volumeId)
|
|
if machines != nil && len(machines) > 0 {
|
|
http.Redirect(w, r, "http://"+machines[0].PublicUrl+r.URL.Path, http.StatusMovedPermanently)
|
|
} else {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "volume id " + volumeId.String() + " not found. "})
|
|
}
|
|
}
|
|
|
|
func (ms *MasterServer) submitFromMasterServerHandler(w http.ResponseWriter, r *http.Request) {
|
|
if ms.IsLeader() {
|
|
submitForClientHandler(w, r, "localhost:"+strconv.Itoa(ms.port))
|
|
} else {
|
|
submitForClientHandler(w, r, ms.raftServer.Leader())
|
|
}
|
|
}
|