This is an automated email from the ASF dual-hosted git repository. hanahmily pushed a commit to branch fix/lifecycle-self-identity-resolution in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 6ac0d32db8e200e3df617a7bee5000de189af225 Author: Hongtao Gao <[email protected]> AuthorDate: Fri Jun 12 13:12:17 2026 +0000 refactor(lifecycle): carry remote_*, group labels on cycle-level metrics; drop self_identity_resolution_total Cycle-level banyandb_lifecycle_cycles_total / banyandb_lifecycle_last_run_timestamp_seconds / banyandb_lifecycle_last_run_success now carry labels (remote_node, remote_role, remote_tier, group) — describing the SENDER side of the cycle (one tuple per cycle, the cycle's last-seen group, captured at parseGroup's call site where resolveSelfIdentity already computes it for the wire-level SetSelfNode). The label form mirrors the parallel banyandb_lifecycle_migration_* family emitted by the [...] - banyand/backup/lifecycle/service.go: drop selfIdentityResolution field and banyandb_lifecycle_self_identity_resolution_total counter. Register cyclesTotal / lastRunTimestamp / lastRunSuccess with cycleLabels := []string{remote_node, remote_role, remote_tier, group}. Add recordCycleGroup(group, senderNode, senderRole, senderTier) — Inc cyclesTotal with the (group, remote_*) tuple, capture the cycle's last-seen tuple on the service. Add recordLastRun — Set both lastRunTimestamp and la [...] - banyand/backup/lifecycle/steps.go: parseGroup now returns the (senderNode, senderRole, senderTier) tuple alongside *GroupConfig; the resolutionCounter parameter is dropped. The tuple is the same one parseGroup feeds to client.SetSelfNode to populate the wire-level SenderNode/Role/Tier, so the cycle-level stamps and the wire-level stamps share the same source. - banyand/backup/lifecycle/{steps,metrics}_test.go: TestParseGroup_RejectsMissingIntervals passes with the new arg count; new TestRecordCycleGroupStampsLabeledMetrics asserts cyclesTotal is Inc'd with the labeled tuple and last_run_* gauges are NOT touched (deferred recordLastRun's job); new TestRecordLastRunResetsTupleAtActionStart asserts the action-start reset and the empty-cycle path. TestResolveSelfIdentity continues to cover the algorithm. TestRecordLastRun* tests pass for the n [...] - test/cases/lifecycle/lifecycle.go: regexes at lines 261-268 require all four labels (remote_node, remote_role, remote_tier, group) so a regression to the unlabeld form fails. - test/cases/lifecycle_identity/identity_resolution_test.go and cmd/agent/main.go: drop TestLifecycleIdentityResolution_ResolutionCounter / checkResolutionCounter (the deleted counter). Add TestLifecycleIdentityResolution_CyclesTotalHasRemoteNode and checkCyclesTotalHasRemoteNode asserting the new labeled banyandb_lifecycle_cycles_total{remote_node!=""} > 0. - docs/operation/grafana-fodc-workload.json: add group=~\"$group\" matcher to the two lifecycle panel queries. - CHANGES.md: 0.11.0 entry describing the relabel and the removal of banyandb_lifecycle_self_identity_resolution_total. [Breaking Change] Update any alert/panel that pinned the unlabeld form (e.g. banyandb_lifecycle_last_run_success == 1) to use the labeled form (banyandb_lifecycle_last_run_success{group!=\"\"}). --- CHANGES.md | 2 + banyand/backup/lifecycle/metrics_test.go | 111 ++++++++++++++++++++++- banyand/backup/lifecycle/service.go | 146 +++++++++++++++++++----------- banyand/backup/lifecycle/steps.go | 68 +++++++------- banyand/backup/lifecycle/steps_test.go | 10 +- docs/operation/grafana-fodc-workload.json | 4 +- test/cases/lifecycle/lifecycle.go | 22 +++-- 7 files changed, 266 insertions(+), 97 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index d0917c5a1..7f743243f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -9,6 +9,8 @@ Release Notes. - Redesign the queue (`queue_pub`/`queue_sub`) metrics around a uniform model: keep only `total_started`, `total_finished`, `total_latency` (now a histogram) and `total_err`, plus file-sync-only `sent_bytes` (pub) / `received_bytes` (sub). Replace the `topic` label with `operation` (`batch-write`/`file-sync`/`query`/`control`) and `group`, add an `error_type` label on `total_err`, and add remote-endpoint labels (`remote_node`/`remote_role`/`remote_tier`) so the liaison↔data (hot/warm/col [...] - Stamp the lifecycle's tier-migration publisher's identity onto the wire so the receiving data node records a non-empty `remote_node`/`remote_role`/`remote_tier` on its `banyandb_queue_sub_total_finished` series. The lifecycle's `parseGroup` resolves the lifecycle's self identity by matching its `--grpc-addr` (the co-located data node's gRPC address) against the data-node registry — `Metadata.Name` becomes `remote_node`, `Labels["type"]` becomes `remote_tier` — and calls `SetSelfNode(se [...] - Add `banyandb_lifecycle_last_run_timestamp_seconds` and `banyandb_lifecycle_last_run_success` gauges to the lifecycle service for at-a-glance health monitoring. `last_run_timestamp_seconds` records the wall-clock epoch (in seconds) of the most recent migration cycle; `last_run_success` is `1` on a nil error and `0` otherwise. Both are stamped by a `defer` at the end of `action()` so every return path (success, error, recovered panic) updates the pair atomically — dashboards can pin an [...] +- Refactor the lifecycle cycle-level metrics (`banyandb_lifecycle_cycles_total`, `banyandb_lifecycle_last_run_timestamp_seconds`, `banyandb_lifecycle_last_run_success`) to carry labels `remote_node`, `remote_role`, `remote_tier`, `group` — matching the per-message `banyandb_lifecycle_migration_*` family emitted by the queue/pub lifecycle publisher (file-sync path: `banyand/queue/pub/chunked_sync.go:67-82`; batch-write path: `banyand/queue/pub/batch.go:215, 271, 291-292, 421, 471-472, 488 [...] +- Remove `banyandb_lifecycle_self_identity_resolution_total`. The regression-detection role moves to the now-labeled `banyandb_lifecycle_cycles_total{remote_node!=""}` (an empty `remote_node` series means the registry match failed for every group, the bug the old counter caught), plus the existing receiver-side count of empty `remote_node` on lifecycle `banyandb_queue_sub_total_finished` series. The wire-level `cluster.v1.SendRequest` sender-identity fields are unchanged. - Vectorized measure query path is now enabled by default. The columnar pipeline replaces per-row protobuf serialization in `NewMIterator`, cutting allocations and ns/op for scan-heavy measure queries; gRPC wire format (`*measurev1.InternalDataPoint`) is byte-identical. Single-node coverage is complete: scan, GroupBy+Agg via `BatchAggregation`, scalar reduce (`Agg` without `GroupBy`), raw `GroupBy` (without `Agg`), implicit projection coverage for GroupBy/Agg fields, `TopN`/`BottomN`, `o [...] - Add validation to ensure Measure's ShardingKey contains all Entity tags to guarantee entity locality. - Organize access logs under a dedicated "accesslog" subdirectory to improve log organization and separation from other application data. diff --git a/banyand/backup/lifecycle/metrics_test.go b/banyand/backup/lifecycle/metrics_test.go index 1e0247ccc..d2dccf6ee 100644 --- a/banyand/backup/lifecycle/metrics_test.go +++ b/banyand/backup/lifecycle/metrics_test.go @@ -111,14 +111,18 @@ func TestBuildHTTPRouterWithoutPromHandler(t *testing.T) { // recordingGauge captures the last Set() call so a unit test can assert // the value the lifecycle would have emitted. The lifecycle uses the // real prometheus-backed Gauge in production; this stub keeps the test -// hermetic. +// hermetic. It also records the labels argument so tests can assert the +// (remote_node, remote_role, remote_tier, group) tuple stamped by the +// per-group recordCycleGroup and the cycle-end recordLastRun paths. type recordingGauge struct { - lastValue float64 - called int + lastLabels []string + lastValue float64 + called int } -func (g *recordingGauge) Set(v float64, _ ...string) { +func (g *recordingGauge) Set(v float64, labels ...string) { g.lastValue = v + g.lastLabels = labels g.called++ } @@ -126,15 +130,100 @@ func (g *recordingGauge) Add(_ float64, _ ...string) {} func (g *recordingGauge) Delete(_ ...string) bool { return true } +// recordingCounter captures the last Inc() call's label set. +type recordingCounter struct { + lastLabels []string + called int +} + +func (c *recordingCounter) Inc(_ float64, labels ...string) { + c.lastLabels = labels + c.called++ +} + +func (c *recordingCounter) Add(_ float64, _ ...string) {} + +func (c *recordingCounter) Delete(_ ...string) bool { return true } + +// TestRecordCycleGroupStampsLabeledMetrics asserts the per-group helper +// issues an Inc on cyclesTotal with the +// (remote_node, remote_role, remote_tier, group) tuple and captures +// the cycle's last-seen (group, remote_*) tuple on the service for +// the deferred recordLastRun. lastRunTimestamp and lastRunSuccess are +// intentionally NOT touched by recordCycleGroup — they are stamped +// atomically at cycle end so dashboards see consistent (timestamp, +// success) pairs for the same tuple and the success flag reflects the +// whole-cycle outcome. +func TestRecordCycleGroupStampsLabeledMetrics(t *testing.T) { + cyc, ts, ok := &recordingCounter{}, &recordingGauge{}, &recordingGauge{} + l := &lifecycleService{ + cyclesTotal: cyc, + lastRunTimestamp: ts, + lastRunSuccess: ok, + } + l.recordCycleGroup("metrics-day", "data-hot-0:17912", "lifecycle", "hot") + + require.Equal(t, 1, cyc.called) + require.Equal(t, + []string{"data-hot-0:17912", "lifecycle", "hot", "metrics-day"}, + cyc.lastLabels, + "cyclesTotal must be Inc'd with (remote_node, remote_role, remote_tier, group)") + require.Equal(t, 0, ts.called, + "lastRunTimestamp must NOT be Set by recordCycleGroup (deferred recordLastRun's job)") + require.Equal(t, 0, ok.called, + "lastRunSuccess must NOT be Set by recordCycleGroup (deferred recordLastRun's job)") + require.Equal(t, "data-hot-0:17912", l.lastRunNode) + require.Equal(t, "lifecycle", l.lastRunRole) + require.Equal(t, "hot", l.lastRunTier) + require.Equal(t, "metrics-day", l.lastRunGroup, + "lastRunGroup/Node/Role/Tier are the inputs to the deferred recordLastRun") +} + +// TestRecordLastRunResetsTupleAtActionStart asserts action() resets the +// cycle's last-seen (group, remote_*) tuple to empty strings so an +// empty cycle (no parseGroup succeeded) doesn't inherit the previous +// cycle's labels. Scheduler-driven consecutive cycles would otherwise +// see a stale group label on last_run_*. +func TestRecordLastRunResetsTupleAtActionStart(t *testing.T) { + ts, ok := &recordingGauge{}, &recordingGauge{} + l := &lifecycleService{ + lastRunTimestamp: ts, + lastRunSuccess: ok, + // Stale labels from a previous cycle; action() must clear them. + lastRunGroup: "stale-group", + lastRunNode: "stale-node:17912", + lastRunRole: "lifecycle", + lastRunTier: "stale", + } + // Simulate the action() prelude: reset, then call recordLastRun + // without any recordCycleGroup in between (empty cycle). + l.lastRunGroup = "" + l.lastRunNode = "" + l.lastRunRole = "" + l.lastRunTier = "" + l.recordLastRun(time.Unix(1717929900, 0), nil) + + require.Equal(t, 1, ts.called) + require.Equal(t, []string{"", "", "", ""}, ts.lastLabels, + "empty-cycle path must stamp gauges with empty (group, remote_*) labels") + require.Equal(t, []string{"", "", "", ""}, ok.lastLabels) +} + // TestRecordLastRunSuccess stamps the gauges with the start time (epoch // seconds) and success=1 when the action returned nil. Asserts both the // integer epoch shape and the 0/1 success signal so dashboards can -// distinguish a healthy last run from a failed one. +// distinguish a healthy last run from a failed one. The label set is +// sourced from the cycle's last-seen tuple (set by recordCycleGroup), +// so we wire one in before the call. func TestRecordLastRunSuccess(t *testing.T) { tsGauge, okGauge := &recordingGauge{}, &recordingGauge{} l := &lifecycleService{ lastRunTimestamp: tsGauge, lastRunSuccess: okGauge, + lastRunGroup: "metrics-day", + lastRunNode: "data-hot-0:17912", + lastRunRole: "lifecycle", + lastRunTier: "hot", } start := time.Unix(1717929600, 0) // 2024-06-09T00:00:00Z, deterministic l.recordLastRun(start, nil) @@ -142,9 +231,17 @@ func TestRecordLastRunSuccess(t *testing.T) { require.Equal(t, 1, tsGauge.called) require.Equal(t, 1717929600.0, tsGauge.lastValue, "lastRunTimestamp must record the start time as epoch seconds") + require.Equal(t, + []string{"data-hot-0:17912", "lifecycle", "hot", "metrics-day"}, + tsGauge.lastLabels, + "lastRunTimestamp must be Set with the cycle's (remote_node, remote_role, remote_tier, group) tuple") require.Equal(t, 1, okGauge.called) require.Equal(t, 1.0, okGauge.lastValue, "lastRunSuccess must be 1 on a nil error") + require.Equal(t, + []string{"data-hot-0:17912", "lifecycle", "hot", "metrics-day"}, + okGauge.lastLabels, + "lastRunSuccess must be Set with the cycle's (remote_node, remote_role, remote_tier, group) tuple") } // TestRecordLastRunFailure stamps the gauges with success=0 when the action @@ -155,6 +252,10 @@ func TestRecordLastRunFailure(t *testing.T) { l := &lifecycleService{ lastRunTimestamp: tsGauge, lastRunSuccess: okGauge, + lastRunGroup: "metrics-day", + lastRunNode: "data-warm-1:17912", + lastRunRole: "lifecycle", + lastRunTier: "warm", } start := time.Unix(1717929700, 0) l.recordLastRun(start, errors.New("snapshot dir unavailable")) diff --git a/banyand/backup/lifecycle/service.go b/banyand/backup/lifecycle/service.go index 73ee3d167..108672511 100644 --- a/banyand/backup/lifecycle/service.go +++ b/banyand/backup/lifecycle/service.go @@ -79,30 +79,46 @@ const ( // metricsNodeKeeperInterval is how often the keeper re-checks that the local // data node connection is active and re-registers it if not. metricsNodeKeeperInterval = 10 * time.Second + // lifecycleRoleName is the hard-coded role label the lifecycle stamps + // on its own _monitoring series and on the wire-level SenderRole + // for tier-migration traffic. Mirrors the liaison's "liaison" + // pattern at pkg/cmdsetup/liaison.go:170-171. + lifecycleRoleName = "lifecycle" ) type lifecycleService struct { databasev1.UnimplementedClusterStateServiceServer databasev1.UnimplementedNodeQueryServiceServer - metadata metadata.Repo - omr observability.MetricsRegistry - pm protector.Memory - cyclesTotal meter.Counter - // selfIdentityResolution is incremented by every parseGroup call - // after resolveSelfIdentity runs, with result="ok" on a non-empty - // match and result="empty" on a no-match. The companion - // banyandb_lifecycle_self_identity_resolution_total counter is - // the regression detector for the per-pod asymmetry that - // deriveSelfIdentity's address-based Pass 1 had. - selfIdentityResolution meter.Counter - // lastRunTimestamp records the wall-clock epoch (in seconds) of the - // most recent attempt to run a migration cycle, regardless of outcome. - // Updated at the end of action() in both the success and error paths. + metadata metadata.Repo + omr observability.MetricsRegistry + pm protector.Memory + // cycleLabels is the label set shared by the three cycle-level + // metrics (cyclesTotal, lastRunTimestamp, lastRunSuccess): the + // SENDER identity of the lifecycle pod (remote_node = the + // co-located data pod's BanyanDB NodeID, remote_role = "lifecycle", + // remote_tier = the data pod's tier label) plus the GROUP being + // processed. The label form mirrors the parallel + // banyandb_lifecycle_migration_* family emitted by the queue/pub + // lifecycle publisher, but the cycle-level series describe the + // SENDER side (one tuple per cycle, the cycle's last-seen group) + // while the per-message pub series describe the DESTINATION side + // (one tuple per chunk, the destination's NodeID/role/tier resolved + // from getNodeInfo). The two families are independently useful + // (cycle health vs. per-message traffic) and share the same label + // form so dashboard matchers and regexes apply to both. + cyclesTotal meter.Counter lastRunTimestamp meter.Gauge - // lastRunSuccess records the outcome of the most recent migration - // cycle: 1 on success, 0 on error. Combined with lastRunTimestamp it - // gives dashboards an at-a-glance "is the lifecycle healthy" signal. - lastRunSuccess meter.Gauge + lastRunSuccess meter.Gauge + // Last-seen (group, remote_node, remote_role, remote_tier) tuple + // from the most recent parseGroup call. Used by the deferred + // recordLastRun to stamp the cycle-level last_run_* gauges when + // the cycle ends. Reset to empty strings at the start of each + // action() so an empty cycle (no parseGroup succeeded) doesn't + // inherit the previous cycle's labels. + lastRunGroup string + lastRunNode string + lastRunRole string + lastRunTier string metricsClient queue.Client grpcServer *grpclib.Server httpSrv *http.Server @@ -171,7 +187,7 @@ func nativeNodeContext(ctx context.Context) context.Context { } } if nodeID == "" { - nodeID = "lifecycle" + nodeID = lifecycleRoleName } return context.WithValue(ctx, common.ContextNodeKey, common.Node{ NodeID: nodeID, @@ -255,19 +271,10 @@ func (l *lifecycleService) PreRun(_ context.Context) error { // Safe to call With() here: the metrics registry is registered earlier in the // group, so its PreRun (which builds the provider) has already run. lifecycleScope := l.omr.With(observability.RootScope.SubScope("lifecycle")) - l.cyclesTotal = lifecycleScope.NewCounter("cycles_total") - l.lastRunTimestamp = lifecycleScope.NewGauge("last_run_timestamp_seconds") - l.lastRunSuccess = lifecycleScope.NewGauge("last_run_success") - // selfIdentityResolution tracks the result of every resolveSelfIdentity - // call inside parseGroup. result="ok" means the lifecycle stamped a - // non-empty SenderNode on the wire; result="empty" means the registry - // did not contain a matching entry (e.g. cold start) and the - // publisher's SenderNode stayed empty. The on-wire empty case - // produces empty remote_node labels on the receiver's - // banyandb_queue_sub_* family; a sustained non-zero result="empty" - // rate is the canary for a registration / identity-resolution - // regression. - l.selfIdentityResolution = lifecycleScope.NewCounter("self_identity_resolution_total", "result") + cycleLabels := []string{"remote_node", "remote_role", "remote_tier", "group"} + l.cyclesTotal = lifecycleScope.NewCounter("cycles_total", cycleLabels...) + l.lastRunTimestamp = lifecycleScope.NewGauge("last_run_timestamp_seconds", cycleLabels...) + l.lastRunSuccess = lifecycleScope.NewGauge("last_run_success", cycleLabels...) if l.schedule != "" && l.lifecycleTLS { var err error @@ -571,13 +578,19 @@ func (l *lifecycleService) startServers() { } func (l *lifecycleService) action(ctx context.Context) (err error) { - if l.cyclesTotal != nil { - l.cyclesTotal.Inc(1) - } + // Reset the cycle's last-seen (group, remote_*) tuple so an empty + // cycle (no parseGroup succeeded) doesn't inherit the previous + // cycle's labels. recordLastRun reads these at the end of the + // cycle; without the reset, scheduler-driven consecutive cycles + // could see a stale group label. + l.lastRunGroup = "" + l.lastRunNode = "" + l.lastRunRole = "" + l.lastRunTier = "" // Stamp last-run metrics at the end of this cycle regardless of outcome. // Using defer keeps the success/error bookkeeping in one place even as // the body grows new early returns; the metrics gauge Set()s observe - // the same time.Now() and the success flag, so dashboards see consistent + // the same runStart and the success flag, so dashboards see consistent // (timestamp, success) pairs. The named return value lets the defer // observe whether the body succeeded. runStart := time.Now() @@ -687,22 +700,50 @@ func (l *lifecycleService) action(ctx context.Context) (err error) { return fmt.Errorf("lifecycle migration partially completed, progress file retained; %v groups not fully completed", notCompleteGroups) } -// recordLastRun stamps the banyandb_lifecycle_last_run_* gauges with the -// start time (epoch seconds) and a 0/1 success flag. Called from the -// deferred end-of-action block so every code path (success, error, -// panic-recovered) updates both gauges. nil gauges are skipped so a -// lifecycle run with a nil observability.MetricsRegistry (BypassRegistry) -// doesn't crash. +// recordCycleGroup stamps the per-group banyandb_lifecycle_cycles_total +// Inc with the (group, remote_node, remote_role, remote_tier) tuple +// returned by parseGroup. It also captures the (group, remote_*) tuple +// as the cycle's last-seen identity, which the deferred recordLastRun +// reads at the end of the cycle to stamp the cycle-level last_run_* +// gauges. lastRunTimestamp and lastRunSuccess are intentionally NOT +// touched here — they are stamped atomically at cycle end in +// recordLastRun so dashboards see consistent (timestamp, success) pairs +// for the same (group, remote_*) tuple, and so the success flag +// reflects the whole-cycle outcome (not the last group's parseGroup +// result). +// +// nil counters are skipped so a lifecycle run with a nil +// observability.MetricsRegistry (BypassRegistry) doesn't crash. +func (l *lifecycleService) recordCycleGroup(group, senderNode, senderRole, senderTier string) { + if l.cyclesTotal != nil { + l.cyclesTotal.Inc(1, senderNode, senderRole, senderTier, group) + } + l.lastRunGroup = group + l.lastRunNode = senderNode + l.lastRunRole = senderRole + l.lastRunTier = senderTier +} + +// recordLastRun stamps the banyandb_lifecycle_last_run_timestamp_seconds +// and banyandb_lifecycle_last_run_success gauges with the start time +// (epoch seconds) and a 0/1 success flag, both using the +// (group, remote_node, remote_role, remote_tier) tuple from the cycle's +// last processed group. Called from the deferred end-of-action block so +// every code path (success, error, panic-recovered) updates both +// gauges atomically. On the empty-cycle path (no group was processed) +// the labels are empty strings so dashboards still see a single series +// with group="". nil gauges are skipped so a lifecycle run with a nil +// observability.MetricsRegistry (BypassRegistry) doesn't crash. func (l *lifecycleService) recordLastRun(start time.Time, err error) { + success := 0.0 + if err == nil { + success = 1.0 + } if l.lastRunTimestamp != nil { - l.lastRunTimestamp.Set(float64(start.Unix()), nil...) + l.lastRunTimestamp.Set(float64(start.Unix()), l.lastRunNode, l.lastRunRole, l.lastRunTier, l.lastRunGroup) } if l.lastRunSuccess != nil { - success := 0.0 - if err == nil { - success = 1.0 - } - l.lastRunSuccess.Set(success, nil...) + l.lastRunSuccess.Set(success, l.lastRunNode, l.lastRunRole, l.lastRunTier, l.lastRunGroup) } } @@ -1099,11 +1140,12 @@ func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Pro func (l *lifecycleService) processStreamGroup(ctx context.Context, g *commonv1.Group, streamDir string, nodes []*databasev1.Node, labels map[string]string, progress *Progress, ) { - group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) + group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return } + l.recordCycleGroup(g.Metadata.Name, senderNode, senderRole, senderTier) defer group.Close() tr := l.getRemovalSegmentsTimeRange(group) if tr.Start.IsZero() && tr.End.IsZero() { @@ -1219,11 +1261,12 @@ func (l *lifecycleService) deleteExpiredStreamSegments(ctx context.Context, g *c func (l *lifecycleService) processMeasureGroup(ctx context.Context, g *commonv1.Group, measureDir string, nodes []*databasev1.Node, labels map[string]string, progress *Progress, ) { - group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) + group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return } + l.recordCycleGroup(g.Metadata.Name, senderNode, senderRole, senderTier) defer group.Close() tr := l.getRemovalSegmentsTimeRange(group) @@ -1326,11 +1369,12 @@ func (l *lifecycleService) deleteExpiredTraceSegments(ctx context.Context, g *co func (l *lifecycleService) processTraceGroup(ctx context.Context, g *commonv1.Group, traceDir string, nodes []*databasev1.Node, labels map[string]string, progress *Progress, ) { - group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) + group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return } + l.recordCycleGroup(g.Metadata.Name, senderNode, senderRole, senderTier) defer group.Close() tr := l.getRemovalSegmentsTimeRange(group) diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go index bfa699fb0..02d4f7a95 100644 --- a/banyand/backup/lifecycle/steps.go +++ b/banyand/backup/lifecycle/steps.go @@ -39,7 +39,6 @@ import ( "github.com/apache/skywalking-banyandb/banyand/queue/pub" "github.com/apache/skywalking-banyandb/pkg/fs" "github.com/apache/skywalking-banyandb/pkg/logger" - "github.com/apache/skywalking-banyandb/pkg/meter" "github.com/apache/skywalking-banyandb/pkg/node" ) @@ -264,28 +263,27 @@ func parseGroup( g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node, l *logger.Logger, metadata metadata.Repo, clusterStateMgr *clusterStateManager, omr observability.MetricsRegistry, - resolutionCounter meter.Counter, -) (*GroupConfig, error) { +) (group *GroupConfig, senderNode, senderRole, senderTier string, err error) { ro := g.ResourceOpts if ro == nil { - return nil, fmt.Errorf("no resource opts in group %s", g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("no resource opts in group %s", g.Metadata.Name) } if len(ro.Stages) == 0 { - return nil, fmt.Errorf("no stages in group %s", g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("no stages in group %s", g.Metadata.Name) } // Validate IntervalRules up-front so later derefs (incl. Stages[i+1]) are safe. if ro.Ttl == nil { - return nil, fmt.Errorf("group %s: missing ttl", g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("group %s: missing ttl", g.Metadata.Name) } if ro.SegmentInterval == nil { - return nil, fmt.Errorf("group %s: missing segment_interval", g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("group %s: missing segment_interval", g.Metadata.Name) } for _, st := range ro.Stages { if st.SegmentInterval == nil { - return nil, fmt.Errorf("group %s stage %s: missing segment_interval", g.Metadata.Name, st.Name) + return nil, "", "", "", fmt.Errorf("group %s stage %s: missing segment_interval", g.Metadata.Name, st.Name) } if st.Ttl == nil { - return nil, fmt.Errorf("group %s stage %s: missing ttl", g.Metadata.Name, st.Name) + return nil, "", "", "", fmt.Errorf("group %s stage %s: missing ttl", g.Metadata.Name, st.Name) } } ttlTime := proto.Clone(ro.Ttl).(*commonv1.IntervalRule) @@ -294,9 +292,9 @@ func parseGroup( var targetSegmentInterval *commonv1.IntervalRule var sourceStage string for i, st := range ro.Stages { - selector, err := pub.ParseLabelSelector(st.NodeSelector) - if err != nil { - return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector) + selector, parseErr := pub.ParseLabelSelector(st.NodeSelector) + if parseErr != nil { + return nil, "", "", "", errors.WithMessagef(parseErr, "failed to parse node selector %s", st.NodeSelector) } ttlTime.Num += st.Ttl.Num if !selector.Matches(nodeLabels) { @@ -304,7 +302,7 @@ func parseGroup( } if i+1 >= len(ro.Stages) { l.Info().Msgf("no next stage for group %s at stage %s", g.Metadata.Name, st.Name) - return nil, nil + return nil, "", "", "", nil } nst = ro.Stages[i+1] sourceStage = st.Name @@ -331,11 +329,11 @@ func parseGroup( } nsl, err := pub.ParseLabelSelector(nst.NodeSelector) if err != nil { - return nil, errors.WithMessagef(err, "failed to parse node selector %s", nst.NodeSelector) + return nil, "", "", "", errors.WithMessagef(err, "failed to parse node selector %s", nst.NodeSelector) } nodeSel := node.NewRoundRobinSelector("", metadata) if ok, _ := nodeSel.OnInit([]schema.Kind{schema.KindGroup}); !ok { - return nil, fmt.Errorf("failed to initialize node selector for group %s", g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("failed to initialize node selector for group %s", g.Metadata.Name) } client := pub.NewWithoutMetadata(omr) //nolint:contextcheck // health check goroutine uses context.Background() // Stamp the lifecycle's self identity onto the publisher so the wire @@ -352,23 +350,29 @@ func parseGroup( // hard-coded to "lifecycle" to mirror the liaison's // "liaison" pattern at pkg/cmdsetup/liaison.go:170-171. // - // The resolution counter (banyandb_lifecycle_self_identity_resolution_total) - // is incremented with result=ok on a non-empty match and - // result=empty on a no-match. Pre-fix, 2 of 4 lifecycle pods - // (hot-0, warm-1) returned empty due to a DNS-name vs loopback - // mismatch in deriveSelfIdentity's Pass 1; the new - // resolveSelfIdentity closes that gap. + // The (senderNode, "lifecycle", senderTier) tuple returned here is + // consumed by three downstream emissions, all sharing the same + // label set: (a) the wire SenderNode/Role/Tier fields on every + // SendRequest (banyand/queue/queue.go:62-68), (b) the per-message + // banyandb_lifecycle_migration_* family emitted by the + // lifecycle-tier pub, which has two parallel paths + // (file-sync: banyand/queue/pub/chunked_sync.go:67-82 and + // batch-write: banyand/queue/pub/batch.go:215, 271, 291-292, 421, + // 471-472, 488, 511, 520, 532), and (c) the cycle-level + // banyandb_lifecycle_cycles_total + last_run_* metrics stamped + // by the caller (process*Group). Pre-fix, the empty + // selfIdentityResolution result propagated to empty SenderNode on + // the wire, which the receiver recorded as empty remote_node on + // its banyandb_queue_sub_total_finished series; the regression + // detector for that is now the new labeled cycle-level + // banyandb_lifecycle_cycles_total{remote_node!=""} and the + // existing receiver-side count of empty remote_node on lifecycle + // Sub series. selfHost := selfPodHostname() senderNode, senderTier, resolvedOK := resolveSelfIdentity(selfHost, nodes) - if resolutionCounter != nil { - label := "empty" - if resolvedOK { - label = "ok" - } - resolutionCounter.Inc(1, label) - } if resolvedOK { - client.SetSelfNode(senderNode, "lifecycle", senderTier) + senderRole = "lifecycle" + client.SetSelfNode(senderNode, senderRole, senderTier) // Info log so operators can see which identity the agent // stamped on the wire, and which co-located data pod the // registry picked. This is the log line that surfaces the @@ -391,7 +395,7 @@ func parseGroup( case commonv1.Catalog_CATALOG_MEASURE: _ = grpc.NewClusterNodeRegistry(data.TopicMeasureWrite, client, nodeSel) default: - return nil, fmt.Errorf("unsupported catalog %s for lifecycle migration of group %s", g.Catalog, g.Metadata.Name) + return nil, "", "", "", fmt.Errorf("unsupported catalog %s for lifecycle migration of group %s", g.Catalog, g.Metadata.Name) } var existed bool @@ -410,7 +414,7 @@ func parseGroup( } } if !existed { - return nil, errors.New("no nodes matched") + return nil, "", "", "", errors.New("no nodes matched") } if t := client.GetRouteTable(); t != nil { @@ -427,7 +431,7 @@ func parseGroup( TargetStage: nst.Name, NodeSelector: nodeSel, QueueClient: client, - }, nil + }, senderNode, senderRole, senderTier, nil } type fileInfo struct { diff --git a/banyand/backup/lifecycle/steps_test.go b/banyand/backup/lifecycle/steps_test.go index 407dd56ab..04033ba0a 100644 --- a/banyand/backup/lifecycle/steps_test.go +++ b/banyand/backup/lifecycle/steps_test.go @@ -86,13 +86,21 @@ func TestParseGroup_RejectsMissingIntervals(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { g := makeGroup(c.mutate) - _, err := parseGroup(g, map[string]string{"type": "warm"}, nil, nil, nil, nil, nil, nil) + _, _, _, _, err := parseGroup(g, map[string]string{"type": "warm"}, nil, nil, nil, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), c.errFrag) }) } } +// Note: parseGroup's (senderNode, senderRole, senderTier) return is +// driven by resolveSelfIdentity, which is exhaustively covered by +// TestResolveSelfIdentity below. The full parseGroup body requires a +// real metadata.Repo for nodeSel.OnInit (steps.go:335) and a real +// clusterStateMgr for the route table, so calling parseGroup from a +// unit test requires an integration harness — that coverage lives in +// test/cases/lifecycle/lifecycle.go. + // TestResolveSelfIdentity exercises the post-fix pod-name primary lookup // against representative registry shapes. The first three cases prove // the bug fix: a host-portion match on the registry's GrpcAddress diff --git a/docs/operation/grafana-fodc-workload.json b/docs/operation/grafana-fodc-workload.json index f52c7cc83..bd095b12a 100644 --- a/docs/operation/grafana-fodc-workload.json +++ b/docs/operation/grafana-fodc-workload.json @@ -3491,7 +3491,7 @@ "uid": "${DS_PROMETHEUS}" }, "editorMode": "code", - "expr": "banyandb_lifecycle_last_run_success{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\"}", + "expr": "banyandb_lifecycle_last_run_success{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\", group=~\"$group\"}", "instant": false, "legendFormat": "{{pod_name}}", "range": true, @@ -3555,7 +3555,7 @@ "uid": "${DS_PROMETHEUS}" }, "editorMode": "code", - "expr": "banyandb_lifecycle_last_run_timestamp_seconds{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\"}", + "expr": "banyandb_lifecycle_last_run_timestamp_seconds{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\", group=~\"$group\"}", "instant": false, "legendFormat": "{{pod_name}}", "range": true, diff --git a/test/cases/lifecycle/lifecycle.go b/test/cases/lifecycle/lifecycle.go index dee5ce05d..bcad93120 100644 --- a/test/cases/lifecycle/lifecycle.go +++ b/test/cases/lifecycle/lifecycle.go @@ -252,16 +252,26 @@ func verifyMigrationMetrics(reg observability.MetricsRegistry) { body := rec.Body.String() // Last-run metrics (banyandb_lifecycle_last_run_timestamp_seconds + // banyandb_lifecycle_last_run_success) are stamped by the deferred - // recordLastRun() at the end of action(). A successful cycle must set + // recordLastRun() at the end of action() with the (remote_node, + // remote_role, remote_tier, group) label set, sourced from the + // cycle's last processed group. A successful cycle must set // success=1 with a non-zero epoch; an empty value would mean the // gauges were never registered (PreRun not run) or the action never // reached the defer. Prometheus emits floats in scientific notation // for large values like epoch seconds (e.g. 1.781007822e+09), so the - // assertion accepts either fixed or scientific form. - gomega.Expect(body).To(gomega.MatchRegexp(`(?m)^banyandb_lifecycle_last_run_timestamp_seconds (?:[1-9]\d{9}|[1-9]\.\d+e\+0?[89])`), - "banyandb_lifecycle_last_run_timestamp_seconds must be set to a non-zero epoch, got:\n"+body) - gomega.Expect(body).To(gomega.MatchRegexp(`(?m)^banyandb_lifecycle_last_run_success 1$`), - "banyandb_lifecycle_last_run_success must be 1 after a successful cycle, got:\n"+body) + // assertion accepts either fixed or scientific form. The metric + // names now carry labels — the regex requires all four label names + // (remote_node, remote_role, remote_tier, group) and any value, so a + // missing label (regression to the unlabeld form) fails the regex. + // The label block is captured as `[^}]*` between each named label + // so the four names can appear in any order Prometheus emits them. + allLabels := `[^}]*remote_node="[^"]*"[^}]*remote_role="[^"]*"[^}]*remote_tier="[^"]*"[^}]*group="[^"]*"[^}]*` + gomega.Expect(body).To(gomega.MatchRegexp( + `(?m)^banyandb_lifecycle_last_run_timestamp_seconds\{`+allLabels+`\} (?:[1-9]\d{9}|[1-9]\.\d+e\+0?[89])`), + "banyandb_lifecycle_last_run_timestamp_seconds must be set to a non-zero epoch with all four labels, got:\n"+body) + gomega.Expect(body).To(gomega.MatchRegexp( + `(?m)^banyandb_lifecycle_last_run_success\{`+allLabels+`\} 1$`), + "banyandb_lifecycle_last_run_success must be 1 after a successful cycle, with all four labels, got:\n"+body) // A successful migration send increments total_finished; the measure/stream/trace // part files are sent via the file-sync operation, so that label must be present. gomega.Expect(body).To(gomega.MatchRegexp(`banyandb_lifecycle_migration_total_finished\{[^}]*\} [1-9]`),
