Merge branch 'master' into circuit_breaker

7 files changed, 62 insertions, 30 deletions

diff --git a/weed/server/common.go b/weed/server/common.go
index 39a8637ac..f02ec67ac 100644
--- a/weed/server/common.go
+++ b/weed/server/common.go
@@ -284,6 +284,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 	if rangeReq == "" {
 		w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
 		if err := writeFn(bufferedWriter, 0, totalSize); err != nil {
+			glog.Errorf("processRangeRequest headers: %+v err: %v", w.Header(), err)
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
@@ -294,6 +295,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 	//mostly copy from src/pkg/net/http/fs.go
 	ranges, err := parseRange(rangeReq, totalSize)
 	if err != nil {
+		glog.Errorf("processRangeRequest headers: %+v err: %v", w.Header(), err)
 		http.Error(w, err.Error(), http.StatusRequestedRangeNotSatisfiable)
 		return
 	}
@@ -326,6 +328,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 		w.WriteHeader(http.StatusPartialContent)
 		err = writeFn(bufferedWriter, ra.start, ra.length)
 		if err != nil {
+			glog.Errorf("processRangeRequest headers: %+v err: %v", w.Header(), err)
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
@@ -365,6 +368,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 	}
 	w.WriteHeader(http.StatusPartialContent)
 	if _, err := io.CopyN(bufferedWriter, sendContent, sendSize); err != nil {
+		glog.Errorf("processRangeRequest err: %v", err)
 		http.Error(w, "Internal Error", http.StatusInternalServerError)
 		return
 	}
diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index 0540400a3..da710234b 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/stats"
 	"strings"
 	"time"
 
@@ -229,6 +230,9 @@ func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRe
 			}
 		}
 
+		// collect timestamps for path
+		stats.FilerServerLastSendTsOfSubscribeGauge.WithLabelValues(fs.option.Host.String(), req.ClientName, req.PathPrefix).Set(float64(tsNs))
+
 		message := &filer_pb.SubscribeMetadataResponse{
 			Directory:         dirPath,
 			EventNotification: eventNotification,
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 8908b5e5f..6bf0261ee 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -3,7 +3,6 @@ package weed_server
 import (
 	"context"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"net/http"
 	"os"
 	"sync"
@@ -17,6 +16,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 
@@ -67,6 +67,7 @@ type FilerOption struct {
 	Cipher                bool
 	SaveToFilerLimit      int64
 	ConcurrentUploadLimit int64
+	ShowUIDirectoryDelete bool
 }
 
 type FilerServer struct {
diff --git a/weed/server/filer_server_handlers_read_dir.go b/weed/server/filer_server_handlers_read_dir.go
index 8382cfc76..eaf17fa18 100644
--- a/weed/server/filer_server_handlers_read_dir.go
+++ b/weed/server/filer_server_handlers_read_dir.go
@@ -73,7 +73,7 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	ui.StatusTpl.Execute(w, struct {
+	err = ui.StatusTpl.Execute(w, struct {
 		Path                  string
 		Breadcrumbs           []ui.Breadcrumb
 		Entries               interface{}
@@ -81,6 +81,7 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque
 		LastFileName          string
 		ShouldDisplayLoadMore bool
 		EmptyFolder           bool
+		ShowDirectoryDelete   bool
 	}{
 		path,
 		ui.ToBreadcrumb(path),
@@ -89,5 +90,9 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque
 		lastFileName,
 		shouldDisplayLoadMore,
 		emptyFolder,
+		fs.option.ShowUIDirectoryDelete,
 	})
+	if err != nil {
+		glog.V(0).Infof("Template Execute Error: %v", err)
+	}
 }
diff --git a/weed/server/filer_ui/filer.html b/weed/server/filer_ui/filer.html
index f9c35440e..c9d832e8f 100644
--- a/weed/server/filer_ui/filer.html
+++ b/weed/server/filer_ui/filer.html
@@ -109,38 +109,37 @@
         <form class="upload-form">
             <input type="file" id="fileElem" multiple onchange="handleFiles(this.files)">
 
-            {{if .EmptyFolder}}
+            {{ if .EmptyFolder }}
             <div class="row add-files">
                 +
             </div>
-            {{else}}
+            {{ else }}
             <table width="100%" class="table table-hover">
-                {{$path := .Path }}
+                {{ $path := .Path }}
+                {{ $showDirDel := .ShowDirectoryDelete }}
                 {{ range $entry_index, $entry := .Entries }}
                 <tr>
                     <td>
-                        {{if $entry.IsDirectory}}
+                        {{ if $entry.IsDirectory }}
                         <span class="glyphicon glyphicon-folder-open" aria-hidden="true"></span>&nbsp;
                         <a href="{{ printpath $path  "/" $entry.Name "/"}}" >
                         {{ $entry.Name }}
                         </a>
-                        {{else}}
+                        {{ else }}
                         <a href="{{ printpath $path  "/" $entry.Name }}" >
                         {{ $entry.Name }}
                         </a>
-                        {{end}}
+                        {{ end }}
                     </td>
                     <td align="right" nowrap>
-                        {{if $entry.IsDirectory}}
-                        {{else}}
+                        {{ if not $entry.IsDirectory }}
                         {{ $entry.Mime }}&nbsp;
-                        {{end}}
+                        {{ end }}
                     </td>
                     <td align="right" nowrap>
-                        {{if $entry.IsDirectory}}
-                        {{else}}
+                        {{ if not $entry.IsDirectory }}
                         {{ $entry.Size | humanizeBytes }}&nbsp;
-                        {{end}}
+                        {{ end }}
                     </td>
                     <td align="right" nowrap>
                         {{ $entry.Timestamp.Format "2006-01-02 15:04" }}
@@ -150,31 +149,32 @@
                             <label class="btn" onclick="handleRename('{{ $entry.Name }}', '{{ printpath $path "/" }}')">
                                 <span class="glyphicon glyphicon-edit" aria-hidden="true"></span>
                             </label>
-                            {{if $entry.IsDirectory}}
-                            <label class="btn" onclick="handleDelete('{{ printpath $path  "/" $entry.Name "/"  }}')">
+                            {{ if and $entry.IsDirectory $showDirDel }}
+                            <label class="btn" onclick="handleDelete('{{ printpath $path  "/" $entry.Name "/" }}')">
                                 <span class="glyphicon glyphicon-trash" aria-hidden="true"></span>
                             </label>
-                            {{else}}
+                            {{ end }}
+                            {{ if not $entry.IsDirectory }}
                             <label class="btn" onclick="handleDelete('{{ printpath $path  "/" $entry.Name }}')">
                                 <span class="glyphicon glyphicon-trash" aria-hidden="true"></span>
                             </label>
-                            {{end}}
+                            {{ end }}
                         </div>
                     </td>
                 </tr>
                 {{ end }}
             </table>
-            {{end}}
+            {{ end }}
         </form>
     </div>
 
-    {{if .ShouldDisplayLoadMore}}
+    {{ if .ShouldDisplayLoadMore }}
     <div class="row">
-        <a href={{ print .Path "?limit=" .Limit "&lastFileName=" .LastFileName}} >
+        <a href={{ print .Path "?limit=" .Limit "&lastFileName=" .LastFileName }} >
         Load more
         </a>
     </div>
-    {{end}}
+    {{ end }}
 
     <br/>
     <br/>
diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go
index 477a3709c..abb30229a 100644
--- a/weed/server/volume_server.go
+++ b/weed/server/volume_server.go
@@ -3,6 +3,7 @@ package weed_server
 import (
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
@@ -24,7 +25,9 @@ type VolumeServer struct {
 	inFlightDownloadDataSize      int64
 	concurrentUploadLimit         int64
 	concurrentDownloadLimit       int64
+	inFlightUploadDataLimitCond   *sync.Cond
 	inFlightDownloadDataLimitCond *sync.Cond
+	inflightUploadDataTimeout     time.Duration
 
 	SeedMasterNodes []pb.ServerAddress
 	currentMaster   pb.ServerAddress
@@ -60,6 +63,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	fileSizeLimitMB int,
 	concurrentUploadLimit int64,
 	concurrentDownloadLimit int64,
+	inflightUploadDataTimeout time.Duration,
 ) *VolumeServer {
 
 	v := util.GetViper()
@@ -84,9 +88,11 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		fileSizeLimitBytes:            int64(fileSizeLimitMB) * 1024 * 1024,
 		isHeartbeating:                true,
 		stopChan:                      make(chan bool),
+		inFlightUploadDataLimitCond:   sync.NewCond(new(sync.Mutex)),
 		inFlightDownloadDataLimitCond: sync.NewCond(new(sync.Mutex)),
 		concurrentUploadLimit:         concurrentUploadLimit,
 		concurrentDownloadLimit:       concurrentDownloadLimit,
+		inflightUploadDataTimeout:     inflightUploadDataTimeout,
 	}
 	vs.SeedMasterNodes = masterNodes
 
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index 49bc297fb..293f36f14 100644
--- a/weed/server/volume_server_handlers.go
+++ b/weed/server/volume_server_handlers.go
@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 	"sync/atomic"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/util"
 
@@ -56,20 +57,31 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 		vs.guard.WhiteList(vs.DeleteHandler)(w, r)
 	case "PUT", "POST":
 
-		// wait until in flight data is less than the limit
 		contentLength := getContentLength(r)
-
 		// exclude the replication from the concurrentUploadLimitMB
-		if vs.concurrentUploadLimit != 0 && r.URL.Query().Get("type") != "replicate" &&
-			atomic.LoadInt64(&vs.inFlightUploadDataSize) > vs.concurrentUploadLimit {
-			err := fmt.Errorf("reject because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
-			glog.V(1).Infof("too many requests: %v", err)
-			writeJsonError(w, r, http.StatusTooManyRequests, err)
-			return
+		if r.URL.Query().Get("type") != "replicate" && vs.concurrentUploadLimit != 0 {
+			startTime := time.Now()
+			vs.inFlightUploadDataLimitCond.L.Lock()
+			for vs.inFlightUploadDataSize > vs.concurrentUploadLimit {
+				//wait timeout check
+				if startTime.Add(vs.inflightUploadDataTimeout).Before(time.Now()) {
+					vs.inFlightUploadDataLimitCond.L.Unlock()
+					err := fmt.Errorf("reject because inflight upload data %d > %d, and wait timeout", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
+					glog.V(1).Infof("too many requests: %v", err)
+					writeJsonError(w, r, http.StatusTooManyRequests, err)
+					return
+				}
+				glog.V(4).Infof("wait because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
+				vs.inFlightUploadDataLimitCond.Wait()
+			}
+			vs.inFlightUploadDataLimitCond.L.Unlock()
 		}
 		atomic.AddInt64(&vs.inFlightUploadDataSize, contentLength)
 		defer func() {
 			atomic.AddInt64(&vs.inFlightUploadDataSize, -contentLength)
+			if vs.concurrentUploadLimit != 0 {
+				vs.inFlightUploadDataLimitCond.Signal()
+			}
 		}()
 
 		// processs uploads