diff --git a/other/metrics/grafana_seaweedfs.json b/other/metrics/grafana_seaweedfs.json index 7b15390a4..8288fea93 100644 --- a/other/metrics/grafana_seaweedfs.json +++ b/other/metrics/grafana_seaweedfs.json @@ -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" } ] }, diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index ca30287c6..07d05d7b9 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -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: diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 4c6451350..e4a8cf1be 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -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) diff --git a/weed/stats/metrics_names.go b/weed/stats/metrics_names.go index 6c2c50ff0..c871ef5d2 100644 --- a/weed/stats/metrics_names.go +++ b/weed/stats/metrics_names.go @@ -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" ) diff --git a/weed/stats/metrics_replication_test.go b/weed/stats/metrics_replication_test.go new file mode 100644 index 000000000..c2fc50f8e --- /dev/null +++ b/weed/stats/metrics_replication_test.go @@ -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) + } +} diff --git a/weed/topology/replication_metrics_test.go b/weed/topology/replication_metrics_test.go new file mode 100644 index 000000000..302b72ad3 --- /dev/null +++ b/weed/topology/replication_metrics_test.go @@ -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) + } + }) + } +} diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index 18d329ba5..59e8d5c9c 100644 --- a/weed/topology/store_replicate.go +++ b/weed/topology/store_replicate.go @@ -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 +} diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index cdb4c5814..8f389c6e6 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -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 {