Merge pull request #16 from chrislusf/master

sync

10 files changed, 52 insertions, 36 deletions

diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go
index 20c2502b9..41dcedba5 100644
--- a/weed/server/filer_grpc_server.go
+++ b/weed/server/filer_grpc_server.go
@@ -164,6 +164,7 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
 		FullPath: util.JoinPath(req.Directory, req.Entry.Name),
 		Attr:     filer.PbToEntryAttribute(req.Entry.Attributes),
 		Chunks:   chunks,
+		Extended: req.Entry.Extended,
 	}, req.OExcl, req.IsFromOtherCluster, req.Signatures)
 
 	if createErr == nil {
@@ -420,13 +421,15 @@ func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsR
 func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.GetFilerConfigurationRequest) (resp *filer_pb.GetFilerConfigurationResponse, err error) {
 
 	t := &filer_pb.GetFilerConfigurationResponse{
-		Masters:     fs.option.Masters,
-		Collection:  fs.option.Collection,
-		Replication: fs.option.DefaultReplication,
-		MaxMb:       uint32(fs.option.MaxMB),
-		DirBuckets:  fs.filer.DirBucketsPath,
-		Cipher:      fs.filer.Cipher,
-		Signature:   fs.filer.Signature,
+		Masters:            fs.option.Masters,
+		Collection:         fs.option.Collection,
+		Replication:        fs.option.DefaultReplication,
+		MaxMb:              uint32(fs.option.MaxMB),
+		DirBuckets:         fs.filer.DirBucketsPath,
+		Cipher:             fs.filer.Cipher,
+		Signature:          fs.filer.Signature,
+		MetricsAddress:     fs.metricsAddress,
+		MetricsIntervalSec: int32(fs.metricsIntervalSec),
 	}
 
 	glog.V(4).Infof("GetFilerConfiguration: %v", t)
diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go
index 35df01665..f9ddeb600 100644
--- a/weed/server/filer_grpc_server_rename.go
+++ b/weed/server/filer_grpc_server_rename.go
@@ -109,6 +109,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
 		FullPath: newPath,
 		Attr:     entry.Attr,
 		Chunks:   entry.Chunks,
+		Extended: entry.Extended,
 	}
 	createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, nil)
 	if createErr != nil {
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 18809162a..4d6d424dc 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -62,6 +62,10 @@ type FilerServer struct {
 	filer          *filer.Filer
 	grpcDialOption grpc.DialOption
 
+	// metrics read from the master
+	metricsAddress     string
+	metricsIntervalSec int
+
 	// notifying clients
 	listenersLock sync.Mutex
 	listenersCond *sync.Cond
@@ -88,7 +92,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	})
 	fs.filer.Cipher = option.Cipher
 
-	maybeStartMetrics(fs, option)
+	fs.maybeStartMetrics()
 
 	go fs.filer.KeepConnectedToMaster()
 
@@ -131,9 +135,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	return fs, nil
 }
 
-func maybeStartMetrics(fs *FilerServer, option *FilerOption) {
+func (fs *FilerServer) maybeStartMetrics() {
 
-	for _, master := range option.Masters {
+	for _, master := range fs.option.Masters {
 		_, err := pb.ParseFilerGrpcAddress(master)
 		if err != nil {
 			glog.Fatalf("invalid master address %s: %v", master, err)
@@ -141,12 +145,10 @@ func maybeStartMetrics(fs *FilerServer, option *FilerOption) {
 	}
 
 	isConnected := false
-	var metricsAddress string
-	var metricsIntervalSec int
 	var readErr error
 	for !isConnected {
-		for _, master := range option.Masters {
-			metricsAddress, metricsIntervalSec, readErr = readFilerConfiguration(fs.grpcDialOption, master)
+		for _, master := range fs.option.Masters {
+			fs.metricsAddress, fs.metricsIntervalSec, readErr = readFilerConfiguration(fs.grpcDialOption, master)
 			if readErr == nil {
 				isConnected = true
 			} else {
@@ -154,10 +156,8 @@ func maybeStartMetrics(fs *FilerServer, option *FilerOption) {
 			}
 		}
 	}
-	if metricsAddress == "" && metricsIntervalSec <= 0 {
-		return
-	}
-	go stats.LoopPushingMetric("filer", stats.SourceName(option.Port), stats.FilerGather, metricsAddress, metricsIntervalSec)
+
+	go stats.LoopPushingMetric("filer", stats.SourceName(fs.option.Port), stats.FilerGather, fs.metricsAddress, fs.metricsIntervalSec)
 }
 
 func readFilerConfiguration(grpcDialOption grpc.DialOption, masterAddress string) (metricsAddress string, metricsIntervalSec int, err error) {
diff --git a/weed/server/filer_server_handlers.go b/weed/server/filer_server_handlers.go
index b6bfc3b04..18f78881c 100644
--- a/weed/server/filer_server_handlers.go
+++ b/weed/server/filer_server_handlers.go
@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
 	"time"
 
@@ -8,6 +9,7 @@ import (
 )
 
 func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 	start := time.Now()
 	switch r.Method {
 	case "GET":
@@ -34,6 +36,7 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 	start := time.Now()
 	switch r.Method {
 	case "GET":
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 93ecefb74..f3a2ee013 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -71,7 +71,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 				int64(heartbeat.MaxVolumeCount))
 			glog.V(0).Infof("added volume server %v:%d", heartbeat.GetIp(), heartbeat.GetPort())
 			if err := stream.Send(&master_pb.HeartbeatResponse{
-				VolumeSizeLimit:        uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
+				VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
 			}); err != nil {
 				glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
 				return err
diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go
index 0c0cc39c1..8698a4c64 100644
--- a/weed/server/volume_grpc_client_to_master.go
+++ b/weed/server/volume_grpc_client_to_master.go
@@ -24,21 +24,25 @@ func (vs *VolumeServer) GetMaster() string {
 }
 
 func (vs *VolumeServer) checkWithMaster() (err error) {
-	for _, master := range vs.SeedMasterNodes {
-		err = operation.WithMasterServerClient(master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
-			resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
-			if err != nil {
-				return fmt.Errorf("get master %s configuration: %v", master, err)
+	isConnected := false
+	for !isConnected {
+		for _, master := range vs.SeedMasterNodes {
+			err = operation.WithMasterServerClient(master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+				resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
+				if err != nil {
+					return fmt.Errorf("get master %s configuration: %v", master, err)
+				}
+				vs.metricsAddress, vs.metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
+				backend.LoadFromPbStorageBackends(resp.StorageBackends)
+				return nil
+			})
+			if err == nil {
+				return
+			} else {
+				glog.V(0).Infof("checkWithMaster %s: %v", master, err)
 			}
-			vs.MetricsAddress, vs.MetricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
-			backend.LoadFromPbStorageBackends(resp.StorageBackends)
-			return nil
-		})
-		if err == nil {
-			return
-		} else {
-			glog.V(0).Infof("checkWithMaster %s: %v", master, err)
 		}
+		time.Sleep(1790 * time.Millisecond)
 	}
 	return
 }
diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go
index c600da21e..5e9ec1369 100644
--- a/weed/server/volume_server.go
+++ b/weed/server/volume_server.go
@@ -28,8 +28,8 @@ type VolumeServer struct {
 	FixJpgOrientation       bool
 	ReadRedirect            bool
 	compactionBytePerSecond int64
-	MetricsAddress          string
-	MetricsIntervalSec      int
+	metricsAddress          string
+	metricsIntervalSec      int
 	fileSizeLimitBytes      int64
 	isHeartbeating          bool
 	stopChan                chan bool
@@ -97,8 +97,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	}
 
 	go vs.heartbeat()
-	hostAddress := fmt.Sprintf("%s:%d", ip, port)
-	go stats.LoopPushingMetric("volumeServer", hostAddress, stats.VolumeServerGather, vs.MetricsAddress, vs.MetricsIntervalSec)
+	go stats.LoopPushingMetric("volumeServer", fmt.Sprintf("%s:%d", ip, port), stats.VolumeServerGather, vs.metricsAddress, vs.metricsIntervalSec)
 
 	return vs
 }
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index 14ad27d42..ad13cdf3b 100644
--- a/weed/server/volume_server_handlers.go
+++ b/weed/server/volume_server_handlers.go
@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
 	"strings"
 
@@ -25,6 +26,7 @@ security settings:
 */
 
 func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	switch r.Method {
 	case "GET", "HEAD":
 		stats.ReadRequest()
@@ -39,6 +41,7 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 }
 
 func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	switch r.Method {
 	case "GET":
 		stats.ReadRequest()
diff --git a/weed/server/volume_server_handlers_admin.go b/weed/server/volume_server_handlers_admin.go
index 34655d833..4d84c9c4d 100644
--- a/weed/server/volume_server_handlers_admin.go
+++ b/weed/server/volume_server_handlers_admin.go
@@ -10,6 +10,7 @@ import (
 )
 
 func (vs *VolumeServer) statusHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	m := make(map[string]interface{})
 	m["Version"] = util.Version()
 	var ds []*volume_server_pb.DiskStatus
@@ -24,6 +25,7 @@ func (vs *VolumeServer) statusHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (vs *VolumeServer) statsDiskHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	m := make(map[string]interface{})
 	m["Version"] = util.Version()
 	var ds []*volume_server_pb.DiskStatus
diff --git a/weed/server/volume_server_handlers_ui.go b/weed/server/volume_server_handlers_ui.go
index 8b2027e7b..e535327e2 100644
--- a/weed/server/volume_server_handlers_ui.go
+++ b/weed/server/volume_server_handlers_ui.go
@@ -13,6 +13,7 @@ import (
 )
 
 func (vs *VolumeServer) uiStatusHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	infos := make(map[string]interface{})
 	infos["Up Time"] = time.Now().Sub(startTime).String()
 	var ds []*volume_server_pb.DiskStatus