mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-25 19:49:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

1. add batched volume lookup handler

Browse Source 
2. working-in-progress batch delete
This commit is contained in:
Chris Lu 2014-04-13 23:41:34 -07:00
parent 161526555f
commit f20ef922fd
3 changed files with 79 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
go/weed/weed_server/common.go
View File

@ -130,6 +130,14 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl st
return
}
func deleteForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl string) {
r.ParseForm()
fids := r.Form["fid"]
fids = fids
m := make(map[string]interface{})
writeJsonQuiet(w, r, m)
}
func parseURLPath(path string) (vid, fid, filename, ext string, isVolumeIdOnly bool) {
switch strings.Count(path, "/") {
case 3:
@ -162,6 +170,7 @@ func parseURLPath(path string) (vid, fid, filename, ext string, isVolumeIdOnly b
}
return
}
func statsCounterHandler(w http.ResponseWriter, r *http.Request) {
m := make(map[string]interface{})
m["Version"] = util.VERSION
@ -169,9 +178,6 @@ func statsCounterHandler(w http.ResponseWriter, r *http.Request) {
writeJsonQuiet(w, r, m)
}
type MemoryStatistics struct {
}
func statsMemoryHandler(w http.ResponseWriter, r *http.Request) {
m := make(map[string]interface{})
m["Version"] = util.VERSION

2
go/weed/weed_server/master_server.go
View File

@ -61,10 +61,12 @@ func NewMasterServer(r *mux.Router, port int, metaFolder string,
r.HandleFunc("/dir/join", ms.proxyToLeader(secure(ms.whiteList, ms.dirJoinHandler)))
r.HandleFunc("/dir/status", ms.proxyToLeader(secure(ms.whiteList, ms.dirStatusHandler)))
r.HandleFunc("/col/delete", ms.proxyToLeader(secure(ms.whiteList, ms.collectionDeleteHandler)))
r.HandleFunc("/vol/lookup", ms.proxyToLeader(secure(ms.whiteList, ms.volumeLookupHandler)))
r.HandleFunc("/vol/grow", ms.proxyToLeader(secure(ms.whiteList, ms.volumeGrowHandler)))
r.HandleFunc("/vol/status", ms.proxyToLeader(secure(ms.whiteList, ms.volumeStatusHandler)))
r.HandleFunc("/vol/vacuum", ms.proxyToLeader(secure(ms.whiteList, ms.volumeVacuumHandler)))
r.HandleFunc("/submit", secure(ms.whiteList, ms.submitFromMasterServerHandler))
r.HandleFunc("/delete", secure(ms.whiteList, ms.deleteFromMasterServerHandler))
r.HandleFunc("/{filekey}", ms.redirectHandler)
r.HandleFunc("/stats/counter", secure(ms.whiteList, statsCounterHandler))
r.HandleFunc("/stats/memory", secure(ms.whiteList, statsMemoryHandler))

71
go/weed/weed_server/master_server_handlers.go
View File

@ -12,30 +12,73 @@ import (
"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
type LookupResultLocation struct {
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
}
type LookupResult struct {
VolumeId string `json:"volumeId,omitempty"`
Locations []LookupResultLocation `json:"locations,omitempty"`
Error string `json:"error,omitempty"`
}
func (ms *MasterServer) lookupVolumeId(vids []string, collection string) (volumeLocations map[string]LookupResult) {
volumeLocations = make(map[string]LookupResult)
for _, vid := range vids {
commaSep := strings.Index(vid, ",")
if commaSep > 0 {
vid = vid[0:commaSep]
}
if _, ok := volumeLocations[vid]; ok {
continue
}
volumeId, err := storage.NewVolumeId(vid)
if err == nil {
machines := ms.Topo.Lookup(collection, volumeId)
if machines != nil {
ret := []map[string]string{}
var ret []LookupResultLocation
for _, dn := range machines {
ret = append(ret, map[string]string{"url": dn.Url(), "publicUrl": dn.PublicUrl})
ret = append(ret, LookupResultLocation{Url: dn.Url(), PublicUrl: dn.PublicUrl})
}
writeJsonQuiet(w, r, map[string]interface{}{"locations": ret})
volumeLocations[vid] = LookupResult{VolumeId: vid, Locations: ret}
} else {
volumeLocations[vid] = LookupResult{VolumeId: vid, Error: "volumeId not found."}
}
} else {
volumeLocations[vid] = LookupResult{VolumeId: vid, Error: "Unknown volumeId format."}
}
}
return
}
// Takes one volumeId only, can not do batch lookup
func (ms *MasterServer) dirLookupHandler(w http.ResponseWriter, r *http.Request) {
vid := r.FormValue("volumeId")
commaSep := strings.Index(vid, ",")
if commaSep > 0 {
vid = vid[0:commaSep]
}
vids := []string{vid}
collection := r.FormValue("collection") //optional, but can be faster if too many collections
volumeLocations := ms.lookupVolumeId(vids, collection)
location := volumeLocations[vid]
if location.Error != "" {
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})
writeJsonQuiet(w, r, location)
}
// This can take batched volumeIds, &volumeId=x&volumeId=y&volumeId=z
func (ms *MasterServer) volumeLookupHandler(w http.ResponseWriter, r *http.Request) {
r.ParseForm()
vids := r.Form["volumeId"]
collection := r.FormValue("collection") //optional, but can be faster if too many collections
volumeLocations := ms.lookupVolumeId(vids, collection)
var ret []LookupResult
for _, volumeLocation := range volumeLocations {
ret = append(ret, volumeLocation)
}
writeJsonQuiet(w, r, ret)
}
func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) {
@ -191,6 +234,14 @@ func (ms *MasterServer) submitFromMasterServerHandler(w http.ResponseWriter, r *
}
}
func (ms *MasterServer) deleteFromMasterServerHandler(w http.ResponseWriter, r *http.Request) {
if ms.Topo.IsLeader() {
deleteForClientHandler(w, r, "localhost:"+strconv.Itoa(ms.port))
} else {
deleteForClientHandler(w, r, ms.Topo.RaftServer.Leader())
}
}
func (ms *MasterServer) hasWriableVolume(option *topology.VolumeGrowOption) bool {
vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement)
return vl.GetActiveVolumeCount(option) > 0