You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/08/04 10:09:05 UTC
(skywalking-banyandb) 02/03: Refactor Progress struct for shard-specific tracking in migration processes
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch lifecyc-sync
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit e98668e7bda980239f7a3ffc31d2aa2908600b1e
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sat Aug 2 17:19:51 2025 +0800
Refactor Progress struct for shard-specific tracking in migration processes
- Updated the `Progress` struct to use nested maps for `CompletedStreamElementIndex` and `StreamElementIndexErrors`, allowing for tracking of completion and errors at the shard level.
- Modified methods in `stream_migration_visitor.go` to incorporate `sourceShardID` in progress tracking, enhancing the accuracy of completed checks and error handling during element index processing.
---
banyand/backup/lifecycle/progress.go | 37 +++++++++++++++-------
.../backup/lifecycle/stream_migration_visitor.go | 7 ++--
2 files changed, 29 insertions(+), 15 deletions(-)
diff --git a/banyand/backup/lifecycle/progress.go b/banyand/backup/lifecycle/progress.go
index 609f7208..450231c8 100644
--- a/banyand/backup/lifecycle/progress.go
+++ b/banyand/backup/lifecycle/progress.go
@@ -37,8 +37,8 @@ type Progress struct {
StreamPartErrors map[string]map[common.ShardID]map[uint64]string `json:"stream_part_errors"`
CompletedStreamSeries map[string]map[common.ShardID]bool `json:"completed_stream_series"`
StreamSeriesErrors map[string]map[common.ShardID]string `json:"stream_series_errors"`
- CompletedStreamElementIndex map[string]bool `json:"completed_stream_element_index"`
- StreamElementIndexErrors map[string]string `json:"stream_element_index_errors"`
+ CompletedStreamElementIndex map[string]map[common.ShardID]bool `json:"completed_stream_element_index"`
+ StreamElementIndexErrors map[string]map[common.ShardID]string `json:"stream_element_index_errors"`
StreamPartCounts map[string]int `json:"stream_part_counts"`
StreamPartProgress map[string]int `json:"stream_part_progress"`
StreamSeriesCounts map[string]int `json:"stream_series_counts"`
@@ -87,8 +87,8 @@ func NewProgress(path string, l *logger.Logger) *Progress {
StreamSeriesErrors: make(map[string]map[common.ShardID]string),
StreamSeriesCounts: make(map[string]int),
StreamSeriesProgress: make(map[string]int),
- CompletedStreamElementIndex: make(map[string]bool),
- StreamElementIndexErrors: make(map[string]string),
+ CompletedStreamElementIndex: make(map[string]map[common.ShardID]bool),
+ StreamElementIndexErrors: make(map[string]map[common.ShardID]string),
StreamElementIndexCounts: make(map[string]int),
StreamElementIndexProgress: make(map[string]int),
CompletedMeasureParts: make(map[string]map[common.ShardID]map[uint64]bool),
@@ -409,34 +409,47 @@ func (p *Progress) ClearStreamSeriesErrors(group string) {
}
// MarkStreamElementIndexCompleted marks a specific element index file of a stream as completed.
-func (p *Progress) MarkStreamElementIndexCompleted(group string) {
+func (p *Progress) MarkStreamElementIndexCompleted(group string, shardID common.ShardID) {
defer p.saveProgress()
p.mu.Lock()
defer p.mu.Unlock()
- // Mark group as completed
- p.CompletedStreamElementIndex[group] = true
+ // Initialize nested maps if they don't exist
+ if p.CompletedStreamElementIndex[group] == nil {
+ p.CompletedStreamElementIndex[group] = make(map[common.ShardID]bool)
+ }
+
+ // Mark shard as completed
+ p.CompletedStreamElementIndex[group][shardID] = true
// Update progress count
p.StreamElementIndexProgress[group]++
}
// IsStreamElementIndexCompleted checks if a specific element index file of a stream has been completed.
-func (p *Progress) IsStreamElementIndexCompleted(group string) bool {
+func (p *Progress) IsStreamElementIndexCompleted(group string, shardID common.ShardID) bool {
p.mu.Lock()
defer p.mu.Unlock()
- return p.CompletedStreamElementIndex[group]
+ if shards, ok := p.CompletedStreamElementIndex[group]; ok {
+ return shards[shardID]
+ }
+ return false
}
// MarkStreamElementIndexError records an error for a specific element index file of a stream.
-func (p *Progress) MarkStreamElementIndexError(group string, errorMsg string) {
+func (p *Progress) MarkStreamElementIndexError(group string, shardID common.ShardID, errorMsg string) {
defer p.saveProgress()
p.mu.Lock()
defer p.mu.Unlock()
+ // Initialize nested maps if they don't exist
+ if p.StreamElementIndexErrors[group] == nil {
+ p.StreamElementIndexErrors[group] = make(map[common.ShardID]string)
+ }
+
// Record the error
- p.StreamElementIndexErrors[group] = errorMsg
+ p.StreamElementIndexErrors[group][shardID] = errorMsg
}
// SetStreamElementIndexCount sets the total number of element index files for a stream.
@@ -559,7 +572,7 @@ func (p *Progress) ClearErrors() {
defer p.mu.Unlock()
p.StreamPartErrors = make(map[string]map[common.ShardID]map[uint64]string)
p.StreamSeriesErrors = make(map[string]map[common.ShardID]string)
- p.StreamElementIndexErrors = make(map[string]string)
+ p.StreamElementIndexErrors = make(map[string]map[common.ShardID]string)
p.MeasurePartErrors = make(map[string]map[common.ShardID]map[uint64]string)
p.MeasureSeriesErrors = make(map[string]map[common.ShardID]string)
}
diff --git a/banyand/backup/lifecycle/stream_migration_visitor.go b/banyand/backup/lifecycle/stream_migration_visitor.go
index 47505e47..181b928e 100644
--- a/banyand/backup/lifecycle/stream_migration_visitor.go
+++ b/banyand/backup/lifecycle/stream_migration_visitor.go
@@ -255,9 +255,10 @@ func (mv *streamMigrationVisitor) VisitPart(_ *timestamp.TimeRange, sourceShardI
// VisitElementIndex implements stream.Visitor.
func (mv *streamMigrationVisitor) VisitElementIndex(segmentTR *timestamp.TimeRange, sourceShardID common.ShardID, indexPath string) error {
- if mv.progress.IsStreamElementIndexCompleted(mv.group) {
+ if mv.progress.IsStreamElementIndexCompleted(mv.group, sourceShardID) {
mv.logger.Debug().
Str("group", mv.group).
+ Uint32("source_shard", uint32(sourceShardID)).
Msg("element index segment already completed, skipping")
return nil
}
@@ -355,12 +356,12 @@ func (mv *streamMigrationVisitor) VisitElementIndex(segmentTR *timestamp.TimeRan
// Stream segment file to target shard replicas
if err := mv.streamPartToTargetShard(partData); err != nil {
errorMsg := fmt.Sprintf("failed to stream element index to target shard: %v", err)
- mv.progress.MarkStreamElementIndexError(mv.group, errorMsg)
+ mv.progress.MarkStreamElementIndexError(mv.group, sourceShardID, errorMsg)
return fmt.Errorf("failed to stream element index to target shard: %w", err)
}
// Mark segment as completed
- mv.progress.MarkStreamElementIndexCompleted(mv.group)
+ mv.progress.MarkStreamElementIndexCompleted(mv.group, sourceShardID)
return nil
}