seaweedfs/weed/server/master_server.go

136 lines
4.8 KiB
Go
Raw Normal View History

package weed_server
import (
"fmt"
"net/http"
"net/http/httputil"
"net/url"
"sync"
2015-05-03 19:37:49 +00:00
"github.com/chrislusf/raft"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/sequence"
"github.com/chrislusf/seaweedfs/weed/topology"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/gorilla/mux"
)
type MasterServer struct {
port int
metaFolder string
volumeSizeLimitMB uint
2017-01-08 19:01:46 +00:00
preallocate int64
pulseSeconds int
defaultReplicaPlacement string
garbageThreshold string
2015-02-07 23:35:28 +00:00
guard *security.Guard
Topo *topology.Topology
vg *topology.VolumeGrowth
vgLock sync.Mutex
2014-02-05 09:54:52 +00:00
bounedLeaderChan chan int
}
func NewMasterServer(r *mux.Router, port int, metaFolder string,
volumeSizeLimitMB uint,
2017-01-08 19:01:46 +00:00
preallocate bool,
pulseSeconds int,
defaultReplicaPlacement string,
garbageThreshold string,
2014-02-05 09:54:52 +00:00
whiteList []string,
2015-01-05 22:20:04 +00:00
secureKey string,
2014-02-05 09:54:52 +00:00
) *MasterServer {
2017-01-08 19:01:46 +00:00
var preallocateSize int64
if preallocate {
preallocateSize = int64(volumeSizeLimitMB) * (1 << 20)
}
ms := &MasterServer{
2014-04-15 16:09:40 +00:00
port: port,
volumeSizeLimitMB: volumeSizeLimitMB,
2017-01-08 19:01:46 +00:00
preallocate: preallocateSize,
pulseSeconds: pulseSeconds,
defaultReplicaPlacement: defaultReplicaPlacement,
garbageThreshold: garbageThreshold,
}
ms.bounedLeaderChan = make(chan int, 16)
seq := sequence.NewMemorySequencer()
2017-02-13 05:58:44 +00:00
ms.Topo = topology.NewTopology("topo", seq, uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds)
ms.vg = topology.NewDefaultVolumeGrowth()
glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")
2015-02-07 23:35:28 +00:00
ms.guard = security.NewGuard(whiteList, secureKey)
2015-01-05 22:20:04 +00:00
2015-03-22 19:50:04 +00:00
r.HandleFunc("/", ms.uiStatusHandler)
2015-03-19 17:39:22 +00:00
r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
2015-02-07 23:35:28 +00:00
r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
r.HandleFunc("/dir/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.dirLookupHandler)))
r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(ms.collectionDeleteHandler)))
r.HandleFunc("/vol/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeLookupHandler)))
r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeGrowHandler)))
r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeStatusHandler)))
r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeVacuumHandler)))
r.HandleFunc("/submit", ms.guard.WhiteList(ms.submitFromMasterServerHandler))
r.HandleFunc("/delete", ms.guard.WhiteList(ms.deleteFromMasterServerHandler))
2018-05-07 07:37:33 +00:00
r.HandleFunc("/stats/health", ms.guard.WhiteList(statsHealthHandler))
2015-02-07 23:35:28 +00:00
r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
2018-05-07 07:37:33 +00:00
r.HandleFunc("/{fileId}", ms.proxyToLeader(ms.redirectHandler))
ms.Topo.StartRefreshWritableVolumes(garbageThreshold, ms.preallocate)
return ms
}
2014-02-05 09:54:52 +00:00
func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
ms.Topo.RaftServer = raftServer.raftServer
ms.Topo.RaftServer.AddEventListener(raft.LeaderChangeEventType, func(e raft.Event) {
glog.V(0).Infof("event: %+v", e)
if ms.Topo.RaftServer.Leader() != "" {
glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "becomes leader.")
}
2014-02-10 07:37:29 +00:00
})
if ms.Topo.IsLeader() {
glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", "I am the leader!")
2014-02-10 07:37:29 +00:00
} else {
if ms.Topo.RaftServer.Leader() != "" {
glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "is the leader.")
}
2014-02-10 07:37:29 +00:00
}
2014-02-05 09:54:52 +00:00
}
func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Request)) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
if ms.Topo.IsLeader() {
2014-02-05 09:54:52 +00:00
f(w, r)
} else if ms.Topo.RaftServer != nil && ms.Topo.RaftServer.Leader() != "" {
ms.bounedLeaderChan <- 1
defer func() { <-ms.bounedLeaderChan }()
targetUrl, err := url.Parse("http://" + ms.Topo.RaftServer.Leader())
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError,
fmt.Errorf("Leader URL http://%s Parse Error: %v", ms.Topo.RaftServer.Leader(), err))
2014-02-05 09:54:52 +00:00
return
}
glog.V(4).Infoln("proxying to leader", ms.Topo.RaftServer.Leader())
2014-02-05 09:54:52 +00:00
proxy := httputil.NewSingleHostReverseProxy(targetUrl)
director := proxy.Director
proxy.Director = func(req *http.Request) {
actualHost, err := security.GetActualRemoteHost(req)
if err == nil {
2015-11-22 18:11:25 +00:00
req.Header.Set("HTTP_X_FORWARDED_FOR", actualHost)
}
director(req)
}
proxy.Transport = util.Transport
2014-02-05 09:54:52 +00:00
proxy.ServeHTTP(w, r)
} else {
//drop it to the floor
2014-05-27 00:34:54 +00:00
//writeJsonError(w, r, errors.New(ms.Topo.RaftServer.Name()+" does not know Leader yet:"+ms.Topo.RaftServer.Leader()))
2014-02-05 09:54:52 +00:00
}
}
}