aboutsummaryrefslogtreecommitdiff
path: root/weed
diff options
context:
space:
mode:
Diffstat (limited to 'weed')
-rw-r--r--weed/command/filer_sync.go25
-rw-r--r--weed/s3api/auth_credentials.go10
-rw-r--r--weed/server/filer_grpc_server_sub_meta.go4
-rw-r--r--weed/stats/metrics.go18
-rw-r--r--weed/storage/needle/needle_read_page.go7
-rw-r--r--weed/util/constants.go2
6 files changed, 54 insertions, 12 deletions
diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go
index 7aa9c1e8d..b7da1baf9 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 {
@@ -182,7 +188,7 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
// if first time, start from now
// if has previously synced, resume from that point of time
- sourceFilerOffsetTsNs, err := getOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature)
+ sourceFilerOffsetTsNs, err := getOffset(grpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature)
if err != nil {
return err
}
@@ -210,14 +216,17 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
}
var lastLogTsNs = time.Now().Nanosecond()
+ var clientName = fmt.Sprintf("syncFrom_%s_To_%s", string(sourceFiler), string(targetFiler))
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
- return setOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature, lastTsNs)
+ // 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)
}
@@ -226,6 +235,16 @@ const (
SyncKeyPrefix = "sync."
)
+// When each business is distinguished according to path, and offsets need to be maintained separately.
+func getSignaturePrefixByPath(path string) string {
+ // compatible historical version
+ if path == "/" {
+ return SyncKeyPrefix
+ } else {
+ return SyncKeyPrefix + path
+ }
+}
+
func getOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32) (lastOffsetTsNs int64, readErr error) {
readErr = pb.WithFilerClient(false, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go
index 757e3ddd9..fb23d9ce9 100644
--- a/weed/s3api/auth_credentials.go
+++ b/weed/s3api/auth_credentials.go
@@ -176,12 +176,12 @@ func (iam *IdentityAccessManagement) lookupAnonymous() (identity *Identity, foun
}
func (iam *IdentityAccessManagement) Auth(f http.HandlerFunc, action Action) http.HandlerFunc {
-
- if !iam.isEnabled() {
- return f
- }
-
return func(w http.ResponseWriter, r *http.Request) {
+ if !iam.isEnabled() {
+ f(w, r)
+ return
+ }
+
identity, errCode := iam.authRequest(r, action)
if errCode == s3err.ErrNone {
if identity != nil && identity.Name != "" {
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{}))
diff --git a/weed/storage/needle/needle_read_page.go b/weed/storage/needle/needle_read_page.go
index c00195e93..e909869f3 100644
--- a/weed/storage/needle/needle_read_page.go
+++ b/weed/storage/needle/needle_read_page.go
@@ -14,9 +14,10 @@ func (n *Needle) ReadNeedleDataInto(r backend.BackendStorageFile, volumeOffset i
crc := CRC(0)
for x := needleOffset; x < needleOffset+size; x += int64(len(buf)) {
count, err := n.ReadNeedleData(r, volumeOffset, buf, x)
- if count > 0 {
- crc = crc.Update(buf[0:count])
- if _, err = writer.Write(buf[0:count]); err != nil {
+ toWrite := min(int64(count), needleOffset+size-x)
+ if toWrite > 0 {
+ crc = crc.Update(buf[0:toWrite])
+ if _, err = writer.Write(buf[0:toWrite]); err != nil {
return fmt.Errorf("ReadNeedleData write: %v", err)
}
}
diff --git a/weed/util/constants.go b/weed/util/constants.go
index 213eafae0..c0fea8b17 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -5,7 +5,7 @@ import (
)
var (
- VERSION_NUMBER = fmt.Sprintf("%.02f", 3.10)
+ VERSION_NUMBER = fmt.Sprintf("%.02f", 3.11)
VERSION = sizeLimit + " " + VERSION_NUMBER
COMMIT = ""
)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 05:33:28 +0000