filer.sync: export replication lag, event counters, and in-flight jobs (#11069)

* filer.sync: count received, processed, and failed events and export in-flight jobs

The metadata processor admits at most -concurrency jobs and blocks the
subscription stream past that, so the backlog lives in the source filer's
metadata log and cannot be counted here. What can be measured honestly:
events read off the stream, replication outcomes, and worker saturation.
in_flight_jobs pinned at the concurrency limit means the sync itself is
the bottleneck; near zero means it is caught up or starved by the source.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

* filer.sync: export replication lag in seconds

Lag is now minus the freshest of the processed watermark and the last
idle heartbeat: the watermark stops at the last real event, so a quiet
caught-up stream would otherwise show phantom lag. A ticker drives the
gauge because the offset callback only fires while events flow and
freezes exactly when the workers are saturated. Until the first event
or heartbeat the gauge stays unset rather than reporting lag against a
zero or stale resume offset.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

* filer.sync: track replicated data sizes alongside event counts

An event count hides that 32 in-flight jobs can be 32 KB or 300 GB. Byte
counters mirror the event counters, and in_flight_bytes pairs with
in_flight_jobs. An event's size is the chunk delta - new chunks the old
entry does not already have - so deletes, renames, and attribute-only
updates count zero and byte rates reflect data movement, not metadata
churn.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

* grafana: chart the new filer.sync metrics

The lag panel reads lag_seconds directly instead of deriving it from
sync_offset, and the sync row gains event rate, throughput, and the
in-flight jobs and bytes gauges.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

* filer.sync: a pinned failure keeps showing as lag

An idle heartbeat means the stream is consumed, not that every event
replicated. While a permanent failure pins the watermark, letting the
heartbeat advance lag_seconds or the sync_offset gauge would report a
caught-up stream with an unreplicated event in it, so both now ignore
heartbeats until a restart replays the failure.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

* filer.sync: in-flight gauges survive subscription retries

A subscription retry builds a new processor sharing the gauge children
while the old processor's jobs may still be draining, so setting the
gauge from either side's local count clobbers the other. Each job now
increments and decrements for itself, keeping the total truthful across
generations.

Claude-Session: https://claude.ai/code/session_01VGUXS7kdqiaXenTYVCg6vw

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)
 		},
 	}