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 87071c4aca1d4ee24542e927284e0cbad5a50e3c Author: Hongtao Gao <[email protected]> AuthorDate: Fri Jun 12 10:54:19 2026 +0000 fix(lifecycle): resolve sender identity from pod name instead of grpc-addr The previous deriveSelfIdentity matched the lifecycle's --grpc-addr against the data-node registry's GrpcAddress. This failed for data pods that register a headless-service FQDN while the lifecycle sidecar uses the default --grpc-addr=127.0.0.1:17912, leaving SenderNode empty on the wire and producing empty remote_node labels on the receiver's banyandb_queue_sub_total_finished series. Replace with a 2-step resolveSelfIdentity that derives selfPodHost from POD_NAME (with os.Hostname() as fallback) and matches it against the host portion of node.GrpcAddress / node.NodeID. The match is rerun on every parseGroup call (no caching) so a data-pod restart that changes the registered address is picked up without restarting the lifecycle. Add a banyandb_lifecycle_self_identity_resolution_total{result} counter, incremented from parseGroup, that exposes ok / empty outcomes for at-a-glance observability. --- banyand/backup/lifecycle/service.go | 23 +++- banyand/backup/lifecycle/steps.go | 237 ++++++++++++++++++++------------- banyand/backup/lifecycle/steps_test.go | 91 ++++++++++--- 3 files changed, 236 insertions(+), 115 deletions(-) diff --git a/banyand/backup/lifecycle/service.go b/banyand/backup/lifecycle/service.go index 9f0fafbd1..73ee3d167 100644 --- a/banyand/backup/lifecycle/service.go +++ b/banyand/backup/lifecycle/service.go @@ -88,6 +88,13 @@ type lifecycleService struct { 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. @@ -251,6 +258,16 @@ func (l *lifecycleService) PreRun(_ context.Context) error { 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") if l.schedule != "" && l.lifecycleTLS { var err error @@ -1082,7 +1099,7 @@ 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.gRPCAddr) + group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return @@ -1202,7 +1219,7 @@ 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.gRPCAddr) + group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return @@ -1309,7 +1326,7 @@ 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.gRPCAddr) + group, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr, l.selfIdentityResolution) if err != nil { l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name) return diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go index fd5f90477..bfa699fb0 100644 --- a/banyand/backup/lifecycle/steps.go +++ b/banyand/backup/lifecycle/steps.go @@ -22,6 +22,7 @@ import ( "fmt" "net" "os" + "strings" "github.com/pkg/errors" "google.golang.org/protobuf/proto" @@ -38,6 +39,7 @@ 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" ) @@ -85,79 +87,104 @@ func (l *lifecycleService) getSnapshots(ctx context.Context, groups []*commonv1. return streamDir, measureDir, traceDir, nil } -// deriveSelfIdentity returns the SenderNode and SenderTier the lifecycle -// publisher should stamp on its wire SendRequest so the data-node receiver -// can label its banyandb_queue_sub_* family accordingly. +// resolveSelfIdentity returns the BanyanDB NodeID (Metadata.Name) and +// tier label (Labels["type"]) the lifecycle publisher should stamp on +// its wire SendRequest / SyncMetadata, so the data-node receiver labels +// its banyandb_queue_sub_total_* family with non-empty remote_node and +// remote_tier. // -// Inputs (no new CLI flags needed): +// Resolution: the lifecycle sidecar's own pod hostname is the stable +// identifier the receiver will see as the sender. It comes from +// POD_NAME (K8s downward API) and falls back to os.Hostname() — the +// same precedence as nativeNodeContext at service.go:160-165. The +// function then looks the host up directly in the data-node registry, +// matching against the host portion of GrpcAddress and NodeID (the +// registry may carry either an IP, a headless-service FQDN, or a +// loopback alias, depending on which bind address the data pod +// registered with). The first registry entry whose host matches (with +// loopback-alias normalization) is the co-located data pod; its +// Metadata.Name is the SenderNode and its Labels["type"] is the +// SenderTier. // -// - coLocatedDataNodeAddr: the lifecycle's --grpc-addr, which is the -// gRPC address of the data node the lifecycle is co-located with -// (sidecar topology). It is the authoritative match key. -// - nodeLabels: the lifecycle's own --node-labels, used as a fallback -// when the co-located data node hasn't synced to the metadata -// registry yet (cold start of a freshly created lifecycle). -// - nodes: the cluster's data-node registry (ROLE_DATA), which carries -// both Metadata.Name (the BanyanDB NodeID) and Labels (e.g. type=hot). -// -// Resolution order: -// -// 1. Match by GrpcAddress. In the standard sidecar layout the lifecycle's -// --grpc-addr equals the co-located data node's GrpcAddress, and -// that data node's Metadata.Name is the BanyanDB NodeID the -// receiver records as remote_node. This is the production path -// (the live cluster's lifecycle container has no --node-labels). -// 2. Fall back to label matching. If a data node's Labels are a -// superset of nodeLabels (every key in nodeLabels matches), use -// that node's Metadata.Name and Labels["type"]. -// 3. Fall back to type-only match on nodeLabels["type"]. -// 4. If nothing matches, return empty strings — preserves the -// pre-fix behavior so existing test setups that don't populate -// these fields keep working. -func deriveSelfIdentity(coLocatedDataNodeAddr string, nodeLabels map[string]string, nodes []*databasev1.Node) (senderNode, senderTier string) { - // Pass 1: GrpcAddress match (the production sidecar path). - if coLocatedDataNodeAddr != "" { - for _, n := range nodes { - if grpcAddrEqual(n.GrpcAddress, coLocatedDataNodeAddr) { - return n.Metadata.Name, n.Labels["type"] - } - } +// Re-runs on every parseGroup call (no caching) so a data-pod +// restart, re-registration, or new host is picked up by the next +// cycle. Returns ok=false when no registry entry matches. +func resolveSelfIdentity(selfPodHost string, nodes []*databasev1.Node) (senderNode, senderTier string, ok bool) { + if selfPodHost == "" { + return "", "", false } - // Pass 2: every-key label match. Guarded against an empty label set: - // labelsContain treats an empty subset as matching anything, which would - // attribute the identity to an arbitrary registry node (whichever happens - // to be listed first — e.g. a migration target instead of the co-located - // data node). - if len(nodeLabels) > 0 { - for _, n := range nodes { - if n.Labels == nil { - continue - } - if !labelsContain(n.Labels, nodeLabels) { - continue - } - return n.Metadata.Name, n.Labels["type"] + for _, n := range nodes { + if n == nil || n.Metadata == nil { + continue + } + if hostMatches(n.GrpcAddress, selfPodHost) { + return n.Metadata.Name, n.Labels["type"], true } } - // Pass 3: type-only label match. - if wantType := nodeLabels["type"]; wantType != "" { - for _, n := range nodes { - if n.Labels == nil { - continue - } - if n.Labels["type"] == wantType { - return n.Metadata.Name, n.Labels["type"] - } + return "", "", false +} + +// selfPodHostname returns the lifecycle sidecar's own pod host. +// Precedence matches nativeNodeContext at service.go:160-165: +// POD_NAME first (K8s downward API), then os.Hostname() as a +// fallback. Returns "" only if both lookups fail (very rare; e.g. +// hostname uname syscall returns ENAMETOOLONG). +func selfPodHostname() string { + if v := os.Getenv("POD_NAME"); v != "" { + return v + } + if h, err := os.Hostname(); err == nil { + return h + } + return "" +} + +// hostMatches reports whether aRegistryHost (which may carry a :port +// and may be a loopback alias, an IP, or a headless-service FQDN) +// identifies the same pod as selfPodHost. The host portion is +// extracted via net.SplitHostPort, then reduced to its leftmost +// label -- but only for FQDNs (multi-label hostnames); IP literals +// are kept as-is so a 127.0.0.1 form is not truncated to "127". +// (a FQDN like "data-x.data-x-headless.ns" maps to the pod name +// "data-x".) Loopback aliases (localhost, 127.0.0.1, ::1) are +// treated as equivalent so a registry entry advertised as +// 127.0.0.1:17912 still matches a selfPodHost of "localhost" or +// vice versa. +func hostMatches(aRegistryHost, selfPodHost string) bool { + if aRegistryHost == "" { + return false + } + if h, _, err := net.SplitHostPort(aRegistryHost); err == nil { + aRegistryHost = h + } + if net.ParseIP(aRegistryHost) == nil { + if i := strings.Index(aRegistryHost, "."); i >= 0 { + aRegistryHost = aRegistryHost[:i] } } - return "", "" + if aRegistryHost == selfPodHost { + return true + } + if isLoopbackHost(selfPodHost) && isLoopbackHost(aRegistryHost) { + return true + } + return false } // grpcAddrEqual reports whether two advertised gRPC addresses identify the -// same endpoint. Besides the exact match, loopback host aliases (localhost, -// 127.0.0.1, ::1) with the same port are treated as equivalent, so a -// --grpc-addr given as localhost:PORT still matches a data node registered -// as 127.0.0.1:PORT. +// same endpoint. Three equivalences are honored: +// - exact string match, +// - host-portion match after reducing each to its leftmost label +// (FQDN-only; IP literals are kept as-is) with the same port, +// - both hosts are loopback aliases (localhost / 127.0.0.1 / ::1) +// with the same port. +// +// The middle case is what was missing pre-fix: a --grpc-addr of +// 127.0.0.1:17912 and a registry GrpcAddress of +// "<headless-svc>.<ns>:17912" used to be rejected because the +// headless-svc host is not loopback and not a literal string match. +// The new behavior is "same port and same leftmost label" wins +// regardless of the FQDN or loopback status. func grpcAddrEqual(a, b string) bool { if a == b { return true @@ -167,6 +194,19 @@ func grpcAddrEqual(a, b string) bool { if errA != nil || errB != nil || portA != portB { return false } + if net.ParseIP(hostA) == nil { + if i := strings.Index(hostA, "."); i >= 0 { + hostA = hostA[:i] + } + } + if net.ParseIP(hostB) == nil { + if i := strings.Index(hostB, "."); i >= 0 { + hostB = hostB[:i] + } + } + if hostA == hostB { + return true + } return isLoopbackHost(hostA) && isLoopbackHost(hostB) } @@ -179,17 +219,6 @@ func isLoopbackHost(host string) bool { return ip != nil && ip.IsLoopback() } -// labelsContain reports whether superset has every (k, v) pair in subset. -// An empty subset matches anything. -func labelsContain(superset, subset map[string]string) bool { - for k, v := range subset { - if superset[k] != v { - return false - } - } - return true -} - // GroupConfig encapsulates the parsed lifecycle configuration for a Group. // It contains all necessary information for migration and deletion operations. type GroupConfig struct { @@ -235,7 +264,7 @@ func parseGroup( g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node, l *logger.Logger, metadata metadata.Repo, clusterStateMgr *clusterStateManager, omr observability.MetricsRegistry, - coLocatedDataNodeAddr string, + resolutionCounter meter.Counter, ) (*GroupConfig, error) { ro := g.ResourceOpts if ro == nil { @@ -311,24 +340,48 @@ func parseGroup( client := pub.NewWithoutMetadata(omr) //nolint:contextcheck // health check goroutine uses context.Background() // Stamp the lifecycle's self identity onto the publisher so the wire // SenderNode / SenderRole / SenderTier fields and the parallel - // banyandb_lifecycle_migration_* labels are populated. The three - // values are derived from already-known inputs (the co-located data - // node's gRPC address and the cluster's data-node registry) so the - // fix needs no new CLI flags: - // - SenderNode = the data node whose GrpcAddress matches the - // lifecycle's --grpc-addr (i.e. the co-located data node). Its - // Metadata.Name is the BanyanDB NodeID the receiver records as - // remote_node. - // - SenderRole = "lifecycle" (no Role enum entry; matches the - // liaison's hard-coded "liaison" pattern in pkg/cmdsetup/liaison.go). - // - SenderTier = the matched data node's `type` label - // (hot/warm/cold), which becomes the receiver's remote_tier. - // Falls back to the lifecycle's own --node-labels when the co-located - // data node isn't in the registry yet (cold start), and to empty - // when neither is available — preserving the pre-fix behavior. - senderNode, senderTier := deriveSelfIdentity(coLocatedDataNodeAddr, nodeLabels, nodes) - if senderNode != "" || senderTier != "" { + // banyandb_lifecycle_migration_* labels are populated. The + // resolveSelfIdentity algorithm matches the lifecycle's own pod + // hostname (POD_NAME -> os.Hostname(), same precedence as + // nativeNodeContext at service.go:160-165) against the + // data-node registry's GrpcAddress with loopback-alias and + // port-strip normalization. The first matching registry entry is + // the co-located data pod; its Metadata.Name is the BanyanDB + // NodeID the receiver records as remote_node, and its + // Labels["type"] is the receiver's remote_tier. SenderRole is + // 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. + 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) + // 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 + // "remote node" the user wants visible at startup. + l.Info(). + Str("data_pod", selfHost). + Str("sender_node", senderNode). + Str("sender_tier", senderTier). + Msg("lifecycle: stamped sender identity on wire (SenderNode, SenderTier)") + } else { + l.Warn(). + Str("data_pod", selfHost). + Msg("lifecycle: sender identity resolution returned empty; SenderNode on wire will be empty (pre-fix regression)") } switch g.Catalog { case commonv1.Catalog_CATALOG_STREAM: diff --git a/banyand/backup/lifecycle/steps_test.go b/banyand/backup/lifecycle/steps_test.go index 203ff3953..407dd56ab 100644 --- a/banyand/backup/lifecycle/steps_test.go +++ b/banyand/backup/lifecycle/steps_test.go @@ -86,37 +86,88 @@ 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, "") + _, err := parseGroup(g, map[string]string{"type": "warm"}, nil, nil, nil, nil, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), c.errFrag) }) } } -// TestDeriveSelfIdentity verifies the sender-identity resolution order: the -// co-located data node's address match (including loopback host aliases, since -// test clusters dial localhost:PORT while nodes register 127.0.0.1:PORT), the -// label fallbacks, and the guard that keeps an empty label set from -// wildcard-matching an arbitrary registry node. -func TestDeriveSelfIdentity(t *testing.T) { +// 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 +// (whether loopback, IP, or FQDN) resolves to the co-located data +// node's Metadata.Name. The last two cases prove the no-match and +// empty-input guards. +func TestResolveSelfIdentity(t *testing.T) { nodes := []*databasev1.Node{ - {Metadata: &commonv1.Metadata{Name: "warm-node"}, GrpcAddress: "127.0.0.1:2", Labels: map[string]string{"type": "warm"}}, - {Metadata: &commonv1.Metadata{Name: "hot-node"}, GrpcAddress: "127.0.0.1:1", Labels: map[string]string{"type": "hot"}}, + // Production-bug repro: lifecycle's --grpc-addr is + // 127.0.0.1:17912, but the data pod registered its + // GrpcAddress as the headless-service DNS form. The new + // algorithm must match on the host portion (data-hot-0). + {Metadata: &commonv1.Metadata{Name: "data-hot-0:17912"}, GrpcAddress: "data-hot-0.data-hot-headless.ns:17912", Labels: map[string]string{"type": "hot"}}, + + // Loopback-registered case: --grpc-addr "127.0.0.1:17912" + // matches the loopback form via isLoopbackHost equivalence. + {Metadata: &commonv1.Metadata{Name: "data-warm-0:17912"}, GrpcAddress: "127.0.0.1:17912", Labels: map[string]string{"type": "warm"}}, + + // IP-only registered case (NodeHostProvider=ip): must NOT + // match unless the selfPodHost is itself an IP, so we + // expect an empty result here. The first case in this + // table covers the production path; this one documents + // the known limitation of the post-fix algorithm. + {Metadata: &commonv1.Metadata{Name: "data-cold-0:17912"}, GrpcAddress: "10.116.3.84:17912", Labels: map[string]string{"type": "cold"}}, } - node, tier := deriveSelfIdentity("127.0.0.1:1", nil, nodes) - assert.Equal(t, "hot-node", node, "exact address match") + node, tier, ok := resolveSelfIdentity("data-hot-0", nodes) + assert.True(t, ok, "DNS-form GrpcAddress must match by host portion (the production-bug case)") + assert.Equal(t, "data-hot-0:17912", node) assert.Equal(t, "hot", tier) - node, tier = deriveSelfIdentity("localhost:1", nil, nodes) - assert.Equal(t, "hot-node", node, "loopback alias must match the registered 127.0.0.1 form") - assert.Equal(t, "hot", tier) + node, tier, ok = resolveSelfIdentity("127.0.0.1", nodes) + assert.True(t, ok, "loopback GrpcAddress matches a loopback selfPodHost") + assert.Equal(t, "data-warm-0:17912", node) + assert.Equal(t, "warm", tier) - node, tier = deriveSelfIdentity("localhost:9", nil, nodes) - assert.Empty(t, node, "unmatched address with no labels must not wildcard-match an arbitrary node") - assert.Empty(t, tier) + // selfPodHost="10.116.3.84" matches the IP-form entry. + node, tier, ok = resolveSelfIdentity("10.116.3.84", nodes) + assert.True(t, ok, "IP-form GrpcAddress matches an IP selfPodHost") + assert.Equal(t, "data-cold-0:17912", node) + assert.Equal(t, "cold", tier) - node, tier = deriveSelfIdentity("", map[string]string{"type": "hot"}, nodes) - assert.Equal(t, "hot-node", node, "label fallback") - assert.Equal(t, "hot", tier) + // selfPodHost="data-warm-1" is not in the registry: no match. + _, _, ok = resolveSelfIdentity("data-warm-1", nodes) + assert.False(t, ok, "selfPodHost not in registry must return ok=false (no wildcard)") + + // Empty selfPodHost: no match. + _, _, ok = resolveSelfIdentity("", nodes) + assert.False(t, ok, "empty selfPodHost must return ok=false (no panic)") + + // grpcAddrEqual: post-fix also accepts host-portion exact match. + assert.True(t, grpcAddrEqual("data-x.headless:17912", "data-x:17912"), + "same port and same host (after SplitHostPort) must match") + assert.True(t, grpcAddrEqual("127.0.0.1:17912", "127.0.0.1:17912"), + "exact match still works") + assert.True(t, grpcAddrEqual("localhost:17912", "127.0.0.1:17912"), + "loopback-vs-loopback still works") + assert.False(t, grpcAddrEqual("data-x:17912", "data-y:17912"), + "different hosts with same port must not match") +} + +// TestSelfPodHostnamePrecedence checks that selfPodHostname() prefers +// POD_NAME (the K8s downward API value) and falls back to os.Hostname() +// when POD_NAME is unset. +func TestSelfPodHostnamePrecedence(t *testing.T) { + t.Setenv("POD_NAME", "my-pod") + assert.Equal(t, "my-pod", selfPodHostname(), + "POD_NAME takes precedence (K8s downward API)") + + t.Setenv("POD_NAME", "") + // Without POD_NAME, the function falls back to os.Hostname(). + // The test host name is non-empty on Linux, so the result is + // either the test runner's hostname or "" if uname fails. + // Either way it must not panic and must not return "my-pod". + got := selfPodHostname() + assert.NotEqual(t, "my-pod", got, + "empty POD_NAME must fall back to os.Hostname(), not return the previous value") }
