Merge branch 'master' of https://github.com/bingoohuang/seaweedfs

1 files changed, 77 insertions, 56 deletions

diff --git a/weed/server/master_server.go b/weed/server/master_server.go
index 33a5129da..e2b2df18d 100644
--- a/weed/server/master_server.go
+++ b/weed/server/master_server.go
@@ -1,19 +1,20 @@
 package weed_server
 
 import (
-	"context"
 	"fmt"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
 	"os"
 	"regexp"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/chrislusf/raft"
+	"github.com/gorilla/mux"
+	"google.golang.org/grpc"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
@@ -22,9 +23,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/topology"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
-	"github.com/gorilla/mux"
-	"github.com/spf13/viper"
-	"google.golang.org/grpc"
 )
 
 const (
@@ -33,11 +31,12 @@ const (
 )
 
 type MasterOption struct {
-	Port                    int
-	MetaFolder              string
-	VolumeSizeLimitMB       uint
-	VolumePreallocate       bool
-	PulseSeconds            int
+	Host              string
+	Port              int
+	MetaFolder        string
+	VolumeSizeLimitMB uint
+	VolumePreallocate bool
+	// PulseSeconds            int
 	DefaultReplicaPlacement string
 	GarbageThreshold        float64
 	WhiteList               []string
@@ -56,7 +55,7 @@ type MasterServer struct {
 	vg     *topology.VolumeGrowth
 	vgLock sync.Mutex
 
-	bounedLeaderChan chan int
+	boundedLeaderChan chan int
 
 	// notifying clients
 	clientChansLock sync.RWMutex
@@ -65,11 +64,13 @@ type MasterServer struct {
 	grpcDialOption grpc.DialOption
 
 	MasterClient *wdclient.MasterClient
+
+	adminLocks *AdminLocks
 }
 
 func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *MasterServer {
 
-	v := viper.GetViper()
+	v := util.GetViper()
 	signingKey := v.GetString("jwt.signing.key")
 	v.SetDefault("jwt.signing.expires_after_seconds", 10)
 	expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
@@ -78,35 +79,39 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste
 	v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
 	readExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")
 
+	v.SetDefault("master.replication.treat_replication_as_minimums", false)
+	replicationAsMin := v.GetBool("master.replication.treat_replication_as_minimums")
+
 	var preallocateSize int64
 	if option.VolumePreallocate {
 		preallocateSize = int64(option.VolumeSizeLimitMB) * (1 << 20)
 	}
 
-	grpcDialOption := security.LoadClientTLS(v.Sub("grpc"), "master")
+	grpcDialOption := security.LoadClientTLS(v, "grpc.master")
 	ms := &MasterServer{
 		option:          option,
 		preallocateSize: preallocateSize,
 		clientChans:     make(map[string]chan *master_pb.VolumeLocation),
 		grpcDialOption:  grpcDialOption,
-		MasterClient:    wdclient.NewMasterClient(context.Background(), grpcDialOption, "master", peers),
+		MasterClient:    wdclient.NewMasterClient(grpcDialOption, "master", option.Host, 0, "", peers),
+		adminLocks:      NewAdminLocks(),
 	}
-	ms.bounedLeaderChan = make(chan int, 16)
+	ms.boundedLeaderChan = make(chan int, 16)
 
 	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.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin)
 	ms.vg = topology.NewDefaultVolumeGrowth()
 	glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
 
 	ms.guard = security.NewGuard(ms.option.WhiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
 
+	handleStaticResources2(r)
+	r.HandleFunc("/", ms.proxyToLeader(ms.uiStatusHandler))
+	r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
 	if !ms.option.DisableHttp {
-		handleStaticResources2(r)
-		r.HandleFunc("/", ms.proxyToLeader(ms.uiStatusHandler))
-		r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
 		r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
 		r.HandleFunc("/dir/lookup", ms.guard.WhiteList(ms.dirLookupHandler))
 		r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
@@ -115,9 +120,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste
 		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("/stats/health", ms.guard.WhiteList(statsHealthHandler))
-		r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
-		r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
+		/*
+			r.HandleFunc("/stats/health", ms.guard.WhiteList(statsHealthHandler))
+			r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
+			r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
+		*/
 		r.HandleFunc("/{fileId}", ms.redirectHandler)
 	}
 
@@ -131,14 +138,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste
 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)
+		glog.V(0).Infof("leader change event: %+v => %+v", e.PrevValue(), e.Value())
 		if ms.Topo.RaftServer.Leader() != "" {
 			glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "becomes leader.")
 		}
 	})
-	ms.Topo.RaftServer.AddEventListener(raft.StateChangeEventType, func(e raft.Event) {
-		glog.V(0).Infof("state change: %+v", e)
-	})
 	if ms.Topo.IsLeader() {
 		glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", "I am the leader!")
 	} else {
@@ -148,13 +152,13 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
 	}
 }
 
-func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Request)) func(w http.ResponseWriter, r *http.Request) {
+func (ms *MasterServer) proxyToLeader(f http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if ms.Topo.IsLeader() {
 			f(w, r)
 		} else if ms.Topo.RaftServer != nil && ms.Topo.RaftServer.Leader() != "" {
-			ms.bounedLeaderChan <- 1
-			defer func() { <-ms.bounedLeaderChan }()
+			ms.boundedLeaderChan <- 1
+			defer func() { <-ms.boundedLeaderChan }()
 			targetUrl, err := url.Parse("http://" + ms.Topo.RaftServer.Leader())
 			if err != nil {
 				writeJsonError(w, r, http.StatusInternalServerError,
@@ -183,7 +187,7 @@ func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Requ
 func (ms *MasterServer) startAdminScripts() {
 	var err error
 
-	v := viper.GetViper()
+	v := util.GetViper()
 	adminScripts := v.GetString("master.maintenance.scripts")
 	glog.V(0).Infof("adminScripts:\n%v", adminScripts)
 	if adminScripts == "" {
@@ -193,20 +197,25 @@ func (ms *MasterServer) startAdminScripts() {
 	v.SetDefault("master.maintenance.sleep_minutes", 17)
 	sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")
 
-	v.SetDefault("master.filer.default_filer_url", "http://localhost:8888/")
-	filerURL := v.GetString("master.filer.default_filer_url")
+	v.SetDefault("master.filer.default", "localhost:8888")
+	filerHostPort := v.GetString("master.filer.default")
 
 	scriptLines := strings.Split(adminScripts, "\n")
+	if !strings.Contains(adminScripts, "lock") {
+		scriptLines = append(append([]string{}, "lock"), scriptLines...)
+		scriptLines = append(scriptLines, "unlock")
+	}
 
-	masterAddress := "localhost:" + strconv.Itoa(ms.option.Port)
+	masterAddress := fmt.Sprintf("%s:%d", ms.option.Host, ms.option.Port)
 
 	var shellOptions shell.ShellOptions
-	shellOptions.GrpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "master")
+	shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
 	shellOptions.Masters = &masterAddress
 
-	shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, err = util.ParseFilerUrl(filerURL)
+	shellOptions.FilerHost, shellOptions.FilerPort, err = util.ParseHostPort(filerHostPort)
+	shellOptions.Directory = "/"
 	if err != nil {
-		glog.V(0).Infof("failed to parse master.filer.default_filer_urll=%s : %v\n", filerURL, err)
+		glog.V(0).Infof("failed to parse master.filer.default = %s : %v\n", filerHostPort, err)
 		return
 	}
 
@@ -220,27 +229,11 @@ func (ms *MasterServer) startAdminScripts() {
 		commandEnv.MasterClient.WaitUntilConnected()
 
 		c := time.Tick(time.Duration(sleepMinutes) * time.Minute)
-		for _ = range c {
+		for range c {
 			if ms.Topo.IsLeader() {
 				for _, line := range scriptLines {
-
-					cmds := reg.FindAllString(line, -1)
-					if len(cmds) == 0 {
-						continue
-					}
-					args := make([]string, len(cmds[1:]))
-					for i := range args {
-						args[i] = strings.Trim(string(cmds[1+i]), "\"'")
-					}
-					cmd := strings.ToLower(cmds[0])
-
-					for _, c := range shell.Commands {
-						if c.Name() == cmd {
-							glog.V(0).Infof("executing: %s %v", cmd, args)
-							if err := c.Do(args, commandEnv, os.Stdout); err != nil {
-								glog.V(0).Infof("error: %v", err)
-							}
-						}
+					for _, c := range strings.Split(line, ";") {
+						processEachCmd(reg, c, commandEnv)
 					}
 				}
 			}
@@ -248,9 +241,30 @@ func (ms *MasterServer) startAdminScripts() {
 	}()
 }
 
+func processEachCmd(reg *regexp.Regexp, line string, commandEnv *shell.CommandEnv) {
+	cmds := reg.FindAllString(line, -1)
+	if len(cmds) == 0 {
+		return
+	}
+	args := make([]string, len(cmds[1:]))
+	for i := range args {
+		args[i] = strings.Trim(string(cmds[1+i]), "\"'")
+	}
+	cmd := strings.ToLower(cmds[0])
+
+	for _, c := range shell.Commands {
+		if c.Name() == cmd {
+			glog.V(0).Infof("executing: %s %v", cmd, args)
+			if err := c.Do(args, commandEnv, os.Stdout); err != nil {
+				glog.V(0).Infof("error: %v", err)
+			}
+		}
+	}
+}
+
 func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer {
 	var seq sequence.Sequencer
-	v := viper.GetViper()
+	v := util.GetViper()
 	seqType := strings.ToLower(v.GetString(SequencerType))
 	glog.V(1).Infof("[%s] : [%s]", SequencerType, seqType)
 	switch strings.ToLower(seqType) {
@@ -263,6 +277,13 @@ func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer
 			glog.Error(err)
 			seq = nil
 		}
+	case "snowflake":
+		var err error
+		seq, err = sequence.NewSnowflakeSequencer(fmt.Sprintf("%s:%d", option.Host, option.Port))
+		if err != nil {
+			glog.Error(err)
+			seq = nil
+		}
 	default:
 		seq = sequence.NewMemorySequencer()
 	}