mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-11-30 07:09:01 +08:00
303 lines
10 KiB
Go
303 lines
10 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"code.google.com/p/weed-fs/go/glog"
|
|
"code.google.com/p/weed-fs/go/operation"
|
|
"code.google.com/p/weed-fs/go/replication"
|
|
"code.google.com/p/weed-fs/go/sequence"
|
|
"code.google.com/p/weed-fs/go/storage"
|
|
"code.google.com/p/weed-fs/go/topology"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"os"
|
|
"path"
|
|
"runtime"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
func init() {
|
|
cmdMaster.Run = runMaster // break init cycle
|
|
cmdMaster.IsDebug = cmdMaster.Flag.Bool("debug", false, "enable debug mode")
|
|
}
|
|
|
|
var cmdMaster = &Command{
|
|
UsageLine: "master -port=9333",
|
|
Short: "start a master server",
|
|
Long: `start a master server to provide volume=>location mapping service
|
|
and sequence number of file ids
|
|
|
|
`,
|
|
}
|
|
|
|
var (
|
|
mport = cmdMaster.Flag.Int("port", 9333, "http listen port")
|
|
metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store mappings")
|
|
volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", 32*1024, "Default Volume Size in MegaBytes")
|
|
mpulse = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
|
|
confFile = cmdMaster.Flag.String("conf", "/etc/weedfs/weedfs.conf", "xml configuration file")
|
|
defaultRepType = cmdMaster.Flag.String("defaultReplicationType", "000", "Default replication type if not specified.")
|
|
mReadTimeout = cmdMaster.Flag.Int("readTimeout", 3, "connection read timeout in seconds")
|
|
mMaxCpu = cmdMaster.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
|
|
garbageThreshold = cmdMaster.Flag.String("garbageThreshold", "0.3", "threshold to vacuum and reclaim spaces")
|
|
masterWhiteListOption = cmdMaster.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
|
|
//etcdCluster = cmdMaster.Flag.String("etcd", "", "comma separated etcd urls, e.g., http://localhost:4001, See github.com/coreos/go-etcd/etcd")
|
|
|
|
masterWhiteList []string
|
|
)
|
|
|
|
var topo *topology.Topology
|
|
var vg *replication.VolumeGrowth
|
|
var vgLock sync.Mutex
|
|
|
|
func 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 := 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 dirAssignHandler(w http.ResponseWriter, r *http.Request) {
|
|
c, e := strconv.Atoi(r.FormValue("count"))
|
|
if e != nil {
|
|
c = 1
|
|
}
|
|
repType := r.FormValue("replication")
|
|
if repType == "" {
|
|
repType = *defaultRepType
|
|
}
|
|
collection := r.FormValue("collection")
|
|
dataCenter := r.FormValue("dataCenter")
|
|
rt, err := storage.NewReplicationTypeFromString(repType)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
|
|
if topo.GetVolumeLayout(collection, rt).GetActiveVolumeCount(dataCenter) <= 0 {
|
|
if topo.FreeSpace() <= 0 {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "No free volumes left!"})
|
|
return
|
|
} else {
|
|
vgLock.Lock()
|
|
defer vgLock.Unlock()
|
|
if topo.GetVolumeLayout(collection, rt).GetActiveVolumeCount(dataCenter) <= 0 {
|
|
if _, err = vg.AutomaticGrowByType(collection, rt, dataCenter, topo); err != nil {
|
|
writeJsonQuiet(w, r, map[string]string{"error": "Cannot grow volume group! " + err.Error()})
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
fid, count, dn, err := topo.PickForWrite(collection, rt, 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 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"))
|
|
topo.RegisterVolumes(init, *volumes, ip, port, publicUrl, maxVolumeCount, r.FormValue("dataCenter"), r.FormValue("rack"))
|
|
m := make(map[string]interface{})
|
|
m["VolumeSizeLimit"] = uint64(*volumeSizeLimitMB) * 1024 * 1024
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func dirStatusHandler(w http.ResponseWriter, r *http.Request) {
|
|
m := make(map[string]interface{})
|
|
m["Version"] = VERSION
|
|
m["Topology"] = topo.ToMap()
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func volumeVacuumHandler(w http.ResponseWriter, r *http.Request) {
|
|
gcThreshold := r.FormValue("garbageThreshold")
|
|
if gcThreshold == "" {
|
|
gcThreshold = *garbageThreshold
|
|
}
|
|
debug("garbageThreshold =", gcThreshold)
|
|
topo.Vacuum(gcThreshold)
|
|
dirStatusHandler(w, r)
|
|
}
|
|
|
|
func volumeGrowHandler(w http.ResponseWriter, r *http.Request) {
|
|
count := 0
|
|
rt, err := storage.NewReplicationTypeFromString(r.FormValue("replication"))
|
|
if err == nil {
|
|
if count, err = strconv.Atoi(r.FormValue("count")); err == nil {
|
|
if topo.FreeSpace() < count*rt.GetCopyCount() {
|
|
err = errors.New("Only " + strconv.Itoa(topo.FreeSpace()) + " volumes left! Not enough for " + strconv.Itoa(count*rt.GetCopyCount()))
|
|
} else {
|
|
count, err = vg.GrowByCountAndType(count, r.FormValue("collection"), rt, r.FormValue("dataCneter"), topo)
|
|
}
|
|
} else {
|
|
err = errors.New("parameter count is not found")
|
|
}
|
|
}
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]string{"error": "parameter replication " + err.Error()})
|
|
} else {
|
|
w.WriteHeader(http.StatusNotAcceptable)
|
|
writeJsonQuiet(w, r, map[string]interface{}{"count": count})
|
|
}
|
|
}
|
|
|
|
func volumeStatusHandler(w http.ResponseWriter, r *http.Request) {
|
|
m := make(map[string]interface{})
|
|
m["Version"] = VERSION
|
|
m["Volumes"] = topo.ToVolumeMap()
|
|
writeJsonQuiet(w, r, m)
|
|
}
|
|
|
|
func 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 := 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 submitFromMasterServerHandler(w http.ResponseWriter, r *http.Request) {
|
|
submitForClientHandler(w, r, "localhost:"+strconv.Itoa(*mport))
|
|
}
|
|
|
|
func runMaster(cmd *Command, args []string) bool {
|
|
if *mMaxCpu < 1 {
|
|
*mMaxCpu = runtime.NumCPU()
|
|
}
|
|
runtime.GOMAXPROCS(*mMaxCpu)
|
|
if *masterWhiteListOption != "" {
|
|
masterWhiteList = strings.Split(*masterWhiteListOption, ",")
|
|
}
|
|
var seq sequence.Sequencer
|
|
//if len(*etcdCluster) == 0 {
|
|
seq = sequence.NewFileSequencer(path.Join(*metaFolder, "weed.seq"))
|
|
//} else {
|
|
// seq = sequence.NewEtcdSequencer(*etcdCluster)
|
|
//}
|
|
var e error
|
|
if topo, e = topology.NewTopology("topo", *confFile, seq,
|
|
uint64(*volumeSizeLimitMB)*1024*1024, *mpulse); e != nil {
|
|
glog.Fatalf("cannot create topology:%s", e)
|
|
}
|
|
vg = replication.NewDefaultVolumeGrowth()
|
|
glog.V(0).Infoln("Volume Size Limit is", *volumeSizeLimitMB, "MB")
|
|
http.HandleFunc("/dir/assign", secure(masterWhiteList, dirAssignHandler))
|
|
http.HandleFunc("/dir/lookup", secure(masterWhiteList, dirLookupHandler))
|
|
http.HandleFunc("/dir/join", secure(masterWhiteList, dirJoinHandler))
|
|
http.HandleFunc("/dir/status", secure(masterWhiteList, dirStatusHandler))
|
|
http.HandleFunc("/vol/grow", secure(masterWhiteList, volumeGrowHandler))
|
|
http.HandleFunc("/vol/status", secure(masterWhiteList, volumeStatusHandler))
|
|
http.HandleFunc("/vol/vacuum", secure(masterWhiteList, volumeVacuumHandler))
|
|
|
|
http.HandleFunc("/submit", secure(masterWhiteList, submitFromMasterServerHandler))
|
|
http.HandleFunc("/", redirectHandler)
|
|
|
|
topo.StartRefreshWritableVolumes(*garbageThreshold)
|
|
|
|
glog.V(0).Infoln("Start Weed Master", VERSION, "at port", strconv.Itoa(*mport))
|
|
srv := &http.Server{
|
|
Addr: ":" + strconv.Itoa(*mport),
|
|
Handler: http.DefaultServeMux,
|
|
ReadTimeout: time.Duration(*mReadTimeout) * time.Second,
|
|
}
|
|
e = srv.ListenAndServe()
|
|
if e != nil {
|
|
glog.Fatalf("Fail to start:%s", e)
|
|
}
|
|
return true
|
|
}
|
|
|
|
func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl string) {
|
|
m := make(map[string]interface{})
|
|
if r.Method != "POST" {
|
|
m["error"] = "Only submit via POST!"
|
|
writeJsonQuiet(w, r, m)
|
|
return
|
|
}
|
|
|
|
debug("parsing upload file...")
|
|
fname, data, mimeType, isGzipped, lastModified, pe := storage.ParseUpload(r)
|
|
if pe != nil {
|
|
writeJsonError(w, r, pe)
|
|
return
|
|
}
|
|
|
|
debug("assigning file id for", fname)
|
|
assignResult, ae := operation.Assign(masterUrl, 1, r.FormValue("replication"))
|
|
if ae != nil {
|
|
writeJsonError(w, r, ae)
|
|
return
|
|
}
|
|
|
|
url := "http://" + assignResult.PublicUrl + "/" + assignResult.Fid
|
|
if lastModified != 0 {
|
|
url = url + "?ts=" + strconv.FormatUint(lastModified, 10)
|
|
}
|
|
|
|
debug("upload file to store", url)
|
|
uploadResult, err := operation.Upload(url, fname, bytes.NewReader(data), isGzipped, mimeType)
|
|
if err != nil {
|
|
writeJsonError(w, r, err)
|
|
return
|
|
}
|
|
|
|
m["fileName"] = fname
|
|
m["fid"] = assignResult.Fid
|
|
m["fileUrl"] = assignResult.PublicUrl + "/" + assignResult.Fid
|
|
m["size"] = uploadResult.Size
|
|
writeJsonQuiet(w, r, m)
|
|
return
|
|
}
|