Merge branch 'master' into mq-subscribe

k8s/charts/seaweedfs/templates/filer-statefulset.yaml

@@ -23,7 +23,6 @@ spec:
   selector:
     matchLabels:
       app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
-      helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
       app.kubernetes.io/instance: {{ .Release.Name }}
       app.kubernetes.io/component: filer
   template:

k8s/charts/seaweedfs/templates/master-statefulset.yaml

@@ -22,7 +22,6 @@ spec:
   selector:
     matchLabels:
       app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
-      helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
       app.kubernetes.io/instance: {{ .Release.Name }}
       app.kubernetes.io/component: master
   template:

k8s/charts/seaweedfs/templates/volume-statefulset.yaml

@@ -16,7 +16,6 @@ spec:
   selector:
     matchLabels:
       app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
-      helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
       app.kubernetes.io/instance: {{ .Release.Name }}
       app.kubernetes.io/component: volume
   template:

weed/command/filer_remote_sync_dir.go

@@ -278,7 +278,10 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer
 }
 
 func isMultipartUploadFile(dir string, name string) bool {
-	return strings.HasPrefix(dir, "/buckets/") &&
+	return isMultipartUploadDir(dir) &&	strings.HasSuffix(name, ".part")
-		strings.Contains(dir, "/"+s3_constants.MultipartUploadsFolder+"/") &&
+}
-		strings.HasSuffix(name, ".part")
+
+func isMultipartUploadDir(dir string) bool {
+	return strings.HasPrefix(dir, "/buckets/") &&
+		strings.Contains(dir, "/"+s3_constants.MultipartUploadsFolder+"/")
 }

weed/command/filer_sync.go

@@ -394,6 +394,10 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 			glog.V(0).Infof("received %v", resp)
 		}
 
+		if isMultipartUploadDir(resp.Directory) {
+			return nil
+		}
+
 		if !strings.HasPrefix(resp.Directory, sourcePath) {
 			return nil
 		}

weed/command/scaffold.go

@@ -21,9 +21,9 @@ var cmdScaffold = &Command{
 	For example, the filer.toml mysql password can be overwritten by environment variable
 		export WEED_MYSQL_PASSWORD=some_password
 	Environment variable rules:
-		* Prefix the variable name with "WEED_"
+		* Prefix the variable name with "WEED_".
-		* Uppercase the reset of variable name.
+		* Uppercase the rest of the variable name.
-		* Replace '.' with '_'
+		* Replace '.' with '_'.
 
   `,
 }

weed/server/filer_server.go

@@ -171,6 +171,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	}
 	if defaultMux != readonlyMux {
 		handleStaticResources(readonlyMux)
+		readonlyMux.HandleFunc("/healthz", fs.filerHealthzHandler)
 		readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
 	}
 

weed/server/filer_server_handlers.go

@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"context"
 	"errors"
 	"net/http"
 	"os"
@@ -9,7 +10,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 
@@ -44,7 +47,7 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
 	}
 
-	if r.Method == "OPTIONS" {
+	if r.Method == http.MethodOptions {
 		OptionsHandler(w, r, false)
 		return
 	}
@@ -66,7 +69,7 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		stats.FilerRequestHistogram.WithLabelValues(r.Method).Observe(time.Since(start).Seconds())
 	}()
 
-	isReadHttpCall := r.Method == "GET" || r.Method == "HEAD"
+	isReadHttpCall := r.Method == http.MethodGet || r.Method == http.MethodHead
 	if !fs.maybeCheckJwtAuthorization(r, !isReadHttpCall) {
 		writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt"))
 		return
@@ -75,17 +78,15 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 
 	switch r.Method {
-	case "GET":
+	case http.MethodGet, http.MethodHead:
 		fs.GetOrHeadHandler(w, r)
-	case "HEAD":
+	case http.MethodDelete:
-		fs.GetOrHeadHandler(w, r)
-	case "DELETE":
 		if _, ok := r.URL.Query()["tagging"]; ok {
 			fs.DeleteTaggingHandler(w, r)
 		} else {
 			fs.DeleteHandler(w, r)
 		}
-	case "POST", "PUT":
+	case http.MethodPost, http.MethodPut:
 		// wait until in flight data is less than the limit
 		contentLength := getContentLength(r)
 		fs.inFlightDataLimitCond.L.Lock()
@@ -102,7 +103,7 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 			fs.inFlightDataLimitCond.Signal()
 		}()
 
-		if r.Method == "PUT" {
+		if r.Method == http.MethodPut {
 			if _, ok := r.URL.Query()["tagging"]; ok {
 				fs.PutTaggingHandler(w, r)
 			} else {
@@ -111,6 +112,8 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		} else { // method == "POST"
 			fs.PostHandler(w, r, contentLength)
 		}
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
 	}
 }
 
@@ -149,7 +152,7 @@ func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Reque
 		stats.FilerRequestHistogram.WithLabelValues(r.Method).Observe(time.Since(start).Seconds())
 	}()
 	// We handle OPTIONS first because it never should be authenticated
-	if r.Method == "OPTIONS" {
+	if r.Method == http.MethodOptions {
 		OptionsHandler(w, r, true)
 		return
 	}
@@ -162,10 +165,10 @@ func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Reque
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 
 	switch r.Method {
-	case "GET":
+	case http.MethodGet, http.MethodHead:
-		fs.GetOrHeadHandler(w, r)
-	case "HEAD":
 		fs.GetOrHeadHandler(w, r)
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
 	}
 }
 
@@ -220,5 +223,10 @@ func (fs *FilerServer) maybeCheckJwtAuthorization(r *http.Request, isWrite bool)
 
 func (fs *FilerServer) filerHealthzHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
-	w.WriteHeader(http.StatusOK)
+	if _, err := fs.filer.Store.FindEntry(context.Background(), filer.TopicsDir); err != nil && err != filer_pb.ErrNotFound {
+		glog.Warningf("filerHealthzHandler FindEntry: %+v", err)
+		w.WriteHeader(http.StatusServiceUnavailable)
+	} else {
+		w.WriteHeader(http.StatusOK)
+	}
 }

weed/server/master_server_handlers.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/security"
@@ -119,34 +120,52 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
 
 	vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
 
-	fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(requestedCount, option, vl)
+	var (
-	if shouldGrow && !vl.HasGrowRequest() {
+		lastErr    error
-		// if picked volume is almost full, trigger a volume-grow request
+		maxTimeout = time.Second * 10
-		glog.V(0).Infof("dirAssign volume growth %v from %v", option.String(), r.RemoteAddr)
+		startTime  = time.Now()
-		if ms.Topo.AvailableSpaceFor(option) <= 0 {
+	)
-			writeJsonQuiet(w, r, http.StatusNotFound, operation.AssignResult{Error: "No free volumes left for " + option.String()})
-			return
-		}
 
-		errCh := make(chan error, 1)
+	for time.Now().Sub(startTime) < maxTimeout {
-		vl.AddGrowRequest()
+		fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(requestedCount, option, vl)
-		ms.vgCh <- &topology.VolumeGrowRequest{
+		if shouldGrow && !vl.HasGrowRequest() {
-			Option: option,
+			// if picked volume is almost full, trigger a volume-grow request
-			Count:  writableVolumeCount,
+			glog.V(0).Infof("dirAssign volume growth %v from %v", option.String(), r.RemoteAddr)
-			ErrCh:  errCh,
+			if ms.Topo.AvailableSpaceFor(option) <= 0 {
+				writeJsonQuiet(w, r, http.StatusNotFound, operation.AssignResult{Error: "No free volumes left for " + option.String()})
+				return
+			}
+
+			errCh := make(chan error, 1)
+			vl.AddGrowRequest()
+			ms.vgCh <- &topology.VolumeGrowRequest{
+				Option: option,
+				Count:  writableVolumeCount,
+				ErrCh:  errCh,
+			}
+			if err := <-errCh; err != nil {
+				writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("cannot grow volume group! %v", err))
+				return
+			}
 		}
-		if err := <-errCh; err != nil {
+		if err != nil {
-			writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("cannot grow volume group! %v", err))
+			// glog.Warningf("PickForWrite %+v: %v", req, err)
+			lastErr = err
+			time.Sleep(200 * time.Millisecond)
+			continue
+		} else {
+			ms.maybeAddJwtAuthorization(w, fid, true)
+			dn := dnList.Head()
+
+			writeJsonQuiet(w, r, http.StatusOK, operation.AssignResult{Fid: fid, Url: dn.Url(), PublicUrl: dn.PublicUrl, Count: count})
 			return
 		}
 	}
-	if err == nil {
-		ms.maybeAddJwtAuthorization(w, fid, true)
-		dn := dnList.Head()
 
-		writeJsonQuiet(w, r, http.StatusOK, operation.AssignResult{Fid: fid, Url: dn.Url(), PublicUrl: dn.PublicUrl, Count: count})
+	if lastErr != nil {
+		writeJsonQuiet(w, r, http.StatusNotAcceptable, operation.AssignResult{Error: lastErr.Error()})
 	} else {
-		writeJsonQuiet(w, r, http.StatusNotAcceptable, operation.AssignResult{Error: err.Error()})
+		writeJsonQuiet(w, r, http.StatusRequestTimeout, operation.AssignResult{Error: "request timeout"})
 	}
 }