diff --git a/other/metrics/grafana_seaweedfs.json b/other/metrics/grafana_seaweedfs.json index 184b35c65..9652c96c2 100644 --- a/other/metrics/grafana_seaweedfs.json +++ b/other/metrics/grafana_seaweedfs.json @@ -5869,8 +5869,8 @@ "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, - "title": "Filer Sync Offset Lag", - "description": "Seconds behind source (filer.sync); only present when filer sync runs", + "title": "Filer Sync Lag", + "description": "Seconds the replicated watermark trails the source filer (filer.sync); only present when filer sync runs", "type": "timeseries", "id": 155, "gridPos": { @@ -5952,10 +5952,10 @@ "uid": "${DS_PROMETHEUS}" }, "editorMode": "code", - "expr": "time() - (SeaweedFS_filerSync_sync_offset{cluster=~\"$cluster\"} > 0) / 1e9", + "expr": "SeaweedFS_filerSync_lag_seconds{cluster=~\"$cluster\"}", "range": true, "refId": "A", - "legendFormat": "{{clientName}} {{path}}" + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}}" } ], "pluginVersion": "10.3.1" @@ -6055,6 +6055,423 @@ } ], "pluginVersion": "10.3.1" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "title": "Filer Sync Event Rate", + "description": "Metadata events read off the source stream, replicated, and failed after retries (filer.sync)", + "type": "timeseries", + "id": 230, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 40 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "unit": "ops", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(SeaweedFS_filerSync_events_received_total{cluster=~\"$cluster\"}[$__rate_interval])", + "range": true, + "refId": "A", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}} received" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(SeaweedFS_filerSync_events_processed_total{cluster=~\"$cluster\"}[$__rate_interval])", + "range": true, + "refId": "B", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}} processed" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(SeaweedFS_filerSync_events_failed_total{cluster=~\"$cluster\"}[$__rate_interval])", + "range": true, + "refId": "C", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}} failed" + } + ], + "pluginVersion": "10.3.1" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "title": "Filer Sync Throughput", + "description": "Chunk data bytes carried by sync events: the chunk delta, so deletes, renames, and attribute-only updates count zero (filer.sync)", + "type": "timeseries", + "id": 231, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 40 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "unit": "Bps", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(SeaweedFS_filerSync_received_bytes_total{cluster=~\"$cluster\"}[$__rate_interval])", + "range": true, + "refId": "A", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}} received" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(SeaweedFS_filerSync_processed_bytes_total{cluster=~\"$cluster\"}[$__rate_interval])", + "range": true, + "refId": "B", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}} processed" + } + ], + "pluginVersion": "10.3.1" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "title": "Filer Sync In-flight Jobs", + "description": "Jobs currently replicating; pinned at -concurrency means the sync itself is the bottleneck, near zero means caught up or starved by the source (filer.sync)", + "type": "timeseries", + "id": 232, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 48 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "unit": "short", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "SeaweedFS_filerSync_in_flight_jobs{cluster=~\"$cluster\"}", + "range": true, + "refId": "A", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}}" + } + ], + "pluginVersion": "10.3.1" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "title": "Filer Sync In-flight Bytes", + "description": "Chunk data carried by the jobs currently replicating; distinguishes workers stuck on a few large files from many small ones (filer.sync)", + "type": "timeseries", + "id": 233, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 48 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "unit": "bytes", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "SeaweedFS_filerSync_in_flight_bytes{cluster=~\"$cluster\"}", + "range": true, + "refId": "A", + "legendFormat": "{{sourceFiler}} -> {{targetFiler}} {{path}}" + } + ], + "pluginVersion": "10.3.1" } ] }, diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index e12ac6dc1..64aebcc6f 100644 --- a/weed/command/filer_sync.go +++ b/weed/command/filer_sync.go @@ -382,7 +382,9 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi glog.Warningf("invalid concurrency value, using default: %d", DefaultConcurrencyLimit) concurrency = DefaultConcurrencyLimit } + clientName := fmt.Sprintf("syncFrom_%s_To_%s", string(sourceFiler), string(targetFiler)) processor := NewMetadataProcessor(processEventFn, concurrency, sourceFilerOffsetTsNs) + processor.SetMetrics(sourceFiler.String(), targetFiler.String(), clientName, sourcePath) // update sync state for graceful shutdown checkpoint saving if statePtr != nil { @@ -396,9 +398,38 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi }) } + // The offset callback below only fires while events flow, so it freezes + // exactly when the workers are saturated; a ticker keeps lag honest. The + // idle heartbeat is folded in because the watermark stops at the last real + // event, so a quiet caught-up stream would otherwise show phantom lag. + lagGauge := statsCollect.FilerSyncLagSecondsGauge.WithLabelValues(sourceFiler.String(), targetFiler.String(), clientName, sourcePath) + var idleHeartbeatTsNs atomic.Int64 + stopLagTicker := make(chan struct{}) + defer close(stopLagTicker) + go func() { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-stopLagTicker: + return + case <-ticker.C: + freshnessTsNs := processor.processedTsWatermark.Load() + // a heartbeat means consumed, not replicated: while a failure + // pins the watermark it must not mask the growing lag + if processor.OldestFailedTsNs() == 0 { + freshnessTsNs = max(freshnessTsNs, idleHeartbeatTsNs.Load()) + } + if freshnessTsNs == 0 { + continue + } + lagGauge.Set(max(0, time.Since(time.Unix(0, freshnessTsNs)).Seconds())) + } + } + }() + var lastLogTsNs = time.Now().UnixNano() var lastProgressedTsNs int64 - var clientName = fmt.Sprintf("syncFrom_%s_To_%s", string(sourceFiler), string(targetFiler)) processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error { processor.AddSyncJob(resp) return nil @@ -449,7 +480,13 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi // The idle heartbeat moves the gauge to the source's current time once we // are caught up, so now-sync_offset reflects real lag and stays alertable. OnIdleHeartbeat: func(tsNs int64) { + // same masking concern as the lag ticker: a pinned failure must + // keep now-sync_offset growing too + if processor.OldestFailedTsNs() != 0 { + return + } statsCollect.FilerSyncOffsetGauge.WithLabelValues(sourceFiler.String(), targetFiler.String(), clientName, sourcePath).Set(float64(tsNs)) + idleHeartbeatTsNs.Store(tsNs) }, } diff --git a/weed/command/filer_sync_jobs.go b/weed/command/filer_sync_jobs.go index 396b2af2c..732c1252d 100644 --- a/weed/command/filer_sync_jobs.go +++ b/weed/command/filer_sync_jobs.go @@ -7,9 +7,12 @@ import ( "sync/atomic" "time" + "github.com/prometheus/client_golang/prometheus" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + statsCollect "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -51,9 +54,23 @@ const ( ) type syncJobPaths struct { - path util.FullPath - newPath util.FullPath // empty for non-renames - kind jobKind + path util.FullPath + newPath util.FullPath // empty for non-renames + kind jobKind + dataSize int64 +} + +// syncStreamMetrics holds the metric children for one sync stream, curried +// once so per-event updates skip the label lookup. +type syncStreamMetrics struct { + received prometheus.Counter + processed prometheus.Counter + failed prometheus.Counter + inFlight prometheus.Gauge + receivedBytes prometheus.Counter + processedBytes prometheus.Counter + failedBytes prometheus.Counter + inFlightBytes prometheus.Gauge } type MetadataProcessor struct { @@ -93,6 +110,9 @@ type MetadataProcessor struct { // past it, so the persisted sync offset stays behind the failure and a // restart replays the event instead of skipping it forever. oldestFailedTsNs int64 + + // metrics is nil for callers that do not report per-event metrics. + metrics *syncStreamMetrics } func NewMetadataProcessor(fn pb.ProcessMetadataFunc, concurrency int, offsetTsNs int64) *MetadataProcessor { @@ -110,6 +130,29 @@ func NewMetadataProcessor(fn pb.ProcessMetadataFunc, concurrency int, offsetTsNs return t } +// SetMetrics enables per-event metrics for this stream, labeled the same way +// as the existing sync_offset gauge. +func (t *MetadataProcessor) SetMetrics(sourceFiler, targetFiler, clientName, path string) { + t.metrics = &syncStreamMetrics{ + received: statsCollect.FilerSyncEventsReceivedCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + processed: statsCollect.FilerSyncEventsProcessedCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + failed: statsCollect.FilerSyncEventsFailedCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + inFlight: statsCollect.FilerSyncInFlightJobsGauge.WithLabelValues(sourceFiler, targetFiler, clientName, path), + receivedBytes: statsCollect.FilerSyncReceivedBytesCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + processedBytes: statsCollect.FilerSyncProcessedBytesCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + failedBytes: statsCollect.FilerSyncFailedBytesCounter.WithLabelValues(sourceFiler, targetFiler, clientName, path), + inFlightBytes: statsCollect.FilerSyncInFlightBytesGauge.WithLabelValues(sourceFiler, targetFiler, clientName, path), + } +} + +// OldestFailedTsNs returns the timestamp of the oldest permanently failed +// event, or 0 when none is pinning the watermark. +func (t *MetadataProcessor) OldestFailedTsNs() int64 { + t.activeJobsLock.Lock() + defer t.activeJobsLock.Unlock() + return t.oldestFailedTsNs +} + // pathAncestors returns all proper ancestor directories of p. // For "/a/b/c", returns ["/a/b", "/a", "/"]. func pathAncestors(p util.FullPath) []util.FullPath { @@ -241,6 +284,15 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse) return } + dataSize := eventDataSize(resp) + + // counted before the admission wait: received-processed-failed is the + // number of events read off the stream but not yet done + if t.metrics != nil { + t.metrics.received.Inc() + t.metrics.receivedBytes.Add(float64(dataSize)) + } + t.activeJobsLock.Lock() defer t.activeJobsLock.Unlock() @@ -249,13 +301,20 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse) } p, newPath, kind := extractJobInfo(resp) - jobPaths := &syncJobPaths{path: p, newPath: newPath, kind: kind} + jobPaths := &syncJobPaths{path: p, newPath: newPath, kind: kind, dataSize: dataSize} t.activeJobs[resp.TsNs] = jobPaths t.addPathToIndex(p, kind) if newPath != "" { t.addPathToIndex(newPath, kind) } + // Inc/Dec rather than Set from local state: after a subscription retry a + // new processor shares these children with the old one's still-draining + // jobs, and each job accounting for itself keeps the total truthful. + if t.metrics != nil { + t.metrics.inFlight.Inc() + t.metrics.inFlightBytes.Add(float64(dataSize)) + } heap.Push(&t.tsHeap, resp.TsNs) @@ -282,6 +341,17 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse) if jobPaths.newPath != "" { t.removePathFromIndex(jobPaths.newPath, jobPaths.kind) } + if t.metrics != nil { + if jobErr != nil { + t.metrics.failed.Inc() + t.metrics.failedBytes.Add(float64(jobPaths.dataSize)) + } else { + t.metrics.processed.Inc() + t.metrics.processedBytes.Add(float64(jobPaths.dataSize)) + } + t.metrics.inFlight.Dec() + t.metrics.inFlightBytes.Sub(float64(jobPaths.dataSize)) + } // Lazy-clean stale entries from heap top (already-completed jobs). // Each entry is pushed once and popped once: O(log n) amortized. @@ -303,6 +373,25 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse) }() } +// eventDataSize is the chunk data this event will copy: chunks on the new +// entry that the old entry does not already have. Deletes, renames, and +// attribute-only updates all come out zero, so byte rates reflect data +// movement rather than metadata churn. +func eventDataSize(resp *filer_pb.SubscribeMetadataResponse) (size int64) { + message := resp.EventNotification + if message.NewEntry == nil { + return 0 + } + newChunks := message.NewEntry.GetChunks() + if message.OldEntry != nil { + newChunks = filer.DoMinusChunks(newChunks, message.OldEntry.GetChunks()) + } + for _, chunk := range newChunks { + size += int64(chunk.Size) + } + return size +} + // extractJobInfo derives the conflict-detection path(s) and job kind for a // metadata event. A rename returns both the source and destination paths; all // other event shapes return only the primary path. diff --git a/weed/command/filer_sync_jobs_test.go b/weed/command/filer_sync_jobs_test.go index 1c6a1f3ce..131bff00f 100644 --- a/weed/command/filer_sync_jobs_test.go +++ b/weed/command/filer_sync_jobs_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -684,3 +685,60 @@ func TestFailedJobHoldsWatermarkAtOldestFailure(t *testing.T) { t.Fatalf("watermark = %d, want it held at 100 by the failure at 200", got) } } + +// TestSyncStreamMetrics verifies the per-event and byte counters and the +// in-flight gauges across success and failure outcomes. Bytes count only the +// chunk delta: the failing create's 40, the successful create's 100, the +// update's new 60-byte chunk but not its shared one, and nothing for the +// delete despite its chunk. +func TestSyncStreamMetrics(t *testing.T) { + fn := func(resp *filer_pb.SubscribeMetadataResponse) error { + if resp.TsNs == 2 { + return errors.New("AccessDenied: Access Denied") + } + return nil + } + p := NewMetadataProcessor(fn, 100, 0) + p.SetMetrics("srcFiler", "dstFiler", "TestSyncStreamMetrics", "/") + + create := makeResp("/dir1", "a.txt", false, 1, true) + create.EventNotification.NewEntry.Chunks = []*filer_pb.FileChunk{{FileId: "1,a0", Size: 100}} + failing := makeResp("/dir1", "b.txt", false, 2, true) + failing.EventNotification.NewEntry.Chunks = []*filer_pb.FileChunk{{FileId: "2,b0", Size: 40}} + shared := &filer_pb.FileChunk{FileId: "3,c0", Size: 30} + update := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir1", + TsNs: 3, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "c.txt", Chunks: []*filer_pb.FileChunk{shared}}, + NewEntry: &filer_pb.Entry{Name: "c.txt", Chunks: []*filer_pb.FileChunk{shared, {FileId: "3,c1", Size: 60}}}, + NewParentPath: "/dir1", + }, + } + del := makeResp("/dir1", "d.txt", false, 4, false) + del.EventNotification.OldEntry.Chunks = []*filer_pb.FileChunk{{FileId: "4,d0", Size: 999}} + + for _, resp := range []*filer_pb.SubscribeMetadataResponse{create, failing, update, del} { + p.AddSyncJob(resp) + } + waitForJobsToDrain(t, p) + + for _, tc := range []struct { + name string + got float64 + want float64 + }{ + {"received", testutil.ToFloat64(p.metrics.received), 4}, + {"processed", testutil.ToFloat64(p.metrics.processed), 3}, + {"failed", testutil.ToFloat64(p.metrics.failed), 1}, + {"in_flight", testutil.ToFloat64(p.metrics.inFlight), 0}, + {"received_bytes", testutil.ToFloat64(p.metrics.receivedBytes), 200}, + {"processed_bytes", testutil.ToFloat64(p.metrics.processedBytes), 160}, + {"failed_bytes", testutil.ToFloat64(p.metrics.failedBytes), 40}, + {"in_flight_bytes", testutil.ToFloat64(p.metrics.inFlightBytes), 0}, + } { + if tc.got != tc.want { + t.Errorf("%s = %v, want %v", tc.name, tc.got, tc.want) + } + } +} diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index ad59ed011..5846d0adf 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -301,6 +301,78 @@ var ( Help: "The offset of the filer synchronization service.", }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + FilerSyncEventsReceivedCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "events_received_total", + Help: "Counter of metadata events read off the source subscription stream.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncEventsProcessedCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "events_processed_total", + Help: "Counter of metadata events successfully replicated to the target.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncEventsFailedCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "events_failed_total", + Help: "Counter of metadata events that failed after retries; the sync offset is held at the oldest failure so it is replayed on restart.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncInFlightJobsGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "in_flight_jobs", + Help: "Number of sync jobs currently being replicated; pinned at the concurrency limit means the sync itself is the bottleneck.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncReceivedBytesCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "received_bytes_total", + Help: "Counter of chunk data bytes carried by received events: new chunks the old entry does not already have, so deletes, renames, and attribute-only updates count zero.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncProcessedBytesCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "processed_bytes_total", + Help: "Counter of chunk data bytes carried by successfully replicated events.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncFailedBytesCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "failed_bytes_total", + Help: "Counter of chunk data bytes carried by events that failed after retries.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncInFlightBytesGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "in_flight_bytes", + Help: "Chunk data bytes carried by the jobs currently being replicated; distinguishes workers stuck on a few large files from many small ones.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + + FilerSyncLagSecondsGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemFilerSync, + Name: "lag_seconds", + Help: "How far the replicated watermark trails the source filer, in seconds.", + }, []string{"sourceFiler", "targetFiler", "clientName", "path"}) + VolumeServerStartTimeSeconds = prometheus.NewGauge( prometheus.GaugeOpts{ Namespace: Namespace, @@ -912,6 +984,15 @@ func init() { Gather.MustRegister(FilerStoreCounter) Gather.MustRegister(FilerStoreHistogram) Gather.MustRegister(FilerSyncOffsetGauge) + Gather.MustRegister(FilerSyncEventsReceivedCounter) + Gather.MustRegister(FilerSyncEventsProcessedCounter) + Gather.MustRegister(FilerSyncEventsFailedCounter) + Gather.MustRegister(FilerSyncInFlightJobsGauge) + Gather.MustRegister(FilerSyncLagSecondsGauge) + Gather.MustRegister(FilerSyncReceivedBytesCounter) + Gather.MustRegister(FilerSyncProcessedBytesCounter) + Gather.MustRegister(FilerSyncFailedBytesCounter) + Gather.MustRegister(FilerSyncInFlightBytesGauge) Gather.MustRegister(FilerServerLastSendTsOfSubscribeGauge) Gather.MustRegister(FilerSubscribeGapStalledGauge) Gather.MustRegister(FilerSubscribeUnprovenGapCrossings)