mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-18 21:26:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add Prometheus metrics for replication operations (#10006)

Browse Source 
* feat: add Prometheus metrics for replication operations

Adds 5 metrics to instrument volume server replication (write/delete):
- Operations counter with success/failure labels
- Duration histogram for latency tracking
- Targets gauge for replica fanout
- Failures counter with error reason labels
- Under-replicated volumes gauge on master

* fix: record replication duration histogram only when replicaCount > 0

* fix: update replication targets gauge for all operations including zero

* fix: ensure symmetric replication success/failure counting and proper metrics updates

* fix: change VolumeServerReplicationTargets from Gauge to Histogram

- Replace .Set() with .Observe() in store_replicate.go (2 occurrences)
- Update test to use CollectAndCount for histogram assertion
- Rename TestReplicationTargetsGauge -> TestReplicationTargetsHistogram
- Update documentation to reflect Histogram type and PromQL examples

* Add comments to replication metrics and improve test coverage

* metrics: add replication panels to grafana dashboard

Master row gets an under-replicated volumes timeseries; Volume Servers
row gets replication operations, failures-by-reason, p99 duration, and
average fan-out panels for the new replication metrics.

* metrics: name the replication duration histogram replication_seconds

Match the volumeServer convention (request_seconds, vacuuming_seconds)
rather than the admin/lifecycle _duration_seconds spelling.

* metrics: guard replication fan-out panel against divide-by-zero

clamp_min the _count rate so the avg-targets ratio reads 0 instead of
NaN when there are no replication events in the window.

---------

Co-authored-by: Ubuntu User <ubuntu@example.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
This commit is contained in:
Rushikesh Deshpande
2026-06-19 11:05:43 -07:00
committed by GitHub
co-authored by Ubuntu User Chris Lu
parent 20e4614fc6
commit df1a25fd3e
8 changed files with 793 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
other/metrics/grafana_seaweedfs.json
+480
View File
@@ -1822,6 +1822,102 @@
}
],
"pluginVersion": "10.3.1"
},
{
"datasource": {
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
"title": "Under-replicated Volumes",
"description": "Volumes that do not have the desired replica count, by collection. 0 = healthy.",
"type": "timeseries",
"id": 210,
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 53
},
"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": "sum by (collection) (SeaweedFS_master_under_replicated_volumes{cluster=~\"$cluster\"})",
"range": true,
"refId": "A",
"legendFormat": "{{collection}}"
}
],
"pluginVersion": "10.3.1"
}
]
},
@@ -3786,6 +3882,390 @@
}
],
"pluginVersion": "10.3.1"
},
{
"datasource": {
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
"title": "Replication Operations",
"description": "Rate of replica fan-out operations on volume servers by type and result",
"type": "timeseries",
"id": 211,
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 102
},
"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": "sum by (operation, result) (rate(SeaweedFS_volumeServer_replication_operations_total{cluster=~\"$cluster\"}[$__rate_interval]))",
"range": true,
"refId": "A",
"legendFormat": "{{operation}} {{result}}"
}
],
"pluginVersion": "10.3.1"
},
{
"datasource": {
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
"title": "Replication Failures by reason",
"description": "Rate of replication failures broken down by operation and classified reason",
"type": "timeseries",
"id": 212,
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 102
},
"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": "sum by (operation, reason) (rate(SeaweedFS_volumeServer_replication_failures_total{cluster=~\"$cluster\"}[$__rate_interval]))",
"range": true,
"refId": "A",
"legendFormat": "{{operation}} {{reason}}"
}
],
"pluginVersion": "10.3.1"
},
{
"datasource": {
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
"title": "Replication Duration p99",
"description": "p99 duration of replicated write/delete operations (includes local apply + replica fan-out)",
"type": "timeseries",
"id": 213,
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 110
},
"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": "s",
"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": "histogram_quantile(0.99, sum(rate(SeaweedFS_volumeServer_replication_seconds_bucket{cluster=~\"$cluster\"}[$__rate_interval])) by (le, operation))",
"range": true,
"refId": "A",
"legendFormat": "{{operation}}"
}
],
"pluginVersion": "10.3.1"
},
{
"datasource": {
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
"title": "Replication Fan-out (avg targets)",
"description": "Average number of replica targets per replicated operation",
"type": "timeseries",
"id": 214,
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 110
},
"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": "sum(rate(SeaweedFS_volumeServer_replication_targets_sum{cluster=~\"$cluster\"}[$__rate_interval])) / clamp_min(sum(rate(SeaweedFS_volumeServer_replication_targets_count{cluster=~\"$cluster\"}[$__rate_interval])), 1e-9)",
"range": true,
"refId": "A",
"legendFormat": "avg targets"
}
],
"pluginVersion": "10.3.1"
}
]
},
weed/server/master_grpc_server_volume.go
+2
View File
@@ -68,8 +68,10 @@ func (ms *MasterServer) ProcessGrowRequest() {
writable, crowded := vl.GetWritableVolumeCount()
mustGrow := int(lastGrowCount) - writable
vgr := vlc.ToVolumeGrowRequest()
underReplicated := vl.CountUnderReplicatedVolumes()
stats.MasterVolumeLayoutWritable.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(writable))
stats.MasterVolumeLayoutCrowded.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(crowded))
stats.MasterUnderReplicatedVolumes.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(underReplicated))
switch {
case mustGrow > 0:
weed/stats/metrics.go
+58
View File
@@ -118,6 +118,16 @@ var (
Help: "Number of crowded volumes in volume layouts",
}, []string{"collection", "disk", "rp", "ttl"})
// MasterUnderReplicatedVolumes tracks volumes that do not have enough replicas,
// partitioned by collection, disk type, replication type, and TTL.
MasterUnderReplicatedVolumes = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: Namespace,
Subsystem: "master",
Name: "under_replicated_volumes",
Help: "Current number of volumes that do not have enough replicas per collection/layout. 0 = healthy.",
}, []string{"collection", "disk", "rp", "ttl"})
MasterPickForWriteErrorCounter = prometheus.NewCounter(
prometheus.CounterOpts{
Namespace: Namespace,
@@ -436,6 +446,48 @@ var (
Help: "Counter of overall EC shards with issues detected during scrubbing.",
}, []string{"mode"})
// VolumeServerReplicationCounter counts replication operations by operation type
// (write, delete) and result (success, failure).
VolumeServerReplicationCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "volumeServer",
Name: "replication_operations_total",
Help: "Counter of replication operations by type (write, delete) and result (success, failure).",
}, []string{"operation", "result"})
// VolumeServerReplicationHistogram records replication operation duration in seconds,
// partitioned by operation type (write, delete).
VolumeServerReplicationHistogram = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "volumeServer",
Name: "replication_seconds",
Help: "Bucketed histogram of replication operation duration in seconds.",
Buckets: prometheus.ExponentialBuckets(0.0001, 2, 24),
}, []string{"operation"})
// VolumeServerReplicationTargets records the number of replica targets per replication
// operation, useful for observing fan-out width.
VolumeServerReplicationTargets = prometheus.NewHistogram(
prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "volumeServer",
Name: "replication_targets",
Help: "Histogram of replica targets count per replication operation.",
Buckets: []float64{1, 2, 3, 4, 5},
})
// VolumeServerReplicationFailures counts replication failures by operation type
// and failure reason (timeout, connection_refused, context_cancelled, server_error).
VolumeServerReplicationFailures = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "volumeServer",
Name: "replication_failures_total",
Help: "Counter of replication failures by operation and reason (timeout, connection_refused, context_cancelled, server_error).",
}, []string{"operation", "reason"})
S3RequestCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: Namespace,
@@ -806,6 +858,11 @@ func init() {
Gather.MustRegister(VolumeServerScrubLastTimeSeconds)
Gather.MustRegister(VolumeServerScrubVolumeFailures)
Gather.MustRegister(VolumeServerScrubShardFailures)
Gather.MustRegister(VolumeServerReplicationCounter)
Gather.MustRegister(VolumeServerReplicationHistogram)
Gather.MustRegister(VolumeServerReplicationTargets)
Gather.MustRegister(VolumeServerReplicationFailures)
Gather.MustRegister(MasterUnderReplicatedVolumes)
Gather.MustRegister(S3RequestCounter)
Gather.MustRegister(S3HandlerCounter)
@@ -931,6 +988,7 @@ func DeleteCollectionMetrics(collection string) {
c := MasterReplicaPlacementMismatch.DeletePartialMatch(labels)
c += MasterVolumeLayoutWritable.DeletePartialMatch(labels)
c += MasterVolumeLayoutCrowded.DeletePartialMatch(labels)
c += MasterUnderReplicatedVolumes.DeletePartialMatch(labels)
c += VolumeServerDiskSizeGauge.DeletePartialMatch(labels)
c += VolumeServerVolumeGauge.DeletePartialMatch(labels)
c += VolumeServerReadOnlyVolumeGauge.DeletePartialMatch(labels)
weed/stats/metrics_names.go
+18
View File
@@ -63,4 +63,22 @@ const (
ErrorCompletedEtagInvalid = "errorCompletedEtagInvalid"
ErrorCompletedEtagMismatch = "errorCompletedEtagMismatch"
ErrorCompletedPartEntryMismatch = "errorCompletedPartEntryMismatch"
// ReplicationOpWrite is the label value for write replication operations.
ReplicationOpWrite = "write"
// ReplicationOpDelete is the label value for delete replication operations.
ReplicationOpDelete = "delete"
// ReplicationSuccess is the label value for successful replication operations.
ReplicationSuccess = "success"
// ReplicationFailure is the label value for failed replication operations.
ReplicationFailure = "failure"
// FailureTimeout is the failure reason label for deadline-exceeded errors.
FailureTimeout = "timeout"
// FailureConnectionRefused is the failure reason label for connection-refused errors.
FailureConnectionRefused = "connection_refused"
// FailureContextCancelled is the failure reason label for context-cancelled errors.
FailureContextCancelled = "context_cancelled"
// FailureServerError is the failure reason label for generic server-side errors.
FailureServerError = "server_error"
)
weed/stats/metrics_replication_test.go
+123
View File
@@ -0,0 +1,123 @@
package stats
import (
"testing"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/testutil"
)
func TestReplicationMetricsRegistered(t *testing.T) {
VolumeServerReplicationCounter.Reset()
VolumeServerReplicationHistogram.Reset()
VolumeServerReplicationFailures.Reset()
MasterUnderReplicatedVolumes.Reset()
t.Cleanup(func() {
VolumeServerReplicationCounter.Reset()
VolumeServerReplicationHistogram.Reset()
VolumeServerReplicationFailures.Reset()
MasterUnderReplicatedVolumes.Reset()
})
// Seed CounterVec and HistogramVec with a value so collection returns them
VolumeServerReplicationCounter.WithLabelValues(ReplicationOpWrite, ReplicationSuccess).Inc()
VolumeServerReplicationHistogram.WithLabelValues(ReplicationOpWrite).Observe(0.1)
VolumeServerReplicationFailures.WithLabelValues(ReplicationOpWrite, FailureTimeout).Inc()
VolumeServerReplicationTargets.Observe(1)
MasterUnderReplicatedVolumes.WithLabelValues("default", "ssd", "1", "").Set(1)
metrics := []struct {
name string
c prometheus.Collector
}{
{"VolumeServerReplicationCounter", VolumeServerReplicationCounter},
{"VolumeServerReplicationHistogram", VolumeServerReplicationHistogram},
{"VolumeServerReplicationTargets", VolumeServerReplicationTargets},
{"VolumeServerReplicationFailures", VolumeServerReplicationFailures},
{"MasterUnderReplicatedVolumes", MasterUnderReplicatedVolumes},
}
for _, m := range metrics {
count := testutil.CollectAndCount(m.c)
if count < 1 {
t.Errorf("%s: expected at least 1 collection, got %d", m.name, count)
}
}
}
func TestReplicationCounterIncrement(t *testing.T) {
VolumeServerReplicationCounter.Reset()
VolumeServerReplicationCounter.WithLabelValues(ReplicationOpWrite, ReplicationSuccess).Inc()
VolumeServerReplicationCounter.WithLabelValues(ReplicationOpWrite, ReplicationFailure).Inc()
VolumeServerReplicationCounter.WithLabelValues(ReplicationOpDelete, ReplicationSuccess).Inc()
got := testutil.ToFloat64(VolumeServerReplicationCounter.WithLabelValues(ReplicationOpWrite, ReplicationSuccess))
if got != 1 {
t.Errorf("expected 1.0, got %f", got)
}
got = testutil.ToFloat64(VolumeServerReplicationCounter.WithLabelValues(ReplicationOpDelete, ReplicationSuccess))
if got != 1 {
t.Errorf("expected 1.0, got %f", got)
}
got = testutil.ToFloat64(VolumeServerReplicationCounter.WithLabelValues(ReplicationOpWrite, ReplicationFailure))
if got != 1 {
t.Errorf("expected 1.0, got %f", got)
}
}
func TestReplicationTargetsHistogram(t *testing.T) {
// VolumeServerReplicationTargets is a plain prometheus.Histogram
// (not a HistogramVec) and does not expose a Reset() method, so the
// histogram state is cumulative across tests that share the Gather
// registry. We check >= to tolerate prior observations.
VolumeServerReplicationTargets.Observe(3)
metrics, err := Gather.Gather()
if err != nil {
t.Fatalf("Failed to gather metrics: %v", err)
}
found := false
for _, mf := range metrics {
if mf.GetName() == "SeaweedFS_volumeServer_replication_targets" {
found = true
h := mf.GetMetric()[0].GetHistogram()
if h.GetSampleCount() < 1 {
t.Errorf("expected histogram_count>=1, got %d", h.GetSampleCount())
}
if h.GetSampleSum() < 3 {
t.Errorf("expected histogram_sum>=3, got %f", h.GetSampleSum())
}
}
}
if !found {
t.Error("SeaweedFS_volumeServer_replication_targets not found in gathered metrics")
}
}
func TestUnderReplicatedVolumesGauge(t *testing.T) {
MasterUnderReplicatedVolumes.WithLabelValues("default", "ssd", "1", "").Set(5)
got := testutil.ToFloat64(MasterUnderReplicatedVolumes.WithLabelValues("default", "ssd", "1", ""))
if got != 5 {
t.Errorf("expected 5.0, got %f", got)
}
}
func TestReplicationFailuresCounter(t *testing.T) {
VolumeServerReplicationFailures.Reset()
VolumeServerReplicationFailures.WithLabelValues(ReplicationOpWrite, FailureTimeout).Inc()
VolumeServerReplicationFailures.WithLabelValues(ReplicationOpDelete, FailureConnectionRefused).Inc()
got := testutil.ToFloat64(VolumeServerReplicationFailures.WithLabelValues(ReplicationOpWrite, FailureTimeout))
if got != 1 {
t.Errorf("expected 1.0, got %f", got)
}
got = testutil.ToFloat64(VolumeServerReplicationFailures.WithLabelValues(ReplicationOpDelete, FailureConnectionRefused))
if got != 1 {
t.Errorf("expected 1.0, got %f", got)
}
}
weed/topology/replication_metrics_test.go
+31
View File
@@ -0,0 +1,31 @@
package topology
import (
"context"
"errors"
"testing"
"github.com/seaweedfs/seaweedfs/weed/stats"
)
func TestClassifyReplicationError(t *testing.T) {
tests := []struct {
name string
err error
want string
}{
{"deadline exceeded", context.DeadlineExceeded, stats.FailureTimeout},
{"context cancelled", context.Canceled, stats.FailureContextCancelled},
{"connection refused", errors.New("connection refused"), stats.FailureConnectionRefused},
{"generic error", errors.New("internal server error"), stats.FailureServerError},
{"nil error", nil, ""},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := classifyReplicationError(tt.err)
if got != tt.want {
t.Errorf("classifyReplicationError() = %q, want %q", got, tt.want)
}
})
}
}
weed/topology/store_replicate.go
+66 -2
View File
@@ -24,6 +24,11 @@ import (
"google.golang.org/grpc"
)
// ReplicatedWrite writes a needle to the local volume and fans it out to all
// remote replica locations. When type=replicate is set, the request is itself
// a forwarded replication and no further remote lookups are performed.
// Returns isUnchanged=true when the local write determined the needle content
// was already present.
func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, s *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request, contentMd5 string) (isUnchanged bool, err error) {
//check JWT
@@ -46,6 +51,15 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
fsync = true
}
replicaCount := len(remoteLocations)
if replicaCount > 0 {
// Record replication duration histogram for the overall write operation
defer func(t time.Time) {
stats.VolumeServerReplicationHistogram.WithLabelValues(stats.ReplicationOpWrite).Observe(time.Since(t).Seconds())
}(time.Now())
}
if s.GetVolume(volumeId) != nil {
start := time.Now()
@@ -63,7 +77,10 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
}
}
if len(remoteLocations) > 0 { //send to other replica locations
// Observe replication targets histogram for all operations (including zero)
stats.VolumeServerReplicationTargets.Observe(float64(replicaCount))
if replicaCount > 0 { //send to other replica locations
start := time.Now()
inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas)
@@ -133,14 +150,22 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
if err != nil {
stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc()
reason := classifyReplicationError(err)
stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc()
err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
glog.V(0).Infoln(err)
return false, err
}
stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc()
}
return
}
// ReplicatedDelete deletes a needle from the local volume and sends delete
// requests to all remote replica locations. Replica deletes use
// context.Background() so that a client disconnect does not orphan replica
// deletes.
func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, store *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request) (size types.Size, err error) {
//check JWT
@@ -155,17 +180,36 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp
}
}
replicaCount := len(remoteLocations)
if replicaCount > 0 {
// Record replication duration and operation counter for delete
defer func(t time.Time) {
stats.VolumeServerReplicationHistogram.WithLabelValues(stats.ReplicationOpDelete).Observe(time.Since(t).Seconds())
if err != nil {
stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpDelete, stats.ReplicationFailure).Inc()
} else {
stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpDelete, stats.ReplicationSuccess).Inc()
}
}(time.Now())
}
size, err = store.DeleteVolumeNeedle(volumeId, n)
if err != nil {
glog.V(0).Infoln("delete error:", err)
return
}
if len(remoteLocations) > 0 { //send to other replica locations
// Observe replication targets histogram for all operations (including zero)
stats.VolumeServerReplicationTargets.Observe(float64(replicaCount))
if replicaCount > 0 { //send to other replica locations
// background, not r.Context(): a client disconnect must not orphan replica deletes
if err = DistributedOperation(context.Background(), remoteLocations, func(ctx context.Context, location operation.Location) error {
return util_http.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", string(jwt))
}); err != nil {
reason := classifyReplicationError(err)
stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpDelete, reason).Inc()
size = 0
}
}
@@ -255,3 +299,23 @@ func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOpt
return
}
// classifyReplicationError maps a Go error to a bounded-cardinality failure
// reason label for the replication_failures_total metric. Returns an empty
// string for nil errors.
func classifyReplicationError(err error) string {
if err == nil {
return ""
}
if errors.Is(err, context.DeadlineExceeded) {
return stats.FailureTimeout
}
if errors.Is(err, context.Canceled) {
return stats.FailureContextCancelled
}
errStr := err.Error()
if strings.Contains(errStr, "connection refused") {
return stats.FailureConnectionRefused
}
return stats.FailureServerError
}
weed/topology/volume_layout.go
+15
View File
@@ -774,6 +774,21 @@ func (vl *VolumeLayout) CloneWritableVolumes() (writables []needle.VolumeId) {
return writables
}
// CountUnderReplicatedVolumes returns the number of volumes in this layout
// that do not have enough replicas according to their replica placement
// configuration. Safe for concurrent access (RLock).
func (vl *VolumeLayout) CountUnderReplicatedVolumes() int {
vl.accessLock.RLock()
defer vl.accessLock.RUnlock()
count := 0
for vid := range vl.vid2location {
if !vl.enoughCopies(vid) {
count++
}
}
return count
}
func (vl *VolumeLayout) removeFromWritable(vid needle.VolumeId) bool {
toDeleteIndex := -1
for k, id := range vl.writables {