You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "mrproliu (via GitHub)" <gi...@apache.org> on 2026/06/26 05:34:16 UTC
[PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
mrproliu opened a new pull request, #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193
## Why
Daily lifecycle tier-migration (hot→warm→cold) was OOM-killing the receiving warm/cold data node during external series-index introduction (exit 137), which cascaded into `chunk N: EOF` / `connection
reset` on the sender and `no nodes available` for subsequent groups. The root cause: bluge's external-segment deduplicator decompressed and retained every segment's stored fields (~1.8 GB heap per
segment), and because each gRPC sync stream runs its own goroutine with no shared memory budget, N concurrent senders stacked these materializations and blew past the cgroup limit.
## What
This PR removes the OOM at the source and adds always-on throttling so an unexpected memory spike degrades gracefully instead of killing the node. The bluge bump is the root-cause fix; the throttling
is defense-in-depth that backstops it.
### Root-cause fix
Bump bluge to a memory-efficient external-segment deduplicator that resolves duplicates via `_id` term-dictionary point lookups (`Dictionary.Contains`) instead of decompressing and retaining stored
fields. This drops per-segment heap from ~1.8 GB to near-zero during external-segment introduction, with no public API change and identical dedup semantics (verified by an upstream differential test).
### Throttling (always on)
The receiver and sender cooperate so the cases that can balloon receiver memory are bounded:
- **Per-chunk load-shed** — a receiver under high memory pressure (protector `StateHigh`) answers `SYNC_STATUS_SERVER_BUSY` for incoming part/series-index chunks; the sender stops streaming, backs
off, and retries the whole part.
- **Introduce back-pressure** — before the heavy `CompleteSegment` introduce step, the receiver waits (on the gRPC stream context) for memory to recover, bounded by
`{measure,stream,trace}-lifecycle-receive-mem-wait-timeout` (default 5m).
- **Transient send retry** — the sender wraps pick+send in one bounded exponential-backoff loop (`pickAndRun`), retrying transient failures (target node restarting / `no nodes available`, disconnect,
`SERVER_BUSY`) and re-picking a node on a failed send, bounded by `lifecycle-send-retry-timeout` (default 15m). This covers both the file-chunk path (all three catalogs) and the row-replay node-pick
path. A part that exhausts its budget is recorded incomplete and resumed on the next migration cycle.
## Flags & wire
Two new flags are added and documented (`docs/operation/configuration.md`, `docs/operation/lifecycle.md`): `{measure,stream,trace}-lifecycle-receive-mem-wait-timeout` (data node, default 5m) and
`lifecycle-send-retry-timeout` (lifecycle command, default 15m). The proto change is additive and backward-compatible: `SYNC_STATUS_SERVER_BUSY = 8` is appended to `cluster.v1.SyncStatus`, so peers
that don't recognize it fall back to default handling.
## Testing
A new integration test (`banyand/measure/migration_oom_throttling_test.go`) exercises the real measure tier-migration receive path with a test-controlled protector: it writes a real source tree,
brings up a real target `measure` receiver on a real chunked-sync gRPC server with an injected protector (a fake that embeds `protector.NewMemory` and overrides only `State()`), then — protector HIGH
— streams the series-index part and asserts the sender sees `queue.ErrServerBusy` (the `SERVER_BUSY` round-trip) and the segment is not introduced, then — protector LOW — asserts (`Eventually`) the
migration completes and the series index lands and is queryable. The injection is test-only via an `export_test.go` helper; no production code changes for the test.
Unit tests additionally cover the memory wait (recover / timeout / parent-cancel), the `queue.ChunkedSyncPartContext` context fallback, the `sub` BUSY translation, the `pub` BUSY recognition, and the
lifecycle retry classifiers + `pickWithRetry`. Existing migration e2e/copy suites pass unchanged, and `make build`, `go vet`, and the full-repo `make lint` (golangci-lint v1.64.8) are clean. This PR
also fixes a pre-existing file-descriptor leak on trace's `generateAll*PartData` error paths (already-opened handles are released before returning on a mid-way failure).
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#<issue number>.
- [x] Update the [`CHANGES` log](https://github.com/apache/skywalking-banyandb/blob/main/CHANGES.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193#discussion_r3479366744
##########
banyand/protector/wait.go:
##########
@@ -0,0 +1,63 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package protector
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ waitInitialBackoff = 200 * time.Millisecond
+ waitMaxBackoff = 2 * time.Second
+ waitHeartbeat = time.Minute
+)
+
+// WaitWhileHigh blocks while memory pressure is high, bounded by maxWait.
+// It only creates a timeout context when it actually has to wait. It returns
+// the timeout/cancel error if ctx or maxWait elapses before recovery.
+func WaitWhileHigh(ctx context.Context, pm Memory, maxWait time.Duration, l *logger.Logger, stage string) error {
+ if pm.State() != StateHigh {
+ return nil
+ }
+ waitCtx, cancel := context.WithTimeout(ctx, maxWait)
+ defer cancel()
+ start := time.Now()
+ l.Warn().Str("stage", stage).Dur("max_wait", maxWait).Msg("memory high, migration receive paused, waiting")
+ ticker := time.NewTicker(waitHeartbeat)
+ defer ticker.Stop()
+ backoff := waitInitialBackoff
+ for pm.State() == StateHigh {
+ select {
+ case <-time.After(backoff):
+ backoff *= 2
+ if backoff > waitMaxBackoff {
+ backoff = waitMaxBackoff
+ }
+ case <-ticker.C:
+ l.Warn().Str("stage", stage).Dur("waited", time.Since(start)).Msg("still waiting for memory")
+ case <-waitCtx.Done():
+ l.Warn().Str("stage", stage).Dur("waited", time.Since(start)).Msg("memory wait aborted (timeout/cancel)")
+ return waitCtx.Err()
+ }
+ }
Review Comment:
WaitWhileHigh uses time.After(backoff) inside the wait loop, which allocates a new timer on each backoff step. This is functionally correct, but a reusable time.Timer avoids these small repeated allocations during sustained high-memory waits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
Posted by "mrproliu (via GitHub)" <gi...@apache.org>.
mrproliu commented on code in PR #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193#discussion_r3479377993
##########
banyand/protector/wait.go:
##########
@@ -0,0 +1,63 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package protector
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ waitInitialBackoff = 200 * time.Millisecond
+ waitMaxBackoff = 2 * time.Second
+ waitHeartbeat = time.Minute
+)
+
+// WaitWhileHigh blocks while memory pressure is high, bounded by maxWait.
+// It only creates a timeout context when it actually has to wait. It returns
+// the timeout/cancel error if ctx or maxWait elapses before recovery.
+func WaitWhileHigh(ctx context.Context, pm Memory, maxWait time.Duration, l *logger.Logger, stage string) error {
+ if pm.State() != StateHigh {
+ return nil
+ }
+ waitCtx, cancel := context.WithTimeout(ctx, maxWait)
+ defer cancel()
+ start := time.Now()
+ l.Warn().Str("stage", stage).Dur("max_wait", maxWait).Msg("memory high, migration receive paused, waiting")
+ ticker := time.NewTicker(waitHeartbeat)
+ defer ticker.Stop()
+ backoff := waitInitialBackoff
+ for pm.State() == StateHigh {
+ select {
+ case <-time.After(backoff):
+ backoff *= 2
+ if backoff > waitMaxBackoff {
+ backoff = waitMaxBackoff
+ }
+ case <-ticker.C:
+ l.Warn().Str("stage", stage).Dur("waited", time.Since(start)).Msg("still waiting for memory")
+ case <-waitCtx.Done():
+ l.Warn().Str("stage", stage).Dur("waited", time.Since(start)).Msg("memory wait aborted (timeout/cancel)")
+ return waitCtx.Err()
+ }
+ }
Review Comment:
no need to be fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193#discussion_r3479914429
##########
banyand/backup/lifecycle/retry.go:
##########
@@ -0,0 +1,133 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package lifecycle
+
+import (
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/cenkalti/backoff/v4"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+)
+
+// defaultLifecycleSendRetryTimeout bounds the total wall-clock spent retrying
+// transient failures when streaming a part to a target node (e.g. while the
+// target node restarts). Configurable via --lifecycle-send-retry-timeout.
+const defaultLifecycleSendRetryTimeout = 15 * time.Minute
+
+// lifecycleSendRetryTimeout is the active bound, set by the lifecycle flag.
+var lifecycleSendRetryTimeout = defaultLifecycleSendRetryTimeout
+
+// isNoNodes reports whether err means the target node could not be picked
+// because no node is currently available (e.g. the target is restarting).
+// It matches both the typed sentinel and the plain round-robin error string.
+func isNoNodes(err error) bool {
+ if errors.Is(err, node.ErrNoAvailableNode) {
+ return true
+ }
+ return err != nil && strings.Contains(err.Error(), "no nodes available")
+}
+
+// isTransientSend reports whether err from streaming a part to a node is
+// transient and worth retrying: receiver back-pressure (SERVER_BUSY) or a
+// transient/failover gRPC error such as Unavailable, ResourceExhausted, or a
+// dropped connection.
+func isTransientSend(err error) bool {
+ return errors.Is(err, queue.ErrServerBusy) || grpchelper.IsTransientError(err) || grpchelper.IsFailoverError(err)
+}
+
+// retryLogInterval throttles the "still retrying" heartbeat so a long wait for
+// a restarting node logs at most once per minute instead of on every attempt.
+const retryLogInterval = time.Minute
+
+// nodePicker is the one method of node.Selector that the send-retry path needs;
+// narrowing it keeps runWithRetry/pickWithRetry decoupled and easy to test.
+type nodePicker interface {
+ Pick(group, name string, shardID, replicaID uint32) (string, error)
+}
+
+// runWithRetry runs op under bounded exponential backoff capped by
+// lifecycleSendRetryTimeout. op returns a transient error to retry or
+// backoff.Permanent(err) to stop immediately. While retrying it logs a
+// heartbeat at most once per minute so a long wait for a restarting node is not
+// mistaken for a hang.
+func runWithRetry(l *logger.Logger, stage string, op func() error) error {
+ bo := backoff.NewExponentialBackOff()
+ bo.MaxElapsedTime = lifecycleSendRetryTimeout
+ start := time.Now()
+ var lastLog time.Time
+ return backoff.RetryNotify(op, bo, func(err error, _ time.Duration) {
Review Comment:
The send-retry loop ignores context cancellation. `backoff.RetryNotify(op, bo, ...)` has no context, and the op it ultimately drives (`streamPartToNode` -> `SyncStreamingParts`) is called with `context.Background()`. So once a part starts retrying, a migration that needs to abort (lifecycle service shutdown, operator cancel) stays stuck here for up to `lifecycle-send-retry-timeout` (15m default).
Note the asymmetry with the receive side: `protector.WaitWhileHigh` already honors its `ctx`, but this send path does not.
Request: thread a cancellable context through and use `backoff.RetryNotify(op, backoff.WithContext(bo, ctx), ...)`. The row-replay callers (`routeColumnar`, `routeAndEnqueue`) already have a `ctx` in scope to pass down; for the file-chunk visitor paths a migration-scoped context would also let `streamPartToNode` stop streaming on cancel instead of relying solely on the 15m cap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(lifecycle): throttle migration under receiver memory pressure; fix bluge dedup OOM [skywalking-banyandb]
Posted by "mrproliu (via GitHub)" <gi...@apache.org>.
mrproliu commented on code in PR #1193:
URL: https://github.com/apache/skywalking-banyandb/pull/1193#discussion_r3480388461
##########
banyand/backup/lifecycle/retry.go:
##########
@@ -0,0 +1,133 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package lifecycle
+
+import (
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/cenkalti/backoff/v4"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+)
+
+// defaultLifecycleSendRetryTimeout bounds the total wall-clock spent retrying
+// transient failures when streaming a part to a target node (e.g. while the
+// target node restarts). Configurable via --lifecycle-send-retry-timeout.
+const defaultLifecycleSendRetryTimeout = 15 * time.Minute
+
+// lifecycleSendRetryTimeout is the active bound, set by the lifecycle flag.
+var lifecycleSendRetryTimeout = defaultLifecycleSendRetryTimeout
+
+// isNoNodes reports whether err means the target node could not be picked
+// because no node is currently available (e.g. the target is restarting).
+// It matches both the typed sentinel and the plain round-robin error string.
+func isNoNodes(err error) bool {
+ if errors.Is(err, node.ErrNoAvailableNode) {
+ return true
+ }
+ return err != nil && strings.Contains(err.Error(), "no nodes available")
+}
+
+// isTransientSend reports whether err from streaming a part to a node is
+// transient and worth retrying: receiver back-pressure (SERVER_BUSY) or a
+// transient/failover gRPC error such as Unavailable, ResourceExhausted, or a
+// dropped connection.
+func isTransientSend(err error) bool {
+ return errors.Is(err, queue.ErrServerBusy) || grpchelper.IsTransientError(err) || grpchelper.IsFailoverError(err)
+}
+
+// retryLogInterval throttles the "still retrying" heartbeat so a long wait for
+// a restarting node logs at most once per minute instead of on every attempt.
+const retryLogInterval = time.Minute
+
+// nodePicker is the one method of node.Selector that the send-retry path needs;
+// narrowing it keeps runWithRetry/pickWithRetry decoupled and easy to test.
+type nodePicker interface {
+ Pick(group, name string, shardID, replicaID uint32) (string, error)
+}
+
+// runWithRetry runs op under bounded exponential backoff capped by
+// lifecycleSendRetryTimeout. op returns a transient error to retry or
+// backoff.Permanent(err) to stop immediately. While retrying it logs a
+// heartbeat at most once per minute so a long wait for a restarting node is not
+// mistaken for a hang.
+func runWithRetry(l *logger.Logger, stage string, op func() error) error {
+ bo := backoff.NewExponentialBackOff()
+ bo.MaxElapsedTime = lifecycleSendRetryTimeout
+ start := time.Now()
+ var lastLog time.Time
+ return backoff.RetryNotify(op, bo, func(err error, _ time.Duration) {
Review Comment:
Done. Add the context in the whole lifecycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org