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/10 06:53:06 UTC
[PR] Refactor dump and lifecycle package to prevint OOM and ignored error message [skywalking-banyandb]
mrproliu opened a new pull request, #1168:
URL: https://github.com/apache/skywalking-banyandb/pull/1168
## Background
The lifecycle migration moves data down the storage tiers (hot → warm → cold). For
multi-target groups it uses **row-replay**: it reads each source part and re-publishes
every row through the normal Write API so the target tier re-shards and re-indexes it.
This PR fixes two production problems found while migrating real SkyWalking metrics data
(`sw_metricsHour` / `sw_metricsMinute`): the data nodes were **OOM-killed** during
row-replay, and parts with unrecoverable rows caused **hard migration failures**.
## Problem 1 — OOM during row-replay
Migrating large measure parts drove a data node's heap to **~1.5 GB** and the pod was
OOM-killed, stalling the whole migration. Three things compounded:
1. The dump reader materialized a part's column data far beyond a single block — the
decoder's internal buffer grew to hold the whole part's string data.
2. Every replayed row allocated a **fresh marshal buffer**, so a part with millions of
rows produced millions of short-lived large allocations.
3. There was **no bound on in-flight batches** — rows were marshaled and queued faster
than they were confirmed, so sent-but-unconfirmed payloads piled up in memory.
### How it's fixed
The row-replay path is rebuilt around **bounded, reusable memory**:
1. **Streaming columnar iterator** — the part is walked one block at a time
(`IterateColumnar`) using reusable scratch buffers. The decoder is reset per block, so
the decode buffer is bounded to a single block instead of the whole part.
2. **Size-classed marshal buffer pool** (`row_replay_bufpool.go`) — each row borrows one
reusable buffer grouped by power-of-two size class; the whole batch's buffers are
returned together once `Publish` has put every body on the wire. Grouping by size class
keeps a small body from pinning a large buffer, and lazy reclamation drops classes left
idle for several flushes so a transient burst of large rows is released rather than
hoarded.
3. **Bounded batch + bounded confirmation pipeline** — a batch flushes when it reaches
either `rowReplayMaxBatchRows` rows or **`rowReplayMaxBatchBytes` (32 MiB)** of marshaled
bytes, and a confirmation pipeline caps how many batches may be sent-but-unconfirmed
(`rowReplayMaxInflight`). Together these put a hard ceiling on outstanding buffers
regardless of part size.
**Result:** peak heap drops from **~1.5 GB to ~200–300 MB**, independent of part size.
## Problem 2 — hard failure on unresolvable rows
Some source parts can't have their rows republished because the entity identity can't be
recovered:
- **`rebuild-failed`** — the series-index (sidx) has a gap (no `EntityValues` for a
`seriesID`); the part's entity columns are present but no candidate's recomputed hash
matches, so the entity can't be rebuilt.
- **`incomplete-part`** — the part carries no entity tag columns at all, so neither sidx
nor column rebuild can recover the entity.
Previously this **hard-failed** the part/segment migration.
### How it's handled
These rows are now **skipped and retained** instead of failing the migration:
- The unresolved rows are skipped; the rest of the part migrates normally and the part is
marked complete.
- A structured error is recorded with a bounded sample of the skipped series (part,
`seriesID`, reason) so an operator can locate the exact source part — not just a count.
- The **source segment is retained** (excluded from post-migration deletion), so skipped
rows are never lost — they stay on the source tier and can still be queried.
## Error reporting consolidation (related)
All migration errors are consolidated into a single flat list of **structured, stage-aware
entries** (`source_stage` / `target_stage` / `group` / `catalog` / `scope` / `segment` /
`shard` / `part` / `interval`), replacing the previous nested error maps. The report still
exposes them under named buckets, and the `summary.*.errors` counts derive from the same
single source.
## Validation (real cluster, full hot→warm→cold migration)
- **Migration: 288/288 parts, 100% complete, 0 hard failures, 0 node-level send errors.**
~40M rows replayed.
- **Memory: 0 OOM-kills, 0 restarts** across all data/lifecycle containers during the
migration window. The active row-replay process peaked at **~205 MiB** (vs ~1.5 GB
before).
- **Skip-and-retain: ~1.1M rows** skipped as `series-index gap / incomplete-part`, each
recorded as a structured error with the source segment retained.
- **Data is queryable** after migration via the gRPC `MeasureService/Query`: old time
ranges return **0 on `hot`** (correctly migrated out) and **full results on `warm` and
`cold`**.
- [ ] 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] Refactor dump and lifecycle package to prevint OOM and ignored error message [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1168:
URL: https://github.com/apache/skywalking-banyandb/pull/1168#discussion_r3386149501
##########
banyand/internal/storage/segment.go:
##########
@@ -733,6 +727,28 @@ func parseSegmentTime(value string, unit IntervalUnit) (time.Time, error) {
panic("invalid interval unit")
}
+// ParseSegmentTime parses a segment-directory suffix back into its start time
+// for the given interval unit, the inverse of the suffix formatting used when a
+// segment directory is named. Callers outside this package use it to map the
+// suffixes returned by VisitSegmentsInTimeRange to the segment instants a
+// visitor observed.
+func ParseSegmentTime(suffix string, rule IntervalRule) (time.Time, error) {
+ return parseSegmentTime(suffix, rule.Unit)
+}
+
+// FormatSegmentTime formats a segment start time into its directory suffix for
+// the given interval rule, the inverse of ParseSegmentTime. The returned suffix
+// excludes the "seg-" directory prefix.
+func FormatSegmentTime(t time.Time, rule IntervalRule) string {
+ switch rule.Unit {
+ case HOUR:
+ return t.Format(hourFormat)
+ case DAY:
+ return t.Format(dayFormat)
+ }
+ panic("invalid interval unit")
+}
Review Comment:
`ParseSegmentTime` parses suffixes in `time.Local`, but `FormatSegmentTime` formats using whatever location is carried by `t`. If callers pass a UTC (or otherwise non-local) time, the formatted suffix may not round-trip through `ParseSegmentTime`, breaking the documented “inverse” relationship and making suffixes timezone-dependent.
Normalizing `t` to `time.Local` before formatting keeps formatting/parsing consistent.
##########
banyand/backup/lifecycle/migration_error_test.go:
##########
@@ -0,0 +1,296 @@
+// 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 (
+ "encoding/json"
+ "fmt"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// TestFormatIntervalRule pins the compact interval rendering for the report's
+// interval field across day and hour granularities.
+func TestFormatIntervalRule(t *testing.T) {
+ assert.Equal(t, "5d", formatIntervalRule(storage.IntervalRule{Unit: storage.DAY, Num: 5}))
+ assert.Equal(t, "1d", formatIntervalRule(storage.IntervalRule{Unit: storage.DAY, Num: 1}))
+ assert.Equal(t, "1h", formatIntervalRule(storage.IntervalRule{Unit: storage.HOUR, Num: 1}))
+ assert.Equal(t, "12h", formatIntervalRule(storage.IntervalRule{Unit: storage.HOUR, Num: 12}))
+}
+
+// TestSegmentErrorLocation pins the source segment directory name + interval
+// derivation, the inverse of storage.ParseSegmentTime, for day and hour units.
+func TestSegmentErrorLocation(t *testing.T) {
+ start := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
+ dayTR := ×tamp.TimeRange{Start: start, End: start.Add(5 * 24 * time.Hour)}
+ seg, interval := segmentErrorLocation(dayTR, storage.IntervalRule{Unit: storage.DAY, Num: 5})
+ assert.Equal(t, "seg-20260601", seg)
+ assert.Equal(t, "5d", interval)
+
+ hourTR := ×tamp.TimeRange{Start: start, End: start.Add(time.Hour)}
+ seg, interval = segmentErrorLocation(hourTR, storage.IntervalRule{Unit: storage.HOUR, Num: 1})
+ assert.Equal(t, "seg-2026060100", seg)
+ assert.Equal(t, "1h", interval)
Review Comment:
This test uses `time.UTC` for the start time, but the segment suffix logic in storage is defined in terms of `time.Local` (`ParseSegmentTime` uses `time.Local`, and `FormatSegmentTime` should match). In non-UTC environments the UTC date/hour can differ from the local one, making the expected suffixes here timezone-dependent.
Construct the start time in `time.Local` so the expectations are stable across environments.
##########
banyand/backup/lifecycle/row_replay_pipeline.go:
##########
@@ -146,33 +175,116 @@ type batchSender struct {
client queue.Client
publisher queue.BatchPublisher
pipeline *confirmPipeline
+ pool *marshalBufferPool
batch []bus.Message
- topic bus.Topic
- timeout time.Duration
- batchSize int
+ lent [][]byte
+ // skippedDetail locates a bounded sample of skipped series (part, seriesID,
+ // reason) so the migration report can point an operator at the source data,
+ // not just a count. Capped at maxSkipDetail to bound memory.
+ skippedDetail []skipError
+ topic bus.Topic
+ timeout time.Duration
+ maxRows int
+ maxBytes int
+ inflightBytes int
+ // Flush-reason tallies: which cap forced each batch out. A flush triggered by
+ // both caps at once is attributed to bytes (the memory-protecting cap that
+ // fires early in the big-body cluster). tailFlushes counts the trailing
+ // end-of-part flush that neither cap triggered.
+ rowLimitFlushes int
+ byteLimitFlushes int
+ tailFlushes int
+ // skippedRows counts rows dropped because their series could not be resolved
+ // (errSkipSeries: a part block referencing a series absent from the segment's
+ // sidx that could not be rebuilt from the part's columns). The part still
+ // completes; this surfaces how much a source-data gap dropped instead of
+ // aborting the whole migration.
+ skippedRows int
+}
+
+// maxSkipDetail bounds how many distinct skipped-series locations are retained
+// for the report; the full count still accrues in skippedRows.
+const maxSkipDetail = 200
+
+// recordSkip tallies one skipped series and, up to the cap, retains its location.
+func (s *batchSender) recordSkip(err error) {
+ s.skippedRows++
+ if c := asSkipError(err); c != nil && len(s.skippedDetail) < maxSkipDetail {
+ s.skippedDetail = append(s.skippedDetail, *c)
+ }
}
Review Comment:
`skippedDetail` is intended to hold a bounded sample of *skipped series locations*, but `recordSkip` appends an entry for every skipped row. For series with many rows, this quickly fills the sample with duplicates of the same (partPath, seriesID, reason), reducing its diagnostic value.
Since rows for a series are emitted contiguously, de-duplicate against the last recorded entry so the sample is effectively per-series while keeping the implementation lightweight.
--
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] Refactor dump and lifecycle package to prevint OOM and ignored error message [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #1168:
URL: https://github.com/apache/skywalking-banyandb/pull/1168
--
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] Refactor dump and lifecycle package to prevint OOM and ignored error message [skywalking-banyandb]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1168:
URL: https://github.com/apache/skywalking-banyandb/pull/1168#issuecomment-4667617518
## [Codecov](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `2 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 45.58%. Comparing base ([`3530dd9`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/3530dd9b5cefaa4fef60eec91ecbfbdfcaf243c5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`95959e1`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/95959e1895bf8b26c18df8a76b5a745a849eb70c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 287 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [pkg/bus/bus.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168?src=pr&el=tree&filepath=pkg%2Fbus%2Fbus.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2J1cy9idXMuZ28=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #1168 +/- ##
==========================================
- Coverage 45.97% 45.58% -0.40%
==========================================
Files 328 220 -108
Lines 55505 24916 -30589
==========================================
- Hits 25520 11358 -14162
+ Misses 27909 12540 -15369
+ Partials 2076 1018 -1058
```
| [Flag](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [fodc](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.70% <ø> (?)` | |
| [pkg](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `41.11% <0.00%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1168?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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] Refactor dump and lifecycle package to prevint OOM and ignored error message [skywalking-banyandb]
Posted by "mrproliu (via GitHub)" <gi...@apache.org>.
mrproliu commented on code in PR #1168:
URL: https://github.com/apache/skywalking-banyandb/pull/1168#discussion_r3386325761
##########
banyand/backup/lifecycle/migration_error_test.go:
##########
@@ -0,0 +1,296 @@
+// 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 (
+ "encoding/json"
+ "fmt"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// TestFormatIntervalRule pins the compact interval rendering for the report's
+// interval field across day and hour granularities.
+func TestFormatIntervalRule(t *testing.T) {
+ assert.Equal(t, "5d", formatIntervalRule(storage.IntervalRule{Unit: storage.DAY, Num: 5}))
+ assert.Equal(t, "1d", formatIntervalRule(storage.IntervalRule{Unit: storage.DAY, Num: 1}))
+ assert.Equal(t, "1h", formatIntervalRule(storage.IntervalRule{Unit: storage.HOUR, Num: 1}))
+ assert.Equal(t, "12h", formatIntervalRule(storage.IntervalRule{Unit: storage.HOUR, Num: 12}))
+}
+
+// TestSegmentErrorLocation pins the source segment directory name + interval
+// derivation, the inverse of storage.ParseSegmentTime, for day and hour units.
+func TestSegmentErrorLocation(t *testing.T) {
+ start := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
+ dayTR := ×tamp.TimeRange{Start: start, End: start.Add(5 * 24 * time.Hour)}
+ seg, interval := segmentErrorLocation(dayTR, storage.IntervalRule{Unit: storage.DAY, Num: 5})
+ assert.Equal(t, "seg-20260601", seg)
+ assert.Equal(t, "5d", interval)
+
+ hourTR := ×tamp.TimeRange{Start: start, End: start.Add(time.Hour)}
+ seg, interval = segmentErrorLocation(hourTR, storage.IntervalRule{Unit: storage.HOUR, Num: 1})
+ assert.Equal(t, "seg-2026060100", seg)
+ assert.Equal(t, "1h", interval)
Review Comment:
No need to change; it's the original design.
##########
banyand/internal/storage/segment.go:
##########
@@ -733,6 +727,28 @@ func parseSegmentTime(value string, unit IntervalUnit) (time.Time, error) {
panic("invalid interval unit")
}
+// ParseSegmentTime parses a segment-directory suffix back into its start time
+// for the given interval unit, the inverse of the suffix formatting used when a
+// segment directory is named. Callers outside this package use it to map the
+// suffixes returned by VisitSegmentsInTimeRange to the segment instants a
+// visitor observed.
+func ParseSegmentTime(suffix string, rule IntervalRule) (time.Time, error) {
+ return parseSegmentTime(suffix, rule.Unit)
+}
+
+// FormatSegmentTime formats a segment start time into its directory suffix for
+// the given interval rule, the inverse of ParseSegmentTime. The returned suffix
+// excludes the "seg-" directory prefix.
+func FormatSegmentTime(t time.Time, rule IntervalRule) string {
+ switch rule.Unit {
+ case HOUR:
+ return t.Format(hourFormat)
+ case DAY:
+ return t.Format(dayFormat)
+ }
+ panic("invalid interval unit")
+}
Review Comment:
No need to change; it's the original design.
--
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