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/06/09 10:45:54 UTC
(skywalking-banyandb) branch lm-e2e created (now 8a586482)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch lm-e2e
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 8a586482 Add e2e test to lifecycle management
This branch includes the following new commits:
new 8a586482 Add e2e test to lifecycle management
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: Add e2e test to lifecycle management
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch lm-e2e
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 8a58648241457e01595897457012f570d236746a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Jun 9 10:28:59 2025 +0000
Add e2e test to lifecycle management
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
.github/workflows/{e2e.storage.yml => e2e.yml} | 12 +
.gitignore | 2 +-
banyand/backup/backup.go | 6 +
banyand/backup/lifecycle/lifecycle.go | 66 ++---
banyand/backup/lifecycle/progress.go | 78 ++++-
banyand/backup/lifecycle/progress_test.go | 8 +-
banyand/backup/lifecycle/service.go | 325 +++++++++++++++++----
banyand/backup/lifecycle/steps.go | 22 +-
banyand/backup/lifecycle/steps_test.go | 7 +-
banyand/backup/restore.go | 9 +-
banyand/metadata/schema/schema.go | 2 +
banyand/observability/bypass.go | 52 ++++
pkg/node/interface.go | 7 +
test/cases/lifecycle/lifecycle.go | 66 ++++-
test/docker/Dockerfile | 1 +
.../lifecycle/data-generator/docker-compose.yml | 56 ++++
test/e2e-v2/cases/lifecycle/data-generator/gen.sh | 56 ++++
test/e2e-v2/cases/lifecycle/docker-compose.yml | 98 ++-----
test/e2e-v2/cases/lifecycle/e2e.yaml | 61 ++++
.../lifecycle/expected/dependency-endpoint.yml | 54 ++++
.../lifecycle/expected/dependency-instance.yml | 42 +++
.../lifecycle/expected/dependency-services.yml | 58 ++++
.../cases/lifecycle/expected/topN-OP-endpoint.yml | 58 ++++
.../cases/lifecycle/expected/topN-OP-service.yml | 58 ++++
.../cases/lifecycle/expected/trace-detail.yml | 45 +++
.../cases/lifecycle/expected/traces-list.yml | 27 ++
test/e2e-v2/script/env | 7 +-
test/stress/classic/env | 6 +-
test/stress/classic/env.dev | 6 +-
29 files changed, 1098 insertions(+), 197 deletions(-)
diff --git a/.github/workflows/e2e.storage.yml b/.github/workflows/e2e.yml
similarity index 88%
rename from .github/workflows/e2e.storage.yml
rename to .github/workflows/e2e.yml
index e09eba27..6f395a76 100644
--- a/.github/workflows/e2e.storage.yml
+++ b/.github/workflows/e2e.yml
@@ -47,6 +47,8 @@ jobs:
config: test/e2e-v2/cases/event/banyandb/e2e.yaml
- name: BanyanDB Cluster Mode
config: test/e2e-v2/cases/cluster/e2e.yaml
+ - name: Lifecycle
+ config: test/e2e-v2/cases/lifecycle/e2e.yaml
env:
TAG: ${{ github.sha }}
steps:
@@ -94,6 +96,10 @@ jobs:
run: |
make -C test/docker build | make -C test/docker build
docker image ls
+ - name: Generate data for lifecycle
+ if: ${{ matrix.test.name == 'Lifecycle' }}
+ run: |
+ bash test/e2e-v2/cases/lifecycle/data-generator/gen.sh
- name: ${{ matrix.test.name }}
uses: apache/skywalking-infra-e2e@cf589b4a0b9f8e6f436f78e9cfd94a1ee5494180
with:
@@ -109,6 +115,12 @@ jobs:
with:
name: test-logs-${{ matrix.test.name }}
path: "${{ env.SW_INFRA_E2E_LOG_DIR }}"
+ - uses: actions/upload-artifact@v4
+ if: ${{ (failure() && matrix.test.name == 'Lifecycle') }}
+ name: Upload generated data
+ with:
+ name: lifecycle-data
+ path: "test/e2e-v2/cases/lifecycle/data-generator/tmp"
Storage:
if: (github.event_name == 'schedule' && github.repository == 'apache/skywalking-banyandb') || (github.event_name != 'schedule')
diff --git a/.gitignore b/.gitignore
index ce674e13..343b30b3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -25,7 +25,7 @@ bin
include
/build
target
-/tmp
+tmp
# Test binary, build with `go test -c`
*.test
diff --git a/banyand/backup/backup.go b/banyand/backup/backup.go
index 7aaaf57d..c77c2a56 100644
--- a/banyand/backup/backup.go
+++ b/banyand/backup/backup.go
@@ -62,6 +62,7 @@ type backupOptions struct {
// NewBackupCommand creates a new backup command.
func NewBackupCommand() *cobra.Command {
var backupOpts backupOptions
+ logging := logger.Logging{}
cmd := &cobra.Command{
Short: "Backup BanyanDB snapshots to remote storage",
DisableAutoGenTag: true,
@@ -70,6 +71,9 @@ func NewBackupCommand() *cobra.Command {
if err := config.Load("logging", cmd.Flags()); err != nil {
return err
}
+ if err := logger.Init(logging); err != nil {
+ return err
+ }
if backupOpts.schedule == "" {
return backupAction(backupOpts)
}
@@ -100,6 +104,8 @@ func NewBackupCommand() *cobra.Command {
},
}
+ cmd.Flags().StringVar(&logging.Env, "logging-env", "prod", "the logging environment")
+ cmd.Flags().StringVar(&logging.Level, "logging-level", "info", "the root level of logging")
cmd.Flags().StringVar(&backupOpts.gRPCAddr, "grpc-addr", "127.0.0.1:17912", "gRPC address of the data node")
cmd.Flags().BoolVar(&backupOpts.enableTLS, "enable-tls", false, "Enable TLS for gRPC connection")
cmd.Flags().BoolVar(&backupOpts.insecure, "insecure", false, "Skip server certificate verification")
diff --git a/banyand/backup/lifecycle/lifecycle.go b/banyand/backup/lifecycle/lifecycle.go
index c89ae929..a3bbfbc5 100644
--- a/banyand/backup/lifecycle/lifecycle.go
+++ b/banyand/backup/lifecycle/lifecycle.go
@@ -21,83 +21,53 @@ package lifecycle
import (
"context"
"os"
- "os/signal"
- "syscall"
- "time"
- "github.com/benbjohnson/clock"
- "github.com/robfig/cron/v3"
"github.com/spf13/cobra"
"github.com/apache/skywalking-banyandb/banyand/metadata"
- "github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
- "github.com/apache/skywalking-banyandb/pkg/timestamp"
+ "github.com/apache/skywalking-banyandb/pkg/signal"
"github.com/apache/skywalking-banyandb/pkg/version"
)
// NewCommand creates a new lifecycle command.
func NewCommand() *cobra.Command {
- var schedule string
-
- l := logger.GetLogger("bootstrap")
-
+ logging := logger.Logging{}
metaSvc, err := metadata.NewClient(false, false)
if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate metadata service")
+ logger.GetLogger().Err(err).Msg("failed to initiate metadata service")
}
- metricSvc := observability.NewMetricService(metaSvc, nil, "lifecycle", nil)
- svc := NewService(metaSvc, metricSvc)
+ svc := NewService(metaSvc)
group := run.NewGroup("lifecycle")
- group.Register(metaSvc, metricSvc, svc)
-
+ group.Register(new(signal.Handler), metaSvc, svc)
cmd := &cobra.Command{
Short: "Run lifecycle migration",
DisableAutoGenTag: true,
Version: version.Build(),
RunE: func(cmd *cobra.Command, _ []string) error {
- if err := config.Load("logging", cmd.Flags()); err != nil {
+ if err = config.Load("logging", cmd.Flags()); err != nil {
return err
}
- if schedule == "" {
- return group.Run(context.Background())
+ if err = logger.Init(logging); err != nil {
+ return err
}
- schedLogger := logger.GetLogger().Named("lifecycle-scheduler")
- schedLogger.Info().Msgf("lifecycle migration will run with schedule: %s", schedule)
- clockInstance := clock.New()
- sch := timestamp.NewScheduler(schedLogger, clockInstance)
- err := sch.Register("lifecycle", cron.Descriptor, schedule, func(_ time.Time, l *logger.Logger) bool {
- err := group.Run(context.Background())
- if err != nil {
- l.Error().Err(err).Msg("lifecycle migration failed")
- } else {
- l.Info().Msg("lifecycle migration succeeded")
+ defer func() {
+ if err := recover(); err != nil {
+ logger.GetLogger().Error().Msgf("panic occurred: %v", err)
+ os.Exit(-1)
}
- return true
- })
- if err != nil {
- return err
+ }()
+ if err := group.Run(context.Background()); err != nil {
+ logger.GetLogger().Error().Err(err).Stack().Str("name", group.Name()).Msg("Exit")
+ os.Exit(-1)
}
-
- sigChan := make(chan os.Signal, 1)
- signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
- schedLogger.Info().Msg("backup scheduler started, press Ctrl+C to exit")
- <-sigChan
- schedLogger.Info().Msg("shutting down lifecycle migration scheduler...")
- sch.Close()
return nil
},
}
cmd.Flags().AddFlagSet(group.RegisterFlags().FlagSet)
-
- cmd.Flags().StringVar(
- &schedule,
- "schedule",
- "",
- "Schedule expression for periodic backup. Options: @yearly, @monthly, @weekly, @daily, @hourly or @every <duration>",
- )
-
+ cmd.Flags().StringVar(&logging.Env, "logging-env", "prod", "the logging")
+ cmd.Flags().StringVar(&logging.Level, "logging-level", "info", "the root level of logging")
return cmd
}
diff --git a/banyand/backup/lifecycle/progress.go b/banyand/backup/lifecycle/progress.go
index 9a5347b9..f9b828a6 100644
--- a/banyand/backup/lifecycle/progress.go
+++ b/banyand/backup/lifecycle/progress.go
@@ -22,17 +22,37 @@ import (
"os"
"sync"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
// Progress tracks the lifecycle migration progress to support resume after crash.
type Progress struct {
- CompletedGroups map[string]bool `json:"completed_groups"`
- CompletedStreams map[string]map[string]bool `json:"completed_streams"`
- CompletedMeasures map[string]map[string]bool `json:"completed_measures"`
- DeletedStreamGroups map[string]bool `json:"deleted_stream_groups"`
- DeletedMeasureGroups map[string]bool `json:"deleted_measure_groups"`
- mu sync.Mutex `json:"-"`
+ CompletedGroups map[string]bool `json:"completed_groups"`
+ CompletedStreams map[string]map[string]bool `json:"completed_streams"`
+ CompletedMeasures map[string]map[string]bool `json:"completed_measures"`
+ DeletedStreamGroups map[string]bool `json:"deleted_stream_groups"`
+ DeletedMeasureGroups map[string]bool `json:"deleted_measure_groups"`
+ StreamErrors map[string]map[string]string `json:"stream_errors"`
+ MeasureErrors map[string]map[string]string `json:"measure_errors"`
+ StreamCounts map[string]map[string]int `json:"stream_counts"`
+ MeasureCounts map[string]map[string]int `json:"measure_counts"`
+ SnapshotStreamDir string `json:"snapshot_stream_dir"`
+ SnapshotMeasureDir string `json:"snapshot_measure_dir"`
+ mu sync.Mutex `json:"-"`
+}
+
+// AllGroupsFullyCompleted checks if all groups are fully completed.
+func (p *Progress) AllGroupsFullyCompleted(groups []*commonv1.Group) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ for _, group := range groups {
+ if !p.CompletedGroups[group.Metadata.Name] {
+ return false
+ }
+ }
+ return true
}
// NewProgress creates a new Progress tracker.
@@ -43,6 +63,10 @@ func NewProgress() *Progress {
CompletedMeasures: make(map[string]map[string]bool),
DeletedStreamGroups: make(map[string]bool),
DeletedMeasureGroups: make(map[string]bool),
+ StreamErrors: make(map[string]map[string]string),
+ MeasureErrors: make(map[string]map[string]string),
+ StreamCounts: make(map[string]map[string]int),
+ MeasureCounts: make(map[string]map[string]int),
}
}
@@ -110,13 +134,17 @@ func (p *Progress) IsGroupCompleted(group string) bool {
}
// MarkStreamCompleted marks a stream as completed.
-func (p *Progress) MarkStreamCompleted(group, stream string) {
+func (p *Progress) MarkStreamCompleted(group, stream string, count int) {
p.mu.Lock()
defer p.mu.Unlock()
if p.CompletedStreams[group] == nil {
p.CompletedStreams[group] = make(map[string]bool)
}
p.CompletedStreams[group][stream] = true
+ if p.StreamCounts[group] == nil {
+ p.StreamCounts[group] = make(map[string]int)
+ }
+ p.StreamCounts[group][stream] = count
}
// IsStreamCompleted checks if a stream has been completed.
@@ -130,13 +158,17 @@ func (p *Progress) IsStreamCompleted(group, stream string) bool {
}
// MarkMeasureCompleted marks a measure as completed.
-func (p *Progress) MarkMeasureCompleted(group, measure string) {
+func (p *Progress) MarkMeasureCompleted(group, measure string, count int) {
p.mu.Lock()
defer p.mu.Unlock()
if p.CompletedMeasures[group] == nil {
p.CompletedMeasures[group] = make(map[string]bool)
}
p.CompletedMeasures[group][measure] = true
+ if p.MeasureCounts[group] == nil {
+ p.MeasureCounts[group] = make(map[string]int)
+ }
+ p.MeasureCounts[group][measure] = count
}
// IsMeasureCompleted checks if a measure has been completed.
@@ -177,6 +209,36 @@ func (p *Progress) IsMeasureGroupDeleted(group string) bool {
return p.DeletedMeasureGroups[group]
}
+// ClearErrors resets all prior stream/measure error records.
+func (p *Progress) ClearErrors() {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.StreamErrors = make(map[string]map[string]string)
+ p.MeasureErrors = make(map[string]map[string]string)
+}
+
+// MarkStreamError records an error message for a specific stream.
+func (p *Progress) MarkStreamError(group, stream, msg string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.StreamErrors[group] == nil {
+ p.StreamErrors[group] = make(map[string]string)
+ }
+ p.StreamErrors[group][stream] = msg
+ p.CompletedStreams[group][stream] = false
+}
+
+// MarkMeasureError records an error message for a specific measure.
+func (p *Progress) MarkMeasureError(group, measure, msg string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.MeasureErrors[group] == nil {
+ p.MeasureErrors[group] = make(map[string]string)
+ }
+ p.MeasureErrors[group][measure] = msg
+ p.CompletedMeasures[group][measure] = false
+}
+
// Remove deletes the progress file.
func (p *Progress) Remove(path string, l *logger.Logger) {
if path == "" {
diff --git a/banyand/backup/lifecycle/progress_test.go b/banyand/backup/lifecycle/progress_test.go
index 46b7252f..5d3943a2 100644
--- a/banyand/backup/lifecycle/progress_test.go
+++ b/banyand/backup/lifecycle/progress_test.go
@@ -53,12 +53,12 @@ func TestProgress(t *testing.T) {
assert.True(t, progress.IsGroupCompleted("group1"))
assert.False(t, progress.IsGroupCompleted("group2"))
- progress.MarkStreamCompleted("group1", "stream1")
+ progress.MarkStreamCompleted("group1", "stream1", 1)
assert.True(t, progress.IsStreamCompleted("group1", "stream1"))
assert.False(t, progress.IsStreamCompleted("group1", "stream2"))
assert.False(t, progress.IsStreamCompleted("group2", "stream1"))
- progress.MarkMeasureCompleted("group1", "measure1")
+ progress.MarkMeasureCompleted("group1", "measure1", 1)
assert.True(t, progress.IsMeasureCompleted("group1", "measure1"))
assert.False(t, progress.IsMeasureCompleted("group1", "measure2"))
assert.False(t, progress.IsMeasureCompleted("group2", "measure1"))
@@ -75,8 +75,8 @@ func TestProgress(t *testing.T) {
t.Run("SaveAndLoad", func(t *testing.T) {
progress := NewProgress()
progress.MarkGroupCompleted("group1")
- progress.MarkStreamCompleted("group1", "stream1")
- progress.MarkMeasureCompleted("group1", "measure1")
+ progress.MarkStreamCompleted("group1", "stream1", 1)
+ progress.MarkMeasureCompleted("group1", "measure1", 1)
progress.MarkStreamGroupDeleted("group2")
progress.MarkMeasureGroupDeleted("group2")
diff --git a/banyand/backup/lifecycle/service.go b/banyand/backup/lifecycle/service.go
index 7338065d..f7e673bc 100644
--- a/banyand/backup/lifecycle/service.go
+++ b/banyand/backup/lifecycle/service.go
@@ -19,8 +19,16 @@ package lifecycle
import (
"context"
+ "encoding/json"
+ "fmt"
"math"
+ "os"
+ "path/filepath"
+ "sort"
+ "time"
+ "github.com/benbjohnson/clock"
+ "github.com/robfig/cron/v3"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/timestamppb"
@@ -35,6 +43,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/stream"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -53,24 +62,31 @@ type service interface {
var _ service = (*lifecycleService)(nil)
type lifecycleService struct {
- metadata metadata.Repo
- omr observability.MetricsRegistry
- l *logger.Logger
- gRPCAddr string
- cert string
- streamRoot string
- measureRoot string
- progressFilePath string
- enableTLS bool
- insecure bool
+ metadata metadata.Repo
+ omr observability.MetricsRegistry
+ pm protector.Memory
+ l *logger.Logger
+ sch *timestamp.Scheduler
+ measureRoot string
+ streamRoot string
+ progressFilePath string
+ reportDir string
+ schedule string
+ cert string
+ gRPCAddr string
+ maxExecutionTimes int
+ enableTLS bool
+ insecure bool
}
// NewService creates a new lifecycle service.
-func NewService(meta metadata.Repo, omr observability.MetricsRegistry) run.Unit {
- return &lifecycleService{
+func NewService(meta metadata.Repo) run.Unit {
+ ls := &lifecycleService{
metadata: meta,
- omr: omr,
+ omr: observability.BypassRegistry,
}
+ ls.pm = protector.NewMemory(ls.omr)
+ return ls
}
func (l *lifecycleService) FlagSet() *run.FlagSet {
@@ -83,6 +99,14 @@ func (l *lifecycleService) FlagSet() *run.FlagSet {
flagS.StringVar(&l.streamRoot, "stream-root-path", "/tmp", "Root directory for stream catalog")
flagS.StringVar(&l.measureRoot, "measure-root-path", "/tmp", "Root directory for measure catalog")
flagS.StringVar(&l.progressFilePath, "progress-file", "/tmp/lifecycle-progress.json", "Path to store progress for crash recovery")
+ flagS.StringVar(&l.reportDir, "report-dir", "/tmp/lifecycle-reports", "Directory to store migration reports")
+ flagS.StringVar(
+ &l.schedule,
+ "schedule",
+ "",
+ "Schedule expression for periodic backup. Options: @yearly, @monthly, @weekly, @daily, @hourly or @every <duration>",
+ )
+ flagS.IntVar(&l.maxExecutionTimes, "max-execution-times", 0, "Maximum number of times to execute the lifecycle migration. 0 means no limit.")
return flagS
}
@@ -91,6 +115,9 @@ func (l *lifecycleService) Validate() error {
}
func (l *lifecycleService) GracefulStop() {
+ if l.sch != nil {
+ l.sch.Close()
+ }
}
func (l *lifecycleService) Name() string {
@@ -99,72 +126,204 @@ func (l *lifecycleService) Name() string {
func (l *lifecycleService) Serve() run.StopNotify {
l.l = logger.GetLogger("lifecycle")
- ctx := context.Background()
done := make(chan struct{})
- close(done)
+ if l.schedule == "" {
+ defer close(done)
+ l.l.Info().Msg("starting lifecycle migration without schedule")
+ if err := l.action(); err != nil {
+ logger.Panicf("failed to run lifecycle migration: %v", err)
+ }
+ return done
+ }
+ l.l.Info().Msgf("lifecycle migration will run with schedule: %s", l.schedule)
+ clockInstance := clock.New()
+ l.sch = timestamp.NewScheduler(l.l, clockInstance)
+ var executionCount int
+ err := l.sch.Register("lifecycle", cron.Descriptor, l.schedule, func(triggerTime time.Time, _ *logger.Logger) bool {
+ l.l.Info().Msgf("lifecycle migration triggered at %s", triggerTime)
+ if err := l.action(); err != nil {
+ l.l.Error().Err(err).Msg("failed to run lifecycle migration action")
+ }
+ executionCount++
+ if l.maxExecutionTimes > 0 && executionCount >= l.maxExecutionTimes {
+ l.l.Info().Msgf("lifecycle migration reached max execution times: %d, stopping scheduler", l.maxExecutionTimes)
+ close(done)
+ return false
+ }
+ return true
+ })
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to register lifecycle migration schedule")
+ return done
+ }
+ return done
+}
+func (l *lifecycleService) action() error {
+ ctx := context.Background()
progress := LoadProgress(l.progressFilePath, l.l)
+ progress.ClearErrors()
groups, err := l.getGroupsToProcess(ctx, progress)
if err != nil {
l.l.Error().Err(err).Msg("failed to get groups to process")
- return done
+ return err
}
if len(groups) == 0 {
l.l.Info().Msg("no groups to process, all groups already completed")
progress.Remove(l.progressFilePath, l.l)
- return done
+ return err
}
- streamDir, measureDir, err := l.getSnapshots(groups)
+ // Pass progress to getSnapshots
+ streamDir, measureDir, err := l.getSnapshots(groups, progress)
if err != nil {
l.l.Error().Err(err).Msg("failed to get snapshots")
- return done
+ return err
}
+ progress.Save(l.progressFilePath, l.l)
streamSVC, measureSVC, err := l.setupQuerySvc(ctx, streamDir, measureDir)
- if err != nil {
- l.l.Error().Err(err).Msg("failed to setup query service")
- return done
- }
if streamSVC != nil {
defer streamSVC.GracefulStop()
}
if measureSVC != nil {
defer measureSVC.GracefulStop()
}
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to setup query service")
+ return err
+ }
+
nodes, err := l.metadata.NodeRegistry().ListNode(ctx, databasev1.Role_ROLE_DATA)
if err != nil {
l.l.Error().Err(err).Msg("failed to list data nodes")
- return done
+ return err
}
labels := common.ParseNodeFlags()
+ allGroupsCompleted := true
for _, g := range groups {
switch g.Catalog {
case commonv1.Catalog_CATALOG_STREAM:
if streamSVC == nil {
l.l.Error().Msgf("stream service is not available, skipping group: %s", g.Metadata.Name)
+ allGroupsCompleted = false
continue
}
l.processStreamGroup(ctx, g, streamSVC, nodes, labels, progress)
case commonv1.Catalog_CATALOG_MEASURE:
if measureSVC == nil {
l.l.Error().Msgf("measure service is not available, skipping group: %s", g.Metadata.Name)
+ allGroupsCompleted = false
continue
}
l.processMeasureGroup(ctx, g, measureSVC, nodes, labels, progress)
default:
l.l.Info().Msgf("group catalog: %s doesn't support lifecycle management", g.Catalog)
}
-
- progress.MarkGroupCompleted(g.Metadata.Name)
progress.Save(l.progressFilePath, l.l)
}
- progress.Remove(l.progressFilePath, l.l)
- l.l.Info().Msg("lifecycle migration completed successfully")
- return done
+ // Only remove progress file if ALL groups are fully completed
+ if allGroupsCompleted && progress.AllGroupsFullyCompleted(groups) {
+ progress.Remove(l.progressFilePath, l.l)
+ l.l.Info().Msg("lifecycle migration completed successfully")
+ l.generateReport(progress)
+ return nil
+ }
+ l.l.Info().Msg("lifecycle migration partially completed, progress file retained")
+ return fmt.Errorf("lifecycle migration partially completed, progress file retained; %d groups not fully completed", len(groups)-len(progress.CompletedGroups))
+}
+
+// generateReport gathers counts & errors from Progress, writes one JSON file per run, and keeps only 5 latest.
+func (l *lifecycleService) generateReport(p *Progress) {
+ type grp struct {
+ StreamCounts map[string]int `json:"stream_counts"`
+ MeasureCounts map[string]int `json:"measure_counts"`
+ StreamErrors map[string]string `json:"stream_errors"`
+ MeasureErrors map[string]string `json:"measure_errors"`
+ Name string `json:"name"`
+ }
+ reportDir := l.reportDir
+ if err := os.MkdirAll(reportDir, 0o755); err != nil {
+ l.l.Error().Err(err).Msgf("failed to create report dir %s", reportDir)
+ return
+ }
+
+ // build report
+ var groups []grp
+ for name := range p.CompletedStreams {
+ groups = append(groups, grp{
+ Name: name,
+ StreamCounts: p.StreamCounts[name],
+ MeasureCounts: p.MeasureCounts[name],
+ StreamErrors: p.StreamErrors[name],
+ MeasureErrors: p.MeasureErrors[name],
+ })
+ }
+ // also include groups that had only measures
+ for name := range p.CompletedMeasures {
+ if _, seen := p.CompletedStreams[name]; seen {
+ continue
+ }
+ groups = append(groups, grp{
+ Name: name,
+ StreamCounts: p.StreamCounts[name],
+ MeasureCounts: p.MeasureCounts[name],
+ StreamErrors: p.StreamErrors[name],
+ MeasureErrors: p.MeasureErrors[name],
+ })
+ }
+ payload := struct {
+ GeneratedAt time.Time `json:"generated_at"`
+ Groups []grp `json:"groups"`
+ }{
+ GeneratedAt: time.Now(),
+ Groups: groups,
+ }
+
+ // write file
+ fname := time.Now().Format("20060102_150405") + ".json"
+ fpath := filepath.Join(reportDir, fname)
+ data, err := json.MarshalIndent(payload, "", " ")
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to marshal migration report")
+ return
+ }
+ if err = os.WriteFile(fpath, data, 0o600); err != nil {
+ l.l.Error().Err(err).Msgf("failed to write report file %s", fpath)
+ return
+ }
+ l.l.Info().Msgf("wrote migration report %s", fpath)
+
+ // rotate: keep only 5 latest
+ entries, err := os.ReadDir(reportDir)
+ if err != nil {
+ return
+ }
+ type fi struct {
+ t time.Time
+ name string
+ }
+ var list []fi
+ for _, e := range entries {
+ if e.IsDir() {
+ continue
+ }
+ info, err := e.Info()
+ if err != nil {
+ continue
+ }
+ list = append(list, fi{name: e.Name(), t: info.ModTime()})
+ }
+ sort.Slice(list, func(i, j int) bool {
+ return list[i].t.After(list[j].t)
+ })
+ for i := 5; i < len(list); i++ {
+ _ = os.Remove(filepath.Join(reportDir, list[i].name))
+ }
+ l.l.Info().Msg("rotated old migration reports")
}
func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Progress) ([]*commonv1.Group, error) {
@@ -183,7 +342,7 @@ func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Pro
continue
}
if progress.IsGroupCompleted(g.Metadata.Name) {
- l.l.Info().Msgf("skipping already completed group: %s", g.Metadata.Name)
+ l.l.Debug().Msgf("skipping already completed group: %s", g.Metadata.Name)
continue
}
groups = append(groups, g)
@@ -195,7 +354,7 @@ func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Pro
func (l *lifecycleService) processStreamGroup(ctx context.Context, g *commonv1.Group, streamSVC stream.Service,
nodes []*databasev1.Node, labels map[string]string, progress *Progress,
) {
- shardNum, replicas, selector, client, err := parseGroup(ctx, g, labels, nodes, l.l, l.metadata)
+ shardNum, replicas, selector, client, err := parseGroup(g, labels, nodes, l.l, l.metadata)
if err != nil {
l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
return
@@ -209,26 +368,59 @@ func (l *lifecycleService) processStreamGroup(ctx context.Context, g *commonv1.G
}
tr := streamSVC.GetRemovalSegmentsTimeRange(g.Metadata.Name)
+ if tr.Start.IsZero() && tr.End.IsZero() {
+ l.l.Info().Msgf("no removal segments time range for group %s, skipping stream migration", g.Metadata.Name)
+ progress.MarkGroupCompleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ return
+ }
l.processStreams(ctx, g, ss, streamSVC, tr, shardNum, replicas, selector, client, progress)
- l.deleteExpiredStreamSegments(ctx, g, tr, progress)
+ allStreamsDone := true
+ for _, s := range ss {
+ if !progress.IsStreamCompleted(g.Metadata.Name, s.Metadata.Name) {
+ allStreamsDone = false
+ break
+ }
+ }
+ if allStreamsDone {
+ l.deleteExpiredStreamSegments(ctx, g, tr, progress)
+ progress.MarkGroupCompleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ } else {
+ l.l.Info().Msgf("skipping delete expired stream segments for group %s: some streams not fully migrated", g.Metadata.Name)
+ }
}
-func (l *lifecycleService) processStreams(ctx context.Context, g *commonv1.Group, streams []*databasev1.Stream,
- streamSVC stream.Service, tr *timestamp.TimeRange, shardNum uint32, replicas uint32, selector node.Selector, client queue.Client, progress *Progress,
+func (l *lifecycleService) processStreams(
+ ctx context.Context,
+ g *commonv1.Group,
+ streams []*databasev1.Stream,
+ streamSVC stream.Service,
+ tr *timestamp.TimeRange,
+ shardNum uint32,
+ replicas uint32,
+ selector node.Selector,
+ client queue.Client,
+ progress *Progress,
) {
for _, s := range streams {
if progress.IsStreamCompleted(g.Metadata.Name, s.Metadata.Name) {
- l.l.Info().Msgf("skipping already completed stream: %s/%s", g.Metadata.Name, s.Metadata.Name)
+ l.l.Debug().Msgf("skipping already completed stream: %s/%s", g.Metadata.Name, s.Metadata.Name)
continue
}
-
- if sum, err := l.processSingleStream(ctx, s, streamSVC, tr, shardNum, replicas, selector, client); err == nil {
- l.l.Info().Msgf("migrated %d elements in stream %s", sum, s.Metadata.Name)
+ sum, err := l.processSingleStream(ctx, s, streamSVC, tr, shardNum, replicas, selector, client)
+ if err != nil {
+ progress.MarkStreamError(g.Metadata.Name, s.Metadata.Name, err.Error())
+ } else {
+ if sum < 1 {
+ l.l.Debug().Msgf("no elements migrated for stream %s", s.Metadata.Name)
+ } else {
+ l.l.Info().Msgf("migrated %d elements in stream %s", sum, s.Metadata.Name)
+ }
+ progress.MarkStreamCompleted(g.Metadata.Name, s.Metadata.Name, sum)
}
-
- progress.MarkStreamCompleted(g.Metadata.Name, s.Metadata.Name)
progress.Save(l.progressFilePath, l.l)
}
}
@@ -300,7 +492,7 @@ func (l *lifecycleService) deleteExpiredStreamSegments(ctx context.Context, g *c
func (l *lifecycleService) processMeasureGroup(ctx context.Context, g *commonv1.Group, measureSVC measure.Service,
nodes []*databasev1.Node, labels map[string]string, progress *Progress,
) {
- shardNum, replicas, selector, client, err := parseGroup(ctx, g, labels, nodes, l.l, l.metadata)
+ shardNum, replicas, selector, client, err := parseGroup(g, labels, nodes, l.l, l.metadata)
if err != nil {
l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
return
@@ -314,26 +506,59 @@ func (l *lifecycleService) processMeasureGroup(ctx context.Context, g *commonv1.
}
tr := measureSVC.GetRemovalSegmentsTimeRange(g.Metadata.Name)
+ if tr.Start.IsZero() && tr.End.IsZero() {
+ l.l.Info().Msgf("no removal segments time range for group %s, skipping measure migration", g.Metadata.Name)
+ progress.MarkGroupCompleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ return
+ }
l.processMeasures(ctx, g, mm, measureSVC, tr, shardNum, replicas, selector, client, progress)
- l.deleteExpiredMeasureSegments(ctx, g, tr, progress)
+ allMeasuresDone := true
+ for _, m := range mm {
+ if !progress.IsMeasureCompleted(g.Metadata.Name, m.Metadata.Name) {
+ allMeasuresDone = false
+ break
+ }
+ }
+ if allMeasuresDone {
+ l.deleteExpiredMeasureSegments(ctx, g, tr, progress)
+ progress.MarkGroupCompleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ } else {
+ l.l.Info().Msgf("skipping delete expired measure segments for group %s: some measures not fully migrated", g.Metadata.Name)
+ }
}
-func (l *lifecycleService) processMeasures(ctx context.Context, g *commonv1.Group, measures []*databasev1.Measure,
- measureSVC measure.Service, tr *timestamp.TimeRange, shardNum uint32, replicas uint32, selector node.Selector, client queue.Client, progress *Progress,
+func (l *lifecycleService) processMeasures(
+ ctx context.Context,
+ g *commonv1.Group,
+ measures []*databasev1.Measure,
+ measureSVC measure.Service,
+ tr *timestamp.TimeRange,
+ shardNum uint32,
+ replicas uint32,
+ selector node.Selector,
+ client queue.Client,
+ progress *Progress,
) {
for _, m := range measures {
if progress.IsMeasureCompleted(g.Metadata.Name, m.Metadata.Name) {
- l.l.Info().Msgf("skipping already completed measure: %s/%s", g.Metadata.Name, m.Metadata.Name)
+ l.l.Debug().Msgf("skipping already completed measure: %s/%s", g.Metadata.Name, m.Metadata.Name)
continue
}
-
- if sum, err := l.processSingleMeasure(ctx, m, measureSVC, tr, shardNum, replicas, selector, client); err == nil {
- l.l.Info().Msgf("migrated %d elements in measure %s", sum, m.Metadata.Name)
+ sum, err := l.processSingleMeasure(ctx, m, measureSVC, tr, shardNum, replicas, selector, client)
+ if err != nil {
+ progress.MarkMeasureError(g.Metadata.Name, m.Metadata.Name, err.Error())
+ } else {
+ if sum < 1 {
+ l.l.Debug().Msgf("no elements migrated for measure %s", m.Metadata.Name)
+ } else {
+ l.l.Info().Msgf("migrated %d elements in measure %s", sum, m.Metadata.Name)
+ }
+ progress.MarkMeasureCompleted(g.Metadata.Name, m.Metadata.Name, sum)
}
-
- progress.MarkMeasureCompleted(g.Metadata.Name, m.Metadata.Name)
progress.Save(l.progressFilePath, l.l)
}
}
diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go
index 53157415..e94f12d3 100644
--- a/banyand/backup/lifecycle/steps.go
+++ b/banyand/backup/lifecycle/steps.go
@@ -38,7 +38,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/measure"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
- "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/queue/pub"
"github.com/apache/skywalking-banyandb/banyand/stream"
@@ -50,7 +49,12 @@ import (
"github.com/apache/skywalking-banyandb/pkg/query/model"
)
-func (l *lifecycleService) getSnapshots(groups []*commonv1.Group) (streamDir string, measureDir string, err error) {
+func (l *lifecycleService) getSnapshots(groups []*commonv1.Group, p *Progress) (streamDir string, measureDir string, err error) {
+ // If we already have snapshot dirs in Progress, reuse them
+ if p.SnapshotStreamDir != "" || p.SnapshotMeasureDir != "" {
+ return p.SnapshotStreamDir, p.SnapshotMeasureDir, nil
+ }
+
snapshotGroups := make([]*databasev1.SnapshotRequest_Group, 0, len(groups))
for _, group := range groups {
snapshotGroups = append(snapshotGroups, &databasev1.SnapshotRequest_Group{
@@ -75,14 +79,16 @@ func (l *lifecycleService) getSnapshots(groups []*commonv1.Group) (streamDir str
measureDir = snapshotDir
}
}
+ // Save the new snapshot paths into Progress
+ p.SnapshotStreamDir = streamDir
+ p.SnapshotMeasureDir = measureDir
return streamDir, measureDir, nil
}
func (l *lifecycleService) setupQuerySvc(ctx context.Context, streamDir, measureDir string) (streamSVC stream.Service, measureSVC measure.Service, err error) {
- pm := protector.NewMemory(l.omr)
ctx = context.WithValue(ctx, common.ContextNodeKey, common.Node{})
if streamDir != "" {
- streamSVC, err = stream.NewReadonlyService(l.metadata, l.omr, pm)
+ streamSVC, err = stream.NewReadonlyService(l.metadata, l.omr, l.pm)
if err != nil {
return nil, nil, err
}
@@ -94,7 +100,7 @@ func (l *lifecycleService) setupQuerySvc(ctx context.Context, streamDir, measure
}
}
if measureDir != "" {
- measureSVC, err = measure.NewReadonlyService(l.metadata, l.omr, pm)
+ measureSVC, err = measure.NewReadonlyService(l.metadata, l.omr, l.pm)
if err != nil {
return nil, nil, err
}
@@ -108,7 +114,7 @@ func (l *lifecycleService) setupQuerySvc(ctx context.Context, streamDir, measure
return streamSVC, measureSVC, nil
}
-func parseGroup(ctx context.Context, g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node,
+func parseGroup(g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node,
l *logger.Logger, metadata metadata.Repo,
) (uint32, uint32, node.Selector, queue.Client, error) {
ro := g.ResourceOpts
@@ -143,8 +149,8 @@ func parseGroup(ctx context.Context, g *commonv1.Group, nodeLabels map[string]st
return 0, 0, nil, nil, errors.WithMessagef(err, "failed to parse node selector %s", nst.NodeSelector)
}
nodeSel := node.NewRoundRobinSelector("", metadata)
- if err = nodeSel.PreRun(ctx); err != nil {
- return 0, 0, nil, nil, errors.WithMessage(err, "failed to run node selector")
+ if ok, _ := nodeSel.OnInit([]schema.Kind{schema.KindGroup}); !ok {
+ return 0, 0, nil, nil, fmt.Errorf("failed to initialize node selector for group %s", g.Metadata.Name)
}
client := pub.NewWithoutMetadata()
if g.Catalog == commonv1.Catalog_CATALOG_STREAM {
diff --git a/banyand/backup/lifecycle/steps_test.go b/banyand/backup/lifecycle/steps_test.go
index 9c3b2c7f..95d6d1ce 100644
--- a/banyand/backup/lifecycle/steps_test.go
+++ b/banyand/backup/lifecycle/steps_test.go
@@ -591,9 +591,14 @@ func TestParseGroup(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
+ mockGroup := schema.NewMockGroup(ctrl)
+ mockGroup.EXPECT().ListGroup(gomock.Any()).Return([]*commonv1.Group{}, nil).AnyTimes()
+
mockRepo := metadata.NewMockRepo(ctrl)
mockRepo.EXPECT().RegisterHandler("", schema.KindGroup, gomock.Any()).MaxTimes(1)
- shardNum, replicas, selector, client, err := parseGroup(context.Background(), tt.group, tt.nodeLabels, tt.nodes, l, mockRepo)
+
+ mockRepo.EXPECT().GroupRegistry().Return(mockGroup).AnyTimes()
+ shardNum, replicas, selector, client, err := parseGroup(tt.group, tt.nodeLabels, tt.nodes, l, mockRepo)
if tt.expectError {
require.Error(t, err)
diff --git a/banyand/backup/restore.go b/banyand/backup/restore.go
index 7c9151e7..da7e42c8 100644
--- a/banyand/backup/restore.go
+++ b/banyand/backup/restore.go
@@ -40,14 +40,21 @@ import (
// NewRestoreCommand creates a new restore command.
func NewRestoreCommand() *cobra.Command {
+ logging := logger.Logging{}
rootCmd := &cobra.Command{
DisableAutoGenTag: true,
Version: version.Build(),
Short: "Restore BanyanDB data from remote storage",
PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
- return config.Load("logging", cmd.Flags())
+ if err := config.Load("logging", cmd.Flags()); err != nil {
+ return err
+ }
+
+ return logger.Init(logging)
},
}
+ rootCmd.Flags().StringVar(&logging.Env, "logging-env", "prod", "the logging environment")
+ rootCmd.Flags().StringVar(&logging.Level, "logging-level", "info", "the root level of logging")
rootCmd.AddCommand(newRunCommand())
rootCmd.AddCommand(NewTimeDirCommand())
return rootCmd
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index d103536f..6b98c05a 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -181,6 +181,8 @@ type Measure interface {
}
// Group allows CRUD groups which is namespaces of resources.
+//
+//go:generate mockgen -destination=group_mock.go -package=schema github.com/apache/skywalking-banyandb/banyand/metadata/schema Group
type Group interface {
GetGroup(ctx context.Context, group string) (*commonv1.Group, error)
ListGroup(ctx context.Context) ([]*commonv1.Group, error)
diff --git a/banyand/observability/bypass.go b/banyand/observability/bypass.go
new file mode 100644
index 00000000..677836be
--- /dev/null
+++ b/banyand/observability/bypass.go
@@ -0,0 +1,52 @@
+// 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 observability
+
+import (
+ "github.com/apache/skywalking-banyandb/pkg/meter"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+)
+
+// BypassRegistry is a MetricsRegistry that does not collect any metrics.
+var BypassRegistry = NewBypassRegistry()
+
+type bypassRegistry struct{}
+
+// NewBypassRegistry creates a new instance of bypassMetricsRegistry.
+func NewBypassRegistry() MetricsRegistry {
+ return &bypassRegistry{}
+}
+
+func (b *bypassRegistry) Name() string {
+ return "bypass-metrics-registry"
+}
+
+func (b *bypassRegistry) Serve() run.StopNotify {
+ return nil
+}
+
+func (b *bypassRegistry) GracefulStop() {
+}
+
+func (b *bypassRegistry) With(_ meter.Scope) *Factory {
+ return &Factory{}
+}
+
+func (b *bypassRegistry) NativeEnabled() bool {
+ return false
+}
diff --git a/pkg/node/interface.go b/pkg/node/interface.go
index c0e91490..bb7e7a77 100644
--- a/pkg/node/interface.go
+++ b/pkg/node/interface.go
@@ -27,6 +27,7 @@ import (
"golang.org/x/exp/slices"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/queue/pub"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -42,6 +43,7 @@ var (
//
//go:generate mockgen -destination=mock/node_selector_mock.go -package=mock github.com/apache/skywalking-banyandb/pkg/node Selector
type Selector interface {
+ OnInit(kinds []schema.Kind) (bool, []int64)
AddNode(node *databasev1.Node)
RemoveNode(node *databasev1.Node)
SetNodeSelector(selector *pub.LabelSelector)
@@ -64,6 +66,11 @@ type pickFirstSelector struct {
mu sync.RWMutex
}
+// OnInit implements Selector.
+func (p *pickFirstSelector) OnInit(_ []schema.Kind) (bool, []int64) {
+ return true, []int64{0}
+}
+
// SetNodeSelector implements Selector.
func (p *pickFirstSelector) SetNodeSelector(_ *pub.LabelSelector) {}
diff --git a/test/cases/lifecycle/lifecycle.go b/test/cases/lifecycle/lifecycle.go
index b1ebbfc3..f831e252 100644
--- a/test/cases/lifecycle/lifecycle.go
+++ b/test/cases/lifecycle/lifecycle.go
@@ -42,11 +42,12 @@ import (
var SharedContext helpers.LifecycleSharedContext
var _ = ginkgo.Describe("Lifecycle", func() {
- ginkgo.It("should migrate data correctly", func() {
+ ginkgo.It("should migrate data once correctly", func() {
dir, err := os.MkdirTemp("", "lifecycle-restore-dest")
gomega.Expect(err).NotTo(gomega.HaveOccurred())
defer os.RemoveAll(dir)
pf := filepath.Join(dir, "progress.json")
+ rf := filepath.Join(dir, "report")
lifecycleCmd := lifecycle.NewCommand()
lifecycleCmd.SetArgs([]string{
"--grpc-addr", SharedContext.DataAddr,
@@ -54,11 +55,74 @@ var _ = ginkgo.Describe("Lifecycle", func() {
"--measure-root-path", SharedContext.SrcDir,
"--etcd-endpoints", SharedContext.EtcdAddr,
"--progress-file", pf,
+ "--report-dir", rf,
})
err = lifecycleCmd.Execute()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
verifySourceDirectoriesAfterMigration()
verifyDestinationDirectoriesAfterMigration()
+ // Check report directory has files
+ rEntries, err := os.ReadDir(rf)
+ gomega.Expect(err).NotTo(gomega.HaveOccurred(), "Report directory should exist")
+ gomega.Expect(len(rEntries)).To(gomega.BeNumerically(">", 0), "Report directory should contain files")
+ conn, err := grpchelper.Conn(SharedContext.LiaisonAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ defer func() {
+ if conn != nil {
+ _ = conn.Close()
+ }
+ }()
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ sc := helpers.SharedContext{
+ Connection: conn,
+ BaseTime: SharedContext.BaseTime,
+ }
+ // Verify measure data lifecycle stages
+ verifyLifecycleStages(sc, measureTestData.VerifyFn, helpers.Args{
+ Input: "all",
+ Duration: 25 * time.Minute,
+ Offset: -20 * time.Minute,
+ })
+
+ // Verify stream data lifecycle stages
+ verifyLifecycleStages(sc, streamTestData.VerifyFn, helpers.Args{
+ Input: "all",
+ Duration: time.Hour,
+ IgnoreElementID: true,
+ })
+
+ // Verify topN data lifecycle stages
+ verifyLifecycleStages(sc, topNTestData.VerifyFn, helpers.Args{
+ Input: "aggr_desc",
+ Duration: 25 * time.Minute,
+ Offset: -20 * time.Minute,
+ })
+ })
+ ginkgo.It("should migrate data correctly with a scheduler", func() {
+ dir, err := os.MkdirTemp("", "lifecycle-restore-dest")
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ defer os.RemoveAll(dir)
+ pf := filepath.Join(dir, "progress.json")
+ rf := filepath.Join(dir, "report")
+ lifecycleCmd := lifecycle.NewCommand()
+ lifecycleCmd.SetArgs([]string{
+ "--grpc-addr", SharedContext.DataAddr,
+ "--stream-root-path", SharedContext.SrcDir,
+ "--measure-root-path", SharedContext.SrcDir,
+ "--etcd-endpoints", SharedContext.EtcdAddr,
+ "--progress-file", pf,
+ "--report-dir", rf,
+ "--schedule", "@every 5s",
+ "--max-execution-times", "2",
+ })
+ err = lifecycleCmd.Execute()
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ verifySourceDirectoriesAfterMigration()
+ verifyDestinationDirectoriesAfterMigration()
+ // Check report directory has files
+ rEntries, err := os.ReadDir(rf)
+ gomega.Expect(err).NotTo(gomega.HaveOccurred(), "Report directory should exist")
+ gomega.Expect(len(rEntries)).To(gomega.BeNumerically(">", 0), "Report directory should contain files")
conn, err := grpchelper.Conn(SharedContext.LiaisonAddr, 10*time.Second,
grpc.WithTransportCredentials(insecure.NewCredentials()))
defer func() {
diff --git a/test/docker/Dockerfile b/test/docker/Dockerfile
index a35c463b..4e492872 100644
--- a/test/docker/Dockerfile
+++ b/test/docker/Dockerfile
@@ -24,6 +24,7 @@ ARG TARGETARCH
COPY banyand/build/bin/linux/${TARGETARCH}/banyand-server-static /banyand
COPY bydbctl/build/bin/linux/${TARGETARCH}/bydbctl-cli-static /bydbctl
+COPY banyand/build/bin/linux/${TARGETARCH}/banyand-lifecycle-static /lifecycle
COPY --from=certs /etc/ssl/certs /etc/ssl/certs
ENV GRPC_GO_LOG_SEVERITY_LEVEL=WARN
diff --git a/test/e2e-v2/cases/lifecycle/data-generator/docker-compose.yml b/test/e2e-v2/cases/lifecycle/data-generator/docker-compose.yml
new file mode 100644
index 00000000..b7e1a1c1
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/data-generator/docker-compose.yml
@@ -0,0 +1,56 @@
+# Licensed to the 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.
+# The 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.
+
+services:
+ banyandb:
+ extends:
+ file: ../../../script/docker-compose/base-compose.yml
+ service: banyandb
+ ports:
+ - 17912:17912
+ - 17913:17913
+ command: standalone --stream-root-path /tmp/stream-data --measure-root-path /tmp/measure-data --measure-metadata-cache-wait-duration 1m --stream-metadata-cache-wait-duration 1m
+ networks:
+ - e2e
+
+ oap:
+ extends:
+ file: ../../../script/docker-compose/base-compose.yml
+ service: oap
+ environment:
+ SW_STORAGE: banyandb
+ SW_STORAGE_BANYANDB_RECORDS_TTL_DAYS: 5
+ SW_STORAGE_BANYANDB_TRACE_TTL_DAYS: 5
+ ports:
+ - 12800
+ networks:
+ - e2e
+ depends_on:
+ banyandb:
+ condition: service_healthy
+
+ trace-mocker:
+ image: "ghcr.io/apache/skywalking/trace-mocker:${SW_TRACE_MOCKER_COMMIT}"
+ environment:
+ OAP_HOST: oap
+ MINUS_DAYS: 3
+ networks:
+ - e2e
+ depends_on:
+ oap:
+ condition: service_healthy
+
+networks:
+ e2e:
diff --git a/test/e2e-v2/cases/lifecycle/data-generator/gen.sh b/test/e2e-v2/cases/lifecycle/data-generator/gen.sh
new file mode 100644
index 00000000..826ba6a9
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/data-generator/gen.sh
@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+# 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.
+
+
+# determine script directory and relevant paths
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ENV_FILE="$DIR/../../../script/env"
+COMPOSE_FILE="$DIR/docker-compose.yml"
+
+# load environment
+export $(grep -v '^#' "$ENV_FILE" | xargs)
+
+# start services
+docker compose -f "$COMPOSE_FILE" up -d
+CONTAINER_ID=$(docker compose -f "$COMPOSE_FILE" ps -q banyandb)
+
+echo "⌛ monitoring segment files..."
+found=false
+for i in {1..60}; do
+ if docker exec "$CONTAINER_ID" sh -c '[ -n "$(ls /tmp/measure-data/measure/data/metricsDay/seg* 2>/dev/null)" ]'; then
+ echo "✅ found segment files"
+ sleep 180
+ # copy out data
+ docker cp "$CONTAINER_ID":/tmp "$DIR/tmp"
+ docker cp "$CONTAINER_ID":/tmp/measure-data/measure/data/metadata "$DIR/tmp"
+ found=true
+ break
+ else
+ echo "⏳ didn't find segment files (retry $i/60)"
+ sleep 10
+ fi
+done
+
+if $found; then
+ echo "✅ segment files copied to $DIR"
+else
+ echo "❌ segment files not found"
+ exit 1
+fi
+
+docker compose -f "$COMPOSE_FILE" down -v
\ No newline at end of file
diff --git a/test/e2e-v2/cases/lifecycle/docker-compose.yml b/test/e2e-v2/cases/lifecycle/docker-compose.yml
index fa776002..ea3db3ac 100644
--- a/test/e2e-v2/cases/lifecycle/docker-compose.yml
+++ b/test/e2e-v2/cases/lifecycle/docker-compose.yml
@@ -13,8 +13,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-version: '2.1'
-
services:
etcd:
extends:
@@ -30,9 +28,7 @@ services:
hostname: data-hot1
command: data --etcd-endpoints=http://etcd:2379 --node-labels type=hot
volumes:
- - /tmp/measure/data-hot1:/tmp/measure
- - /tmp/stream/data-hot1:/tmp/stream
- - /tmp/property/data-hot1:/tmp/property
+ - ./data-generator/tmp/metadata:/tmp/measure/data/metadata
networks:
- e2e
@@ -43,35 +39,9 @@ services:
hostname: data-hot2
command: data --etcd-endpoints=http://etcd:2379 --node-labels type=hot
volumes:
- - /tmp/measure/data-hot2:/tmp/measure
- - /tmp/stream/data-hot2:/tmp/stream
- - /tmp/property/data-hot2:/tmp/property
- networks:
- - e2e
-
- data-warm1:
- extends:
- file: ../../script/docker-compose/base-compose.yml
- service: data
- hostname: data-warm1
- command: data --etcd-endpoints=http://etcd:2379 --node-labels type=warm
- volumes:
- - /tmp/measure/data-warm1:/tmp/measure
- - /tmp/stream/data-warm1:/tmp/stream
- - /tmp/property/data-warm1:/tmp/property
- networks:
- - e2e
-
- data-warm2:
- extends:
- file: ../../script/docker-compose/base-compose.yml
- service: data
- hostname: data-warm2
- command: data --etcd-endpoints=http://etcd:2379 --node-labels type=warm
- volumes:
- - /tmp/measure/data-warm2:/tmp/measure
- - /tmp/stream/data-warm2:/tmp/stream
- - /tmp/property/data-warm2:/tmp/property
+ - ./data-generator/tmp/measure-data/measure:/tmp/measure
+ - ./data-generator/tmp/stream-data/stream:/tmp/stream
+ - ./data-generator/tmp/property/data:/tmp/property
networks:
- e2e
@@ -103,48 +73,44 @@ services:
environment:
SW_STORAGE: banyandb
SW_STORAGE_BANYANDB_TARGETS: "liaison:17912"
- SW_STORAGE_BANYANDB_GM_MINUTE_ENABLE_WARM_STAGE: true
- SW_STORAGE_BANYANDB_GM_MINUTE_ENABLE_COLD_STAGE: true
+ SW_STORAGE_BANYANDB_METRICS_MINUTE_ENABLE_WARM_STAGE: "false"
+ SW_STORAGE_BANYANDB_METRICS_MINUTE_ENABLE_COLD_STAGE: "true"
+ SW_STORAGE_BANYANDB_METRICS_HOUR_ENABLE_WARM_STAGE: "false"
+ SW_STORAGE_BANYANDB_METRICS_HOUR_ENABLE_COLD_STAGE: "true"
+ SW_STORAGE_BANYANDB_METRICS_DAY_ENABLE_WARM_STAGE: "false"
+ SW_STORAGE_BANYANDB_METRICS_DAY_ENABLE_COLD_STAGE: "true"
+ SW_STORAGE_BANYANDB_RECORDS_ENABLE_WARM_STAGE: "false"
+ SW_STORAGE_BANYANDB_RECORDS_ENABLE_COLD_STAGE: "true"
+ SW_STORAGE_BANYANDB_TRACE_ENABLE_WARM_STAGE: "false"
+ SW_STORAGE_BANYANDB_TRACE_ENABLE_COLD_STAGE: "true"
+ SW_STORAGE_BANYANDB_TRACE_TTL_DAYS: "1"
+ SW_STORAGE_BANYANDB_RECORDS_TTL_DAYS: "1"
+ SW_STORAGE_BANYANDB_METRICS_MINUTE_TTL_DAYS: "1"
+ SW_STORAGE_BANYANDB_METRICS_HOUR_TTL_DAYS: "1"
+ SW_STORAGE_BANYANDB_METRICS_HOUR_SI_DAYS: "1"
+ SW_STORAGE_BANYANDB_METRICS_DAY_TTL_DAYS: "1"
+ SW_STORAGE_BANYANDB_METRICS_DAY_SI_DAYS: "1"
ports:
- 12800
- depends_on:
- liaison:
- condition: service_healthy
-
- agent:
- extends:
- file: ../../script/docker-compose/base-compose.yml
- service: agent
- networks:
- - e2e
-
- provider:
- extends:
- file: ../../script/docker-compose/base-compose.yml
- service: provider
- ports:
- - 9090
networks:
- e2e
depends_on:
- oap:
+ liaison:
condition: service_healthy
- agent:
- condition: service_completed_successfully
- consumer:
- extends:
- file: ../../script/docker-compose/base-compose.yml
- service: consumer
- ports:
- - 9092
+ data-hot1-lifecycle-agent:
+ image: "apache/skywalking-banyandb:${TAG}-testing"
+ network_mode: service:data-hot2
+ volumes_from:
+ - data-hot2
+ entrypoint: ["/lifecycle", "--node-labels", "type=hot", "--etcd-endpoints=http://etcd:2379"]
+ volumes:
+ - /tmp/lifecycle:/tmp
depends_on:
- oap:
+ data-hot2:
condition: service_healthy
- provider:
+ oap:
condition: service_healthy
- agent:
- condition: service_completed_successfully
networks:
e2e:
diff --git a/test/e2e-v2/cases/lifecycle/e2e.yaml b/test/e2e-v2/cases/lifecycle/e2e.yaml
new file mode 100644
index 00000000..c0f2f41a
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/e2e.yaml
@@ -0,0 +1,61 @@
+# Licensed to the 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.
+# The 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.
+
+# This file is used to show how to write configuration files and can be used to test.
+
+setup:
+ env: compose
+ file: docker-compose.yml
+ timeout: 20m
+ init-system-environment: ../../script/env
+ steps:
+ - name: set PATH
+ command: export PATH=/tmp/skywalking-infra-e2e/bin:$PATH
+ - name: install yq
+ command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh yq
+ - name: install swctl
+ command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh swctl
+
+verify:
+ # verify with retry strategy
+ retry:
+ # max retry count
+ count: 3
+ # the interval between two retries, in millisecond.
+ interval: 10s
+ cases:
+ # cold data test
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics exec --expression=service_resp_time --service-name=mock_a_service --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ../storage/expected/metrics-has-value.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql dependency service --service-name=mock_a_service --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ./expected/dependency-services.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql dependency instance --service-name=mock_a_service --dest-service-name=mock_b_service --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ./expected/dependency-instance.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql dependency endpoint --service-name=mock_a_service --endpoint-name=/dubbox-case/case/dubbox-rest/404-test --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ./expected/dependency-endpoint.yml
+ # trace segment list
+ # - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --tags http.method=get --start="-96h" --end="-48h" --cold=true
+ # expected: ./expected/traces-list.yml
+ # - query: |
+ # swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --trace-id=$( \
+ # swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --tags http.method=get --start="-96h" --end="-48h" --cold=true \
+ # | yq e '.traces | select(.[].endpointnames[0]=="/dubbox-case/case/dubbox-rest/404-test") | .[0].traceids[0]' -
+ # ) --start="-96h" --end="-48h" --cold=true
+ # expected: ./expected/trace-detail.yml
+ ## topN
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics exec --expression="top_n(service_resp_time,3,des,attr0='GENERAL')" --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ./expected/topN-OP-service.yml
+ - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics exec --expression="top_n(endpoint_resp_time,3,des)" --start="-96h" --end="-48h" --cold=true --step=DAY
+ expected: ./expected/topN-OP-endpoint.yml
\ No newline at end of file
diff --git a/test/e2e-v2/cases/lifecycle/expected/dependency-endpoint.yml b/test/e2e-v2/cases/lifecycle/expected/dependency-endpoint.yml
new file mode 100644
index 00000000..d381b073
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/dependency-endpoint.yml
@@ -0,0 +1,54 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+nodes:
+{{- contains .nodes }}
+- id: {{ b64enc "mock_a_service" }}.1_{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+ name: /dubbox-case/case/dubbox-rest/404-test
+ serviceid: {{ b64enc "mock_a_service" }}.1
+ servicename: mock_a_service
+ type: ""
+ isreal: true
+- id: {{ b64enc "mock_b_service" }}.1_{{ b64enc "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()" }}
+ name: org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()
+ serviceid: {{ b64enc "mock_b_service" }}.1
+ servicename: mock_b_service
+ type: ""
+ isreal: true
+- id: {{ b64enc "User" }}.0_{{ b64enc "User" }}
+ name: User
+ serviceid: {{ b64enc "User" }}.0
+ servicename: User
+ type: ""
+ isreal: false
+{{- end }}
+calls:
+{{- contains .calls }}
+- source: {{ b64enc "mock_a_service" }}.1_{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+ sourcecomponents: []
+ target: {{ b64enc "mock_b_service" }}.1_{{ b64enc "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()" }}
+ targetcomponents: []
+ id: {{ b64enc "mock_a_service" }}.1-{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}-{{ b64enc "mock_b_service" }}.1-{{ b64enc "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()" }}
+ detectpoints:
+ - SERVER
+- source: {{ b64enc "User" }}.0_{{ b64enc "User" }}
+ sourcecomponents: []
+ target: {{ b64enc "mock_a_service" }}.1_{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+ targetcomponents: []
+ id: {{ b64enc "User" }}.0-{{ b64enc "User" }}-{{ b64enc "mock_a_service" }}.1-{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+ detectpoints:
+ - SERVER
+{{- end }}
diff --git a/test/e2e-v2/cases/lifecycle/expected/dependency-instance.yml b/test/e2e-v2/cases/lifecycle/expected/dependency-instance.yml
new file mode 100644
index 00000000..7792498d
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/dependency-instance.yml
@@ -0,0 +1,42 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+nodes:
+{{- contains .nodes }}
+- id: {{ b64enc "mock_a_service" }}.1_{{ b64enc "mock_a_service_instance" }}
+ name: mock_a_service_instance
+ serviceid: {{ b64enc "mock_a_service" }}.1
+ servicename: mock_a_service
+ type: ""
+ isreal: true
+- id: {{ b64enc "mock_b_service" }}.1_{{ b64enc "mock_b_service_instance" }}
+ name: mock_b_service_instance
+ serviceid: {{ b64enc "mock_b_service" }}.1
+ servicename: mock_b_service
+ type: ""
+ isreal: true
+{{- end }}
+calls:
+{{- contains .calls }}
+- source: {{ b64enc "mock_a_service" }}.1_{{ b64enc "mock_a_service_instance" }}
+ sourcecomponents: []
+ target: {{ b64enc "mock_b_service" }}.1_{{ b64enc "mock_b_service_instance" }}
+ targetcomponents: []
+ id: {{ b64enc "mock_a_service" }}.1_{{ b64enc "mock_a_service_instance" }}-{{ b64enc "mock_b_service" }}.1_{{ b64enc "mock_b_service_instance" }}
+ detectpoints:
+ - CLIENT
+ - SERVER
+{{- end }}
diff --git a/test/e2e-v2/cases/lifecycle/expected/dependency-services.yml b/test/e2e-v2/cases/lifecycle/expected/dependency-services.yml
new file mode 100644
index 00000000..d2d62e17
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/dependency-services.yml
@@ -0,0 +1,58 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+nodes:
+{{- contains .nodes }}
+- id: {{ b64enc "User"}}.0
+ name: User
+ type: USER
+ isreal: false
+ layers:
+ - UNDEFINED
+- id: {{ b64enc "mock_a_service"}}.1
+ name: mock_a_service
+ type: Tomcat
+ isreal: true
+ layers:
+ - GENERAL
+- id: {{ b64enc "mock_b_service"}}.1
+ name: mock_b_service
+ type: Dubbo
+ isreal: true
+ layers:
+ - GENERAL
+{{- end }}
+calls:
+{{- contains .calls }}
+- source: {{ b64enc "mock_a_service"}}.1
+ sourcecomponents:
+ - Dubbo
+ target: {{ b64enc "mock_b_service"}}.1
+ targetcomponents:
+ - Dubbo
+ id: {{ b64enc "mock_a_service"}}.1-{{ b64enc "mock_b_service"}}.1
+ detectpoints:
+ - CLIENT
+ - SERVER
+- source: {{ b64enc "User" }}.0
+ sourcecomponents: []
+ target: {{ b64enc "mock_a_service"}}.1
+ targetcomponents:
+ - Tomcat
+ id: {{ b64enc "User" }}.0-{{ b64enc "mock_a_service"}}.1
+ detectpoints:
+ - SERVER
+{{- end }}
diff --git a/test/e2e-v2/cases/lifecycle/expected/topN-OP-endpoint.yml b/test/e2e-v2/cases/lifecycle/expected/topN-OP-endpoint.yml
new file mode 100644
index 00000000..c2bca184
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/topN-OP-endpoint.yml
@@ -0,0 +1,58 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+type: SORTED_LIST
+results:
+ - metric:
+ labels: []
+ values:
+ - id: "mock_a_service - /dubbox-case/case/dubbox-rest/404-test"
+ value: "{{ (index (index .results 0).values 0).value }}"
+ traceid: null
+ owner:
+ scope: Endpoint
+ serviceid: {{ b64enc "mock_a_service"}}.1
+ servicename: mock_a_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: {{ b64enc "mock_a_service" }}.1_{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+ endpointname: /dubbox-case/case/dubbox-rest/404-test
+ - id: "mock_b_service - org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()"
+ value: "{{ (index (index .results 0).values 1).value }}"
+ traceid: null
+ owner:
+ scope: Endpoint
+ serviceid: {{ b64enc "mock_b_service" }}.1
+ servicename: mock_b_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: {{ b64enc "mock_b_service" }}.1_{{ b64enc "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()" }}
+ endpointname: org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()
+ - id: "mock_c_service - org.apache.skywalking.RocketMQ"
+ value: "{{ (index (index .results 0).values 2).value }}"
+ traceid: null
+ owner:
+ scope: Endpoint
+ serviceid: {{ b64enc "mock_c_service" }}.1
+ servicename: mock_c_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: {{ b64enc "mock_c_service" }}.1_{{ b64enc "org.apache.skywalking.RocketMQ" }}
+ endpointname: org.apache.skywalking.RocketMQ
+error: null
diff --git a/test/e2e-v2/cases/lifecycle/expected/topN-OP-service.yml b/test/e2e-v2/cases/lifecycle/expected/topN-OP-service.yml
new file mode 100644
index 00000000..6f9f0786
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/topN-OP-service.yml
@@ -0,0 +1,58 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+type: SORTED_LIST
+results:
+ - metric:
+ labels: []
+ values:
+ - id: mock_a_service
+ value: "{{ (index (index .results 0).values 0).value }}"
+ traceid: null
+ owner:
+ scope: Service
+ serviceid: {{ b64enc "mock_a_service"}}.1
+ servicename: mock_a_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: null
+ endpointname: null
+ - id: mock_b_service
+ value: "{{ (index (index .results 0).values 1).value }}"
+ traceid: null
+ owner:
+ scope: Service
+ serviceid: {{ b64enc "mock_b_service"}}.1
+ servicename: mock_b_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: null
+ endpointname: null
+ - id: mock_c_service
+ value: "{{ (index (index .results 0).values 2).value }}"
+ traceid: null
+ owner:
+ scope: Service
+ serviceid: {{ b64enc "mock_c_service" }}.1
+ servicename: mock_c_service
+ normal: true
+ serviceinstanceid: null
+ serviceinstancename: null
+ endpointid: null
+ endpointname: null
+error: null
diff --git a/test/e2e-v2/cases/lifecycle/expected/trace-detail.yml b/test/e2e-v2/cases/lifecycle/expected/trace-detail.yml
new file mode 100644
index 00000000..2b8d6837
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/trace-detail.yml
@@ -0,0 +1,45 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+spans:
+ {{- contains .spans }}
+ - traceid: {{ .traceid }}
+ segmentid: {{ .segmentid }}
+ spanid: {{ .spanid }}
+ parentspanid: {{ .parentspanid }}
+ refs: []
+ servicecode: mock_a_service
+ serviceinstancename: mock_a_service_instance
+ starttime: {{ gt .starttime 0 }}
+ endtime: {{ gt .endtime 0 }}
+ endpointname: /dubbox-case/case/dubbox-rest/404-test
+ type: Entry
+ peer: ""
+ component: Tomcat
+ iserror: false
+ layer: Http
+ tags:
+ {{- contains .tags }}
+ - key: http.method
+ value: get
+ - key: http.status_code
+ value: "404"
+ - key: http.status_code
+ value: "200"
+ {{- end }}
+ logs: []
+ attachedevents: []
+ {{- end }}
diff --git a/test/e2e-v2/cases/lifecycle/expected/traces-list.yml b/test/e2e-v2/cases/lifecycle/expected/traces-list.yml
new file mode 100644
index 00000000..9c5e4f15
--- /dev/null
+++ b/test/e2e-v2/cases/lifecycle/expected/traces-list.yml
@@ -0,0 +1,27 @@
+# Licensed to the 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.
+# The 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.
+
+debuggingtrace: null
+traces:
+{{- contains .traces }}
+- segmentid: {{ notEmpty .segmentid }}
+ endpointnames:
+ - /dubbox-case/case/dubbox-rest/404-test
+ duration: {{ ge .duration 0 }}
+ start: "{{ notEmpty .start}}"
+ iserror: false
+ traceids:
+ - {{ index .traceids 0 }}
+{{- end }}
diff --git a/test/e2e-v2/script/env b/test/e2e-v2/script/env
index 103c9a00..790a3cda 100644
--- a/test/e2e-v2/script/env
+++ b/test/e2e-v2/script/env
@@ -26,6 +26,7 @@ SW_ROVER_COMMIT=4c0cb8429a96f190ea30eac1807008d523c749c3
SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
SW_PREDICTOR_COMMIT=54a0197654a3781a6f73ce35146c712af297c994
-SW_OAP_COMMIT=03b03518616a91129cfced2972621df3f0bac3db
-SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=d3f8fe894d1a206164b73f5b523d2eb62d9e9965
-SW_CTL_COMMIT=67cbc89dd7b214d5791321a7ca992f940cb586ba
\ No newline at end of file
+SW_OAP_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
+SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
+SW_CTL_COMMIT=77b4c49e89c9c000278f44e62729d534f2ec842e
+SW_TRACE_MOCKER_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
\ No newline at end of file
diff --git a/test/stress/classic/env b/test/stress/classic/env
index c004cc4a..5ae3a643 100644
--- a/test/stress/classic/env
+++ b/test/stress/classic/env
@@ -24,9 +24,9 @@ SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
SW_KUBERNETES_COMMIT_SHA=6fe5e6f0d3b7686c6be0457733e825ee68cb9b35
SW_ROVER_COMMIT=40d03c14d638339bf4aa6ec7376d421323de865a
SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
-SW_CTL_COMMIT=67cbc89dd7b214d5791321a7ca992f940cb586ba
+SW_CTL_COMMIT=77b4c49e89c9c000278f44e62729d534f2ec842e
-SW_OAP_COMMIT=6273cb2c16e69a0a4df4e20d4a182469ff3c276c
-SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=d3f8fe894d1a206164b73f5b523d2eb62d9e9965
+SW_OAP_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
+SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
VUS=10
diff --git a/test/stress/classic/env.dev b/test/stress/classic/env.dev
index f425b43a..7b44a3f3 100644
--- a/test/stress/classic/env.dev
+++ b/test/stress/classic/env.dev
@@ -24,9 +24,9 @@ SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
SW_KUBERNETES_COMMIT_SHA=6fe5e6f0d3b7686c6be0457733e825ee68cb9b35
SW_ROVER_COMMIT=40d03c14d638339bf4aa6ec7376d421323de865a
SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
-SW_CTL_COMMIT=67cbc89dd7b214d5791321a7ca992f940cb586ba
+SW_CTL_COMMIT=77b4c49e89c9c000278f44e62729d534f2ec842e
-SW_OAP_COMMIT=6273cb2c16e69a0a4df4e20d4a182469ff3c276c
-SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=d3f8fe894d1a206164b73f5b523d2eb62d9e9965
+SW_OAP_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
+SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=a2a67ca63084cddf82303155c185e3c24cf07eef
VUS=1