diff options
| -rw-r--r-- | weed/command/filer_sync.go | 11 | ||||
| -rw-r--r-- | weed/server/filer_grpc_server_sub_meta.go | 4 | ||||
| -rw-r--r-- | weed/stats/metrics.go | 18 |
3 files changed, 32 insertions, 1 deletions
diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index dc7c569c3..dca4e693f 100644 --- a/weed/command/filer_sync.go +++ b/weed/command/filer_sync.go @@ -12,6 +12,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink" "github.com/chrislusf/seaweedfs/weed/replication/source" "github.com/chrislusf/seaweedfs/weed/security" + statsCollect "github.com/chrislusf/seaweedfs/weed/stats" "github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util/grace" "google.golang.org/grpc" @@ -40,6 +41,7 @@ type SyncOptions struct { bFromTsMs *int64 aProxyByFiler *bool bProxyByFiler *bool + metricsHttpPort *int clientId int32 } @@ -72,6 +74,7 @@ func init() { syncOptions.bFromTsMs = cmdFilerSynchronize.Flag.Int64("b.fromTsMs", 0, "synchronization from timestamp on filer B. The unit is millisecond") syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file") syncMemProfile = cmdFilerSynchronize.Flag.String("memprofile", "", "memory profile output file") + syncOptions.metricsHttpPort = cmdFilerSynchronize.Flag.Int("metricsPort", 0, "metrics listen port") syncOptions.clientId = util.RandomInt32() } @@ -103,6 +106,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool { filerA := pb.ServerAddress(*syncOptions.filerA) filerB := pb.ServerAddress(*syncOptions.filerB) + // start filer.sync metrics server + go statsCollect.StartMetricsServer(*syncOptions.metricsHttpPort) + // read a filer signature aFilerSignature, aFilerErr := replication.ReadFilerSignature(grpcDialOption, filerA) if aFilerErr != nil { @@ -210,14 +216,17 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption, } var lastLogTsNs = time.Now().Nanosecond() + var clientName = fmt.Sprintf("syncTo_%s_From_%s", string(targetFiler), string(sourceFiler)) processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error { now := time.Now().Nanosecond() glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, lastTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9)) lastLogTsNs = now + // collect synchronous offset + statsCollect.FilerSyncOffsetGauge.WithLabelValues(sourceFiler.String(), targetFiler.String(), clientName, sourcePath).Set(float64(lastTsNs)) return setOffset(grpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature, lastTsNs) }) - return pb.FollowMetadata(sourceFiler, grpcDialOption, "syncTo_"+string(targetFiler), clientId, + return pb.FollowMetadata(sourceFiler, grpcDialOption, clientName, clientId, sourcePath, nil, sourceFilerOffsetTsNs, 0, targetFilerSignature, processEventFnWithOffset, pb.RetryForeverOnError) } 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/stats/metrics.go b/weed/stats/metrics.go index 943aafff9..207b37c81 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -77,6 +77,14 @@ var ( Buckets: prometheus.ExponentialBuckets(0.0001, 2, 24), }, []string{"type"}) + FilerServerLastSendTsOfSubscribeGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "SeaweedFS", + Subsystem: "filer", + Name: "last_send_timestamp_of_subscribe", + Help: "The last send timestamp of the filer subscription.", + }, []string{"sourceFiler", "clientName", "path"}) + FilerStoreCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: "SeaweedFS", @@ -94,6 +102,14 @@ var ( Buckets: prometheus.ExponentialBuckets(0.0001, 2, 24), }, []string{"store", "type"}) + FilerSyncOffsetGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "SeaweedFS", + Subsystem: "filerSync", + Name: "sync_offset", + Help: "The offset of the filer synchronization service.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + VolumeServerRequestCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: "SeaweedFS", @@ -179,6 +195,8 @@ func init() { Gather.MustRegister(FilerRequestHistogram) Gather.MustRegister(FilerStoreCounter) Gather.MustRegister(FilerStoreHistogram) + Gather.MustRegister(FilerSyncOffsetGauge) + Gather.MustRegister(FilerServerLastSendTsOfSubscribeGauge) Gather.MustRegister(collectors.NewGoCollector()) Gather.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) |