lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 22:47:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #1102 from stlpmo-jn/add_ETCD_sequencer

Browse Source 
Add etcd sequencer
This commit is contained in:
Chris Lu
2019-11-11 12:43:24 -08:00
committed by GitHub
parent 9c2f3b1b0f d07701fa75
commit d7852ebb3b
4 changed files with 343 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
weed/server/master_server.go
View File

@@ -27,6 +27,12 @@ import (
"google.golang.org/grpc"
)
const (
MasterPrefix = "master.maintenance"
SequencerType = MasterPrefix + ".sequencer_type"
SequencerEtcdUrls = MasterPrefix + ".sequencer_etcd_urls"
)
type MasterOption struct {
Port int
MetaFolder string
@@ -87,7 +93,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste
MasterClient: wdclient.NewMasterClient(context.Background(), grpcDialOption, "master", peers),
}
ms.bounedLeaderChan = make(chan int, 16)
seq := sequence.NewMemorySequencer()
seq := ms.createSequencer(option)
if nil == seq {
glog.Fatalf("create sequencer failed.")
}
ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, ms.option.PulseSeconds)
ms.vg = topology.NewDefaultVolumeGrowth()
glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
@@ -165,8 +175,8 @@ func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Requ
proxy.Transport = util.Transport
proxy.ServeHTTP(w, r)
} else {
//drop it to the floor
//writeJsonError(w, r, errors.New(ms.Topo.RaftServer.Name()+" does not know Leader yet:"+ms.Topo.RaftServer.Leader()))
// drop it to the floor
// writeJsonError(w, r, errors.New(ms.Topo.RaftServer.Name()+" does not know Leader yet:"+ms.Topo.RaftServer.Leader()))
}
}
}
@@ -230,3 +240,25 @@ func (ms *MasterServer) startAdminScripts() {
}
}()
}
func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer {
var seq sequence.Sequencer
seqType := strings.ToLower(util.Config().GetString(SequencerType))
glog.V(0).Infof("[%s] : [%s]", SequencerType, seqType)
switch strings.ToLower(seqType) {
case "memory":
seq = sequence.NewMemorySequencer()
case "etcd":
var err error
urls := util.Config().GetString(SequencerEtcdUrls)
glog.V(0).Infof("[%s] : [%s]", SequencerEtcdUrls, urls)
seq, err = sequence.NewEtcdSequencer(urls, option.MetaFolder)
if err != nil {
glog.Error(err)
seq = nil
}
default:
seq = sequence.NewMemorySequencer()
}
return seq
}