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/03/10 02:58:41 UTC
(skywalking-banyandb) branch lifecycle created (now a3744462)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch lifecycle
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at a3744462 Add lifecycle management
This branch includes the following new commits:
new a3744462 Add 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 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 lifecycle
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit a3744462a7777f42a31c816ba7e19cb79b150d3b
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Mar 10 09:54:32 2025 +0800
Add lifecycle management
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
api/common/id.go | 5 +-
api/data/measure.go | 9 +
api/data/stream.go | 9 +
api/proto/banyandb/measure/v1/rpc.proto | 11 +
api/proto/banyandb/stream/v1/rpc.proto | 12 +
banyand/backup/backup.go | 49 +---
banyand/backup/backup_test.go | 3 +-
banyand/backup/lifecycle/lifecycle.go | 102 +++++++++
banyand/backup/lifecycle/progress.go | 193 ++++++++++++++++
banyand/backup/lifecycle/service.go | 381 ++++++++++++++++++++++++++++++++
banyand/backup/lifecycle/steps.go | 270 ++++++++++++++++++++++
banyand/backup/restore.go | 25 +--
banyand/backup/restore_test.go | 5 +-
banyand/backup/snapshot/snapshot.go | 100 +++++++++
banyand/internal/storage/index.go | 3 +
banyand/internal/storage/rotation.go | 12 +-
banyand/internal/storage/segment.go | 35 +++
banyand/internal/storage/storage.go | 2 +
banyand/internal/storage/tsdb.go | 13 +-
banyand/measure/metadata.go | 59 ++++-
banyand/measure/query.go | 2 +
banyand/measure/service.go | 55 ++++-
banyand/metadata/client.go | 6 +-
banyand/metadata/metadata.go | 1 +
banyand/queue/local.go | 4 +
banyand/queue/pub/pub.go | 6 +-
banyand/queue/pub/selector.go | 13 +-
banyand/queue/pub/selector_test.go | 30 +--
banyand/queue/queue.go | 1 +
banyand/queue/sub/segments.go | 82 +++++++
banyand/queue/sub/server.go | 5 +
banyand/stream/metadata.go | 60 ++++-
banyand/stream/service.go | 59 ++++-
banyand/stream/stream.go | 2 +
docs/api-reference.md | 70 ++++++
pkg/index/inverted/inverted_series.go | 1 +
37 files changed, 1580 insertions(+), 116 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 42ea115c..39564254 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -31,6 +31,7 @@ Release Notes.
- Add concurrent barrier to partition merge to improve the performance of the partition merge.
- Improve the write performance.
- Add node labels to classify the nodes.
+- Add lifecycle management for the node.
### Bug Fixes
diff --git a/api/common/id.go b/api/common/id.go
index 36c6692f..586b6535 100644
--- a/api/common/id.go
+++ b/api/common/id.go
@@ -222,11 +222,12 @@ func GenerateNode(grpcPort, httpPort *uint32) (node Node, err error) {
if httpPort != nil {
node.HTTPAddress = net.JoinHostPort(nodeHost, strconv.FormatUint(uint64(*httpPort), 10))
}
- node.Labels = parseNodeFlags()
+ node.Labels = ParseNodeFlags()
return node, nil
}
-func parseNodeFlags() map[string]string {
+// ParseNodeFlags parses the node labels from flag.
+func ParseNodeFlags() map[string]string {
labels := make(map[string]string)
for _, label := range FlagNodeLabels {
parts := strings.Split(label, "=")
diff --git a/api/data/measure.go b/api/data/measure.go
index 8b94dcd0..512848e3 100644
--- a/api/data/measure.go
+++ b/api/data/measure.go
@@ -48,3 +48,12 @@ var TopNQueryKindVersion = common.KindVersion{
// TopicTopNQuery is the top-n query topic.
var TopicTopNQuery = bus.BiTopic(TopNQueryKindVersion.String())
+
+// MeasureDeleteExpiredSegmentsKindVersion is the version tag of measure delete kind.
+var MeasureDeleteExpiredSegmentsKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "measure-delete-expired-segments",
+}
+
+// TopicMeasureDeleteExpiredSegments is the measure delete topic.
+var TopicMeasureDeleteExpiredSegments = bus.BiTopic(MeasureDeleteExpiredSegmentsKindVersion.String())
diff --git a/api/data/stream.go b/api/data/stream.go
index 7677731b..08877d7c 100644
--- a/api/data/stream.go
+++ b/api/data/stream.go
@@ -39,3 +39,12 @@ var StreamQueryKindVersion = common.KindVersion{
// TopicStreamQuery is the stream query topic.
var TopicStreamQuery = bus.BiTopic(StreamQueryKindVersion.String())
+
+// StreamDeleteExpiredSegmentsKindVersion is the version tag of stream delete segments kind.
+var StreamDeleteExpiredSegmentsKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "stream-delete-expired-segments",
+}
+
+// TopicDeleteExpiredStreamSegments is the delete stream segments topic.
+var TopicDeleteExpiredStreamSegments = bus.BiTopic(StreamDeleteExpiredSegmentsKindVersion.String())
diff --git a/api/proto/banyandb/measure/v1/rpc.proto b/api/proto/banyandb/measure/v1/rpc.proto
index 5f216834..22e71173 100644
--- a/api/proto/banyandb/measure/v1/rpc.proto
+++ b/api/proto/banyandb/measure/v1/rpc.proto
@@ -22,6 +22,7 @@ package banyandb.measure.v1;
import "banyandb/measure/v1/query.proto";
import "banyandb/measure/v1/topn.proto";
import "banyandb/measure/v1/write.proto";
+import "banyandb/model/v1/query.proto";
import "google/api/annotations.proto";
import "protoc-gen-openapiv2/options/annotations.proto";
@@ -29,6 +30,15 @@ option go_package = "github.com/apache/skywalking-banyandb/api/proto/banyandb/me
option java_package = "org.apache.skywalking.banyandb.measure.v1";
option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_swagger) = {base_path: "/api"};
+message DeleteExpiredSegmentsRequest {
+ string group = 1;
+ model.v1.TimeRange time_range = 2;
+}
+
+message DeleteExpiredSegmentsResponse {
+ int64 deleted = 1;
+}
+
service MeasureService {
rpc Query(QueryRequest) returns (QueryResponse) {
option (google.api.http) = {
@@ -44,4 +54,5 @@ service MeasureService {
body: "*"
};
}
+ rpc DeleteExpiredSegments(DeleteExpiredSegmentsRequest) returns (DeleteExpiredSegmentsResponse);
}
diff --git a/api/proto/banyandb/stream/v1/rpc.proto b/api/proto/banyandb/stream/v1/rpc.proto
index adff33db..dcbe266e 100644
--- a/api/proto/banyandb/stream/v1/rpc.proto
+++ b/api/proto/banyandb/stream/v1/rpc.proto
@@ -19,6 +19,7 @@ syntax = "proto3";
package banyandb.stream.v1;
+import "banyandb/model/v1/query.proto";
import "banyandb/stream/v1/query.proto";
import "banyandb/stream/v1/write.proto";
import "google/api/annotations.proto";
@@ -28,6 +29,15 @@ option go_package = "github.com/apache/skywalking-banyandb/api/proto/banyandb/st
option java_package = "org.apache.skywalking.banyandb.stream.v1";
option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_swagger) = {base_path: "/api"};
+message DeleteExpiredSegmentsRequest {
+ string group = 1;
+ model.v1.TimeRange time_range = 2;
+}
+
+message DeleteExpiredSegmentsResponse {
+ int64 deleted = 1;
+}
+
service StreamService {
rpc Query(QueryRequest) returns (QueryResponse) {
option (google.api.http) = {
@@ -37,4 +47,6 @@ service StreamService {
}
rpc Write(stream WriteRequest) returns (stream WriteResponse);
+
+ rpc DeleteExpiredSegments(DeleteExpiredSegmentsRequest) returns (DeleteExpiredSegmentsResponse);
}
diff --git a/banyand/backup/backup.go b/banyand/backup/backup.go
index e8ed8003..fa45e46d 100644
--- a/banyand/backup/backup.go
+++ b/banyand/backup/backup.go
@@ -35,13 +35,10 @@ import (
"github.com/spf13/cobra"
"go.uber.org/multierr"
- commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/fs/remote"
"github.com/apache/skywalking-banyandb/pkg/fs/remote/local"
- "github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
"github.com/apache/skywalking-banyandb/pkg/version"
@@ -134,21 +131,21 @@ func backupAction(dest, gRPCAddr string, enableTLS, insecure bool, cert,
}
defer fs.Close()
- snapshots, err := getSnapshots(gRPCAddr, enableTLS, insecure, cert)
+ snapshots, err := snapshot.Get(gRPCAddr, enableTLS, insecure, cert)
if err != nil {
return err
}
timeDir := getTimeDir(timeStyle)
- for _, snapshot := range snapshots {
+ for _, snp := range snapshots {
var snapshotDir string
- snapshotDir, err = getSnapshotDir(snapshot, streamRoot, measureRoot, propertyRoot)
+ snapshotDir, err = snapshot.Dir(snp, streamRoot, measureRoot, propertyRoot)
if err != nil {
- logger.Warningf("Failed to get snapshot directory for %s: %v", snapshot.Name, err)
+ logger.Warningf("Failed to get snapshot directory for %s: %v", snp.Name, err)
continue
}
- multierr.AppendInto(&err, backupSnapshot(fs, snapshotDir, getCatalogName(snapshot.Catalog), timeDir))
+ multierr.AppendInto(&err, backupSnapshot(fs, snapshotDir, snapshot.CatalogName(snp.Catalog), timeDir))
}
return err
}
@@ -167,40 +164,6 @@ func newFS(dest string) (remote.FS, error) {
}
}
-func getSnapshots(gRPCAddr string, enableTLS, insecure bool, cert string) ([]*databasev1.Snapshot, error) {
- opts, err := grpchelper.SecureOptions(nil, enableTLS, insecure, cert)
- if err != nil {
- return nil, err
- }
- connection, err := grpchelper.Conn(gRPCAddr, 10*time.Second, opts...)
- if err != nil {
- return nil, fmt.Errorf("failed to connect to gRPC server: %w", err)
- }
- defer connection.Close()
- ctx := context.Background()
- client := databasev1.NewSnapshotServiceClient(connection)
- snapshotResp, err := client.Snapshot(ctx, &databasev1.SnapshotRequest{})
- if err != nil {
- return nil, fmt.Errorf("failed to request snapshot: %w", err)
- }
- return snapshotResp.Snapshots, nil
-}
-
-func getSnapshotDir(snapshot *databasev1.Snapshot, streamRoot, measureRoot, propertyRoot string) (string, error) {
- var baseDir string
- switch snapshot.Catalog {
- case commonv1.Catalog_CATALOG_STREAM:
- baseDir = getLocalDir(streamRoot, snapshot.Catalog)
- case commonv1.Catalog_CATALOG_MEASURE:
- baseDir = getLocalDir(measureRoot, snapshot.Catalog)
- case commonv1.Catalog_CATALOG_PROPERTY:
- baseDir = getLocalDir(propertyRoot, snapshot.Catalog)
- default:
- return "", errors.New("unknown catalog type")
- }
- return filepath.Join(baseDir, storage.SnapshotsDir, snapshot.Name), nil
-}
-
func getTimeDir(style string) string {
now := time.Now()
switch style {
diff --git a/banyand/backup/backup_test.go b/banyand/backup/backup_test.go
index 8d9bd5d3..641aad35 100644
--- a/banyand/backup/backup_test.go
+++ b/banyand/backup/backup_test.go
@@ -28,6 +28,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
)
@@ -78,7 +79,7 @@ func TestGetSnapshotDir(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := getSnapshotDir(tt.snapshot, tt.streamRoot, tt.measureRoot, tt.propRoot)
+ got, err := snapshot.Dir(tt.snapshot, tt.streamRoot, tt.measureRoot, tt.propRoot)
if (err != nil) != tt.wantErr {
t.Errorf("getSnapshotDir() error = %v, wantErr %v", err, tt.wantErr)
return
diff --git a/banyand/backup/lifecycle/lifecycle.go b/banyand/backup/lifecycle/lifecycle.go
new file mode 100644
index 00000000..3d0a07aa
--- /dev/null
+++ b/banyand/backup/lifecycle/lifecycle.go
@@ -0,0 +1,102 @@
+// 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 provides the lifecycle migration service.
+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/version"
+)
+
+// NewLifecycleCommand creates a new lifecycle command.
+func NewLifecycleCommand() *cobra.Command {
+ var schedule string
+
+ l := logger.GetLogger("bootstrap")
+
+ metaSvc, err := metadata.NewClient(false)
+ if err != nil {
+ l.Fatal().Err(err).Msg("failed to initiate metadata service")
+ }
+ metricSvc := observability.NewMetricService(metaSvc, nil, "lifecycle", nil)
+ svc := NewService(metaSvc, metricSvc)
+ group := run.NewGroup("lifecycle")
+ group.Register(metaSvc, metricSvc, svc)
+
+ cmd := &cobra.Command{
+ Short: "Backup BanyanDB snapshots to remote storage",
+ DisableAutoGenTag: true,
+ Version: version.Build(),
+ RunE: func(cmd *cobra.Command, _ []string) error {
+ if err := config.Load("logging", cmd.Flags()); err != nil {
+ return err
+ }
+ if schedule == "" {
+ return group.Run(context.Background())
+ }
+ 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")
+ }
+ return true
+ })
+ if err != nil {
+ return err
+ }
+
+ 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 backup scheduler...")
+ sch.Close()
+ return nil
+ },
+ }
+
+ cmd.Flags().StringVar(
+ &schedule,
+ "schedule",
+ "",
+ "Schedule expression for periodic backup. Options: @yearly, @monthly, @weekly, @daily, @hourly or @every <duration>",
+ )
+
+ return cmd
+}
diff --git a/banyand/backup/lifecycle/progress.go b/banyand/backup/lifecycle/progress.go
new file mode 100644
index 00000000..9a5347b9
--- /dev/null
+++ b/banyand/backup/lifecycle/progress.go
@@ -0,0 +1,193 @@
+// 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"
+ "os"
+ "sync"
+
+ "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:"-"`
+}
+
+// NewProgress creates a new Progress tracker.
+func NewProgress() *Progress {
+ return &Progress{
+ CompletedGroups: make(map[string]bool),
+ CompletedStreams: make(map[string]map[string]bool),
+ CompletedMeasures: make(map[string]map[string]bool),
+ DeletedStreamGroups: make(map[string]bool),
+ DeletedMeasureGroups: make(map[string]bool),
+ }
+}
+
+// LoadProgress loads progress from a file if it exists.
+func LoadProgress(path string, l *logger.Logger) *Progress {
+ if path == "" {
+ return NewProgress()
+ }
+
+ data, err := os.ReadFile(path)
+ if err != nil {
+ if os.IsNotExist(err) {
+ l.Info().Msgf("No existing progress file at %s, starting fresh", path)
+ } else {
+ l.Warn().Err(err).Msgf("Failed to read progress file at %s, starting fresh", path)
+ }
+ return NewProgress()
+ }
+
+ progress := NewProgress()
+ if err := json.Unmarshal(data, progress); err != nil {
+ l.Warn().Err(err).Msgf("Failed to parse progress file at %s, starting fresh", path)
+ return NewProgress()
+ }
+
+ l.Info().Msgf("Loaded existing progress from %s", path)
+ return progress
+}
+
+// Save writes the progress to the specified file.
+func (p *Progress) Save(path string, l *logger.Logger) {
+ if path == "" {
+ return
+ }
+
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ data, err := json.MarshalIndent(p, "", " ")
+ if err != nil {
+ l.Error().Err(err).Msg("Failed to marshal progress data")
+ return
+ }
+
+ err = os.WriteFile(path, data, 0o600)
+ if err != nil {
+ l.Error().Err(err).Msgf("Failed to write progress file at %s", path)
+ return
+ }
+ l.Debug().Msg("Progress saved successfully")
+}
+
+// MarkGroupCompleted marks a group as completed.
+func (p *Progress) MarkGroupCompleted(group string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.CompletedGroups[group] = true
+}
+
+// IsGroupCompleted checks if a group has been completed.
+func (p *Progress) IsGroupCompleted(group string) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.CompletedGroups[group]
+}
+
+// MarkStreamCompleted marks a stream as completed.
+func (p *Progress) MarkStreamCompleted(group, stream string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.CompletedStreams[group] == nil {
+ p.CompletedStreams[group] = make(map[string]bool)
+ }
+ p.CompletedStreams[group][stream] = true
+}
+
+// IsStreamCompleted checks if a stream has been completed.
+func (p *Progress) IsStreamCompleted(group, stream string) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if streams, ok := p.CompletedStreams[group]; ok {
+ return streams[stream]
+ }
+ return false
+}
+
+// MarkMeasureCompleted marks a measure as completed.
+func (p *Progress) MarkMeasureCompleted(group, measure string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.CompletedMeasures[group] == nil {
+ p.CompletedMeasures[group] = make(map[string]bool)
+ }
+ p.CompletedMeasures[group][measure] = true
+}
+
+// IsMeasureCompleted checks if a measure has been completed.
+func (p *Progress) IsMeasureCompleted(group, measure string) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if measures, ok := p.CompletedMeasures[group]; ok {
+ return measures[measure]
+ }
+ return false
+}
+
+// MarkStreamGroupDeleted marks a stream group segments as deleted.
+func (p *Progress) MarkStreamGroupDeleted(group string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.DeletedStreamGroups[group] = true
+}
+
+// IsStreamGroupDeleted checks if a stream group segments have been deleted.
+func (p *Progress) IsStreamGroupDeleted(group string) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.DeletedStreamGroups[group]
+}
+
+// MarkMeasureGroupDeleted marks a measure group segments as deleted.
+func (p *Progress) MarkMeasureGroupDeleted(group string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.DeletedMeasureGroups[group] = true
+}
+
+// IsMeasureGroupDeleted checks if a measure group segments have been deleted.
+func (p *Progress) IsMeasureGroupDeleted(group string) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.DeletedMeasureGroups[group]
+}
+
+// Remove deletes the progress file.
+func (p *Progress) Remove(path string, l *logger.Logger) {
+ if path == "" {
+ return
+ }
+
+ if err := os.Remove(path); err != nil {
+ if !os.IsNotExist(err) {
+ l.Warn().Err(err).Msgf("Failed to remove progress file at %s", path)
+ }
+ return
+ }
+ l.Info().Msgf("Removed progress file at %s", path)
+}
diff --git a/banyand/backup/lifecycle/service.go b/banyand/backup/lifecycle/service.go
new file mode 100644
index 00000000..36bc25d4
--- /dev/null
+++ b/banyand/backup/lifecycle/service.go
@@ -0,0 +1,381 @@
+// 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 (
+ "context"
+
+ "google.golang.org/grpc"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
+ "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/observability"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/stream"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+type service interface {
+ run.Config
+ run.Service
+}
+
+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
+}
+
+// NewService creates a new lifecycle service.
+func NewService(meta metadata.Repo, omr observability.MetricsRegistry) run.Unit {
+ return &lifecycleService{
+ metadata: meta,
+ omr: omr,
+ }
+}
+
+func (l *lifecycleService) FlagSet() *run.FlagSet {
+ flagS := run.NewFlagSet(l.Name())
+ flagS.StringSliceVar(&common.FlagNodeLabels, "node-labels", nil, "the node labels. e.g. key1=value1,key2=value2")
+ flagS.StringVar(&l.gRPCAddr, "grpc-addr", "127.0.0.1:17912", "gRPC address of the data node")
+ flagS.BoolVar(&l.enableTLS, "enable-tls", false, "Enable TLS for gRPC connection")
+ flagS.BoolVar(&l.insecure, "insecure", false, "Skip server certificate verification")
+ flagS.StringVar(&l.cert, "cert", "", "Path to the gRPC server certificate")
+ 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")
+ return flagS
+}
+
+func (l *lifecycleService) Validate() error {
+ return nil
+}
+
+func (l *lifecycleService) GracefulStop() {
+}
+
+func (l *lifecycleService) Name() string {
+ return "lifecycle"
+}
+
+func (l *lifecycleService) Serve() run.StopNotify {
+ l.l = logger.GetLogger("lifecycle")
+ ctx := context.Background()
+
+ // Load progress from file if it exists
+ progress := LoadProgress(l.progressFilePath, l.l)
+
+ groups, err := l.getGroupsToProcess(ctx, progress)
+ if err != nil {
+ return nil
+ }
+
+ if len(groups) == 0 {
+ l.l.Info().Msg("no groups to process, all groups already completed")
+ progress.Remove(l.progressFilePath, l.l)
+ return nil
+ }
+
+ streamDir, measureDir, err := l.getSnapshots(groups)
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to get snapshots")
+ return nil
+ }
+ streamSVC, measureSVC, err := l.setupQuerySvc(ctx, streamDir, measureDir)
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to setup query service")
+ return nil
+ }
+ 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 nil
+ }
+ labels := common.ParseNodeFlags()
+
+ for _, g := range groups {
+ switch g.Catalog {
+ case commonv1.Catalog_CATALOG_STREAM:
+ l.processStreamGroup(ctx, g, streamSVC, nodes, labels, progress)
+ case commonv1.Catalog_CATALOG_MEASURE:
+ 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 nil
+}
+
+func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Progress) ([]*commonv1.Group, error) {
+ gg, err := l.metadata.GroupRegistry().ListGroup(ctx)
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to list groups")
+ return nil, err
+ }
+
+ groups := make([]*commonv1.Group, 0, len(gg))
+ for _, g := range gg {
+ if g.ResourceOpts == nil {
+ continue
+ }
+ if len(g.ResourceOpts.Stages) == 0 {
+ continue
+ }
+ if progress.IsGroupCompleted(g.Metadata.Name) {
+ l.l.Info().Msgf("skipping already completed group: %s", g.Metadata.Name)
+ continue
+ }
+ groups = append(groups, g)
+ }
+
+ return groups, nil
+}
+
+func (l *lifecycleService) processStreamGroup(ctx context.Context, g *commonv1.Group, streamSVC stream.Service,
+ nodes []*databasev1.Node, labels map[string]string, progress *Progress,
+) {
+ shardNum, selector, client, err := parseGroup(g, labels, nodes, l.l)
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
+ return
+ }
+
+ ss, err := l.metadata.StreamRegistry().ListStream(ctx, schema.ListOpt{Group: g.Metadata.Name})
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to list streams in group %s", g.Metadata.Name)
+ return
+ }
+
+ tr := streamSVC.GetRemovalSegmentsTimeRange(g.Metadata.Name)
+
+ l.processStreams(ctx, g, ss, streamSVC, tr, shardNum, selector, client, progress)
+
+ l.deleteExpiredStreamSegments(ctx, g, tr, progress)
+}
+
+func (l *lifecycleService) processStreams(ctx context.Context, g *commonv1.Group, streams []*databasev1.Stream,
+ streamSVC stream.Service, tr *timestamp.TimeRange, shardNum 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)
+ continue
+ }
+
+ if err := l.processSingleStream(ctx, s, streamSVC, tr, shardNum, selector, client); err != nil {
+ continue
+ }
+
+ progress.MarkStreamCompleted(g.Metadata.Name, s.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ }
+}
+
+func (l *lifecycleService) processSingleStream(ctx context.Context, s *databasev1.Stream,
+ streamSVC stream.Service, tr *timestamp.TimeRange, shardNum uint32, selector node.Selector, client queue.Client,
+) error {
+ q, err := streamSVC.Stream(s.Metadata)
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to get stream %s", s.Metadata.Name)
+ return err
+ }
+
+ tagProjection := make([]model.TagProjection, len(s.TagFamilies))
+ for i, tf := range s.TagFamilies {
+ tagProjection[i] = model.TagProjection{
+ Family: tf.Name,
+ Names: make([]string, len(tf.Tags)),
+ }
+ for j, t := range tf.Tags {
+ tagProjection[i].Names[j] = t.Name
+ }
+ }
+
+ result, err := q.Query(ctx, model.StreamQueryOptions{
+ Name: s.Metadata.Name,
+ TagProjection: tagProjection,
+ TimeRange: tr,
+ })
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to query stream %s", s.Metadata.Name)
+ return err
+ }
+
+ migrateStream(ctx, s, result, shardNum, selector, client, l.l)
+ return nil
+}
+
+func (l *lifecycleService) deleteExpiredStreamSegments(ctx context.Context, g *commonv1.Group, tr *timestamp.TimeRange, progress *Progress) {
+ if progress.IsStreamGroupDeleted(g.Metadata.Name) {
+ l.l.Info().Msgf("skipping already deleted stream group segments: %s", g.Metadata.Name)
+ return
+ }
+
+ resp, err := snapshot.Conn(l.gRPCAddr, l.enableTLS, l.insecure, l.cert, func(conn *grpc.ClientConn) (*streamv1.DeleteExpiredSegmentsResponse, error) {
+ client := streamv1.NewStreamServiceClient(conn)
+ return client.DeleteExpiredSegments(ctx, &streamv1.DeleteExpiredSegmentsRequest{
+ Group: g.Metadata.Name,
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(tr.Start),
+ End: timestamppb.New(tr.End),
+ },
+ })
+ })
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to delete expired segments in group %s", g.Metadata.Name)
+ return
+ }
+
+ l.l.Info().Msgf("deleted %d expired segments in group %s", resp.Deleted, g.Metadata.Name)
+ progress.MarkStreamGroupDeleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+}
+
+func (l *lifecycleService) processMeasureGroup(ctx context.Context, g *commonv1.Group, measureSVC measure.Service,
+ nodes []*databasev1.Node, labels map[string]string, progress *Progress,
+) {
+ shardNum, selector, client, err := parseGroup(g, labels, nodes, l.l)
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
+ return
+ }
+
+ mm, err := l.metadata.MeasureRegistry().ListMeasure(ctx, schema.ListOpt{Group: g.Metadata.Name})
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to list measures in group %s", g.Metadata.Name)
+ return
+ }
+
+ tr := measureSVC.GetRemovalSegmentsTimeRange(g.Metadata.Name)
+
+ l.processMeasures(ctx, g, mm, measureSVC, tr, shardNum, selector, client, progress)
+
+ l.deleteExpiredMeasureSegments(ctx, g, tr, progress)
+}
+
+func (l *lifecycleService) processMeasures(ctx context.Context, g *commonv1.Group, measures []*databasev1.Measure,
+ measureSVC measure.Service, tr *timestamp.TimeRange, shardNum 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)
+ continue
+ }
+
+ if err := l.processSingleMeasure(ctx, m, measureSVC, tr, shardNum, selector, client); err != nil {
+ continue
+ }
+
+ progress.MarkMeasureCompleted(g.Metadata.Name, m.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+ }
+}
+
+func (l *lifecycleService) processSingleMeasure(ctx context.Context, m *databasev1.Measure,
+ measureSVC measure.Service, tr *timestamp.TimeRange, shardNum uint32, selector node.Selector, client queue.Client,
+) error {
+ q, err := measureSVC.Measure(m.Metadata)
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to get measure %s", m.Metadata.Name)
+ return err
+ }
+
+ tagProjection := make([]model.TagProjection, len(m.TagFamilies))
+ for i, tf := range m.TagFamilies {
+ tagProjection[i] = model.TagProjection{
+ Family: tf.Name,
+ Names: make([]string, len(tf.Tags)),
+ }
+ for j, t := range tf.Tags {
+ tagProjection[i].Names[j] = t.Name
+ }
+ }
+ fieldProjection := make([]string, len(m.Fields))
+ for i, f := range m.Fields {
+ fieldProjection[i] = f.Name
+ }
+
+ result, err := q.Query(ctx, model.MeasureQueryOptions{
+ Name: m.Metadata.Name,
+ TagProjection: tagProjection,
+ FieldProjection: fieldProjection,
+ TimeRange: tr,
+ })
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to query measure %s", m.Metadata.Name)
+ return err
+ }
+
+ migrateMeasure(ctx, m, result, shardNum, selector, client, l.l)
+ return nil
+}
+
+func (l *lifecycleService) deleteExpiredMeasureSegments(ctx context.Context, g *commonv1.Group, tr *timestamp.TimeRange, progress *Progress) {
+ if progress.IsMeasureGroupDeleted(g.Metadata.Name) {
+ l.l.Info().Msgf("skipping already deleted measure group segments: %s", g.Metadata.Name)
+ return
+ }
+
+ resp, err := snapshot.Conn(l.gRPCAddr, l.enableTLS, l.insecure, l.cert, func(conn *grpc.ClientConn) (*measurev1.DeleteExpiredSegmentsResponse, error) {
+ client := measurev1.NewMeasureServiceClient(conn)
+ return client.DeleteExpiredSegments(ctx, &measurev1.DeleteExpiredSegmentsRequest{
+ Group: g.Metadata.Name,
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(tr.Start),
+ End: timestamppb.New(tr.End),
+ },
+ })
+ })
+ if err != nil {
+ l.l.Error().Err(err).Msgf("failed to delete expired segments in group %s", g.Metadata.Name)
+ return
+ }
+
+ l.l.Info().Msgf("deleted %d expired segments in group %s", resp.Deleted, g.Metadata.Name)
+ progress.MarkMeasureGroupDeleted(g.Metadata.Name)
+ progress.Save(l.progressFilePath, l.l)
+}
diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go
new file mode 100644
index 00000000..b954d710
--- /dev/null
+++ b/banyand/backup/lifecycle/steps.go
@@ -0,0 +1,270 @@
+// 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 (
+ "context"
+ "encoding/base64"
+ "fmt"
+ "time"
+
+ "github.com/pkg/errors"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "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"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+func (l *lifecycleService) getSnapshots(groups []*commonv1.Group) (streamDir string, measureDir string, err error) {
+ snapshotGroups := make([]*databasev1.SnapshotRequest_Group, 0, len(groups))
+ for _, group := range groups {
+ snapshotGroups = append(snapshotGroups, &databasev1.SnapshotRequest_Group{
+ Group: group.Metadata.Name,
+ Catalog: group.Catalog,
+ })
+ }
+ snn, err := snapshot.Get(l.gRPCAddr, l.enableTLS, l.insecure, l.cert, snapshotGroups...)
+ if err != nil {
+ return "", "", err
+ }
+ for _, snp := range snn {
+ snapshotDir, errDir := snapshot.Dir(snp, l.streamRoot, l.measureRoot, "")
+ if errDir != nil {
+ logger.Warningf("Failed to get snapshot directory for %s: %v", snp.Name, errDir)
+ continue
+ }
+ if snp.Catalog == commonv1.Catalog_CATALOG_STREAM {
+ streamDir = snapshotDir
+ }
+ if snp.Catalog == commonv1.Catalog_CATALOG_MEASURE {
+ measureDir = snapshotDir
+ }
+ }
+ return streamDir, measureDir, nil
+}
+
+func (l *lifecycleService) setupQuerySvc(ctx context.Context, streamDir, measureDir string) (stream.Service, measure.Service, error) {
+ pm := protector.NewMemory(l.omr)
+ streamSVC, err := stream.NewReadonlyService(l.metadata, l.omr, pm)
+ if err != nil {
+ return nil, nil, err
+ }
+ if err = streamSVC.FlagSet().Parse([]string{"--stream-root-path", streamDir}); err != nil {
+ return nil, nil, err
+ }
+ if err = streamSVC.PreRun(ctx); err != nil {
+ return nil, nil, err
+ }
+ defer streamSVC.GracefulStop()
+ measureSVC, err := measure.NewReadonlyService(l.metadata, l.omr, pm)
+ if err != nil {
+ return nil, nil, err
+ }
+ if err = measureSVC.FlagSet().Parse([]string{"--measure-root-path", measureDir}); err != nil {
+ return nil, nil, err
+ }
+ if err = measureSVC.PreRun(ctx); err != nil {
+ return nil, nil, err
+ }
+ defer measureSVC.GracefulStop()
+ return streamSVC, measureSVC, nil
+}
+
+func parseGroup(g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node, l *logger.Logger) (uint32, node.Selector, queue.Client, error) {
+ ro := g.ResourceOpts
+ if ro == nil {
+ return 0, nil, nil, fmt.Errorf("no resource opts in group %s", g.Metadata.Name)
+ }
+ if len(ro.Stages) == 0 {
+ return 0, nil, nil, fmt.Errorf("no stages in group %s", g.Metadata.Name)
+ }
+ var nst *commonv1.LifecycleStage
+ for i, st := range ro.Stages {
+ selector, err := pub.ParseLabelSelector(st.NodeSelector)
+ if err != nil {
+ return 0, nil, nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
+ }
+ if !selector.Matches(nodeLabels) {
+ continue
+ }
+ if i+1 >= len(ro.Stages) {
+ l.Info().Msgf("no next stage for group %s at stage %s", g.Metadata.Name, st.Name)
+ return 0, nil, nil, nil
+ }
+ nst = ro.Stages[i+1]
+ l.Info().Msgf("migrating group %s at stage %s to stage %s", g.Metadata.Name, st.Name, nst.Name)
+ break
+ }
+ if nst == nil {
+ nst = ro.Stages[0]
+ }
+ nsl, err := pub.ParseLabelSelector(nst.NodeSelector)
+ if err != nil {
+ return 0, nil, nil, errors.WithMessagef(err, "failed to parse node selector %s", nst.NodeSelector)
+ }
+ nodeSel := node.NewRoundRobinSelector("", nil)
+ client := pub.New(nil)
+
+ var existed bool
+ for _, n := range nodes {
+ if n.Labels == nil {
+ continue
+ }
+ if nsl.Matches(n.Labels) {
+ existed = true
+ nodeSel.AddNode(n)
+ client.OnAddOrUpdate(schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Kind: schema.KindNode,
+ },
+ Spec: n,
+ })
+ }
+ }
+ if !existed {
+ return 0, nil, nil, errors.New("no nodes matched")
+ }
+ return nst.ShardNum, nodeSel, client, nil
+}
+
+func migrateStream(ctx context.Context, s *databasev1.Stream, result model.StreamQueryResult,
+ shardNum uint32, selector node.Selector, client queue.Client, l *logger.Logger,
+) {
+ defer result.Release()
+
+ entityLocator := partition.NewEntityLocator(s.TagFamilies, s.Entity, 0)
+
+ batch := client.NewBatchPublisher(30 * time.Second)
+ defer batch.Close()
+ for sr := result.Pull(ctx); sr != nil; sr = result.Pull(ctx) {
+ for i := range sr.ElementIDs {
+ writeEntity := &streamv1.WriteRequest{
+ Metadata: s.Metadata,
+ Element: &streamv1.ElementValue{},
+ }
+ ev := writeEntity.Element
+ ev.ElementId = base64.StdEncoding.EncodeToString(convert.Uint64ToBytes(sr.ElementIDs[i]))
+ ev.Timestamp = timestamppb.New(time.Unix(0, sr.Timestamps[i]))
+ for _, tf := range sr.TagFamilies {
+ tfw := &modelv1.TagFamilyForWrite{}
+ for _, tag := range tf.Tags {
+ tfw.Tags = append(tfw.Tags, tag.Values[i])
+ }
+ ev.TagFamilies = append(ev.TagFamilies, tfw)
+ }
+ entity, tagValues, shardID, err := entityLocator.Locate(s.Metadata.Name, ev.TagFamilies, shardNum)
+ if err != nil {
+ l.Error().Err(err).Msg("failed to locate entity")
+ continue
+ }
+ nodeID, err := selector.Pick(s.Metadata.Group, s.Metadata.Name, uint32(shardID))
+ if err != nil {
+ l.Error().Err(err).Msg("failed to pick node")
+ continue
+ }
+ iwr := &streamv1.InternalWriteRequest{
+ Request: writeEntity,
+ ShardId: uint32(shardID),
+ SeriesHash: pbv1.HashEntity(entity),
+ EntityValues: tagValues[1:].Encode(),
+ }
+ message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
+ _, err = batch.Publish(ctx, data.TopicStreamWrite, message)
+ if err != nil {
+ l.Error().Err(err).Msg("failed to publish message")
+ }
+ }
+ }
+}
+
+func migrateMeasure(ctx context.Context, m *databasev1.Measure, result model.MeasureQueryResult,
+ shardNum uint32, selector node.Selector, client queue.Client, l *logger.Logger,
+) {
+ defer result.Release()
+
+ entityLocator := partition.NewEntityLocator(m.TagFamilies, m.Entity, 0)
+
+ batch := client.NewBatchPublisher(30 * time.Second)
+ defer batch.Close()
+ for mr := result.Pull(); mr != nil; mr = result.Pull() {
+ for i := range mr.Timestamps {
+ writeRequest := &measurev1.WriteRequest{
+ Metadata: m.Metadata,
+ DataPoint: &measurev1.DataPointValue{
+ Timestamp: timestamppb.New(time.Unix(0, mr.Timestamps[i])),
+ },
+ MessageId: uint64(time.Now().UnixNano()),
+ }
+
+ for _, tf := range mr.TagFamilies {
+ tfWrite := &modelv1.TagFamilyForWrite{}
+ for _, tag := range tf.Tags {
+ tfWrite.Tags = append(tfWrite.Tags, tag.Values[i])
+ }
+ writeRequest.DataPoint.TagFamilies = append(writeRequest.DataPoint.TagFamilies, tfWrite)
+ }
+
+ for _, field := range mr.Fields {
+ writeRequest.DataPoint.Fields = append(writeRequest.DataPoint.Fields, field.Values[i])
+ }
+
+ entity, tagValues, shardID, err := entityLocator.Locate(m.Metadata.Name, writeRequest.DataPoint.TagFamilies, shardNum)
+ if err != nil {
+ l.Error().Err(err).Msg("failed to locate entity")
+ continue
+ }
+
+ nodeID, err := selector.Pick(m.Metadata.Group, m.Metadata.Name, uint32(shardID))
+ if err != nil {
+ l.Error().Err(err).Msg("failed to pick node")
+ continue
+ }
+
+ iwr := &measurev1.InternalWriteRequest{
+ Request: writeRequest,
+ ShardId: uint32(shardID),
+ SeriesHash: pbv1.HashEntity(entity),
+ EntityValues: tagValues[1:].Encode(),
+ }
+
+ message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
+ _, err = batch.Publish(ctx, data.TopicMeasureWrite, message)
+ if err != nil {
+ l.Error().Err(err).Msg("failed to publish message")
+ }
+ }
+ }
+}
diff --git a/banyand/backup/restore.go b/banyand/backup/restore.go
index fab184c9..1bd66666 100644
--- a/banyand/backup/restore.go
+++ b/banyand/backup/restore.go
@@ -30,6 +30,7 @@ import (
"go.uber.org/multierr"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/fs/remote"
@@ -129,19 +130,19 @@ func newRunCommand() *cobra.Command {
}
func restoreCatalog(fs remote.FS, timeDir, rootPath string, catalog commonv1.Catalog) error {
- remotePrefix := filepath.Join(timeDir, getCatalogName(catalog), "/")
+ remotePrefix := filepath.Join(timeDir, snapshot.CatalogName(catalog), "/")
remoteFiles, err := fs.List(context.Background(), remotePrefix)
if err != nil {
return fmt.Errorf("failed to list remote files: %w", err)
}
- localDir := filepath.Join(getLocalDir(rootPath, catalog), storage.DataDir)
+ localDir := filepath.Join(snapshot.LocalDir(rootPath, catalog), storage.DataDir)
if err = os.MkdirAll(localDir, storage.DirPerm); err != nil {
return fmt.Errorf("failed to create local directory %s: %w", localDir, err)
}
- logger.Infof("Restoring %s to %s from %s", getCatalogName(catalog), localDir, remotePrefix)
+ logger.Infof("Restoring %s to %s from %s", snapshot.CatalogName(catalog), localDir, remotePrefix)
remoteRelSet := make(map[string]bool)
var relPath string
@@ -205,24 +206,6 @@ func cleanEmptyDirs(dir, stopDir string) {
}
}
-func getLocalDir(rootPath string, catalog commonv1.Catalog) string {
- return filepath.Join(rootPath, getCatalogName(catalog))
-}
-
-func getCatalogName(catalog commonv1.Catalog) string {
- switch catalog {
- case commonv1.Catalog_CATALOG_STREAM:
- return "stream"
- case commonv1.Catalog_CATALOG_MEASURE:
- return "measure"
- case commonv1.Catalog_CATALOG_PROPERTY:
- return "property"
- default:
- logger.Panicf("unknown catalog type: %v", catalog)
- return ""
- }
-}
-
func downloadFile(ctx context.Context, fs remote.FS, remotePath, localPath string) error {
reader, err := fs.Download(ctx, remotePath)
if err != nil {
diff --git a/banyand/backup/restore_test.go b/banyand/backup/restore_test.go
index 830595af..0feb891b 100644
--- a/banyand/backup/restore_test.go
+++ b/banyand/backup/restore_test.go
@@ -25,6 +25,7 @@ import (
"testing"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/fs/remote/local"
)
@@ -39,7 +40,7 @@ func TestRestoreDownload(t *testing.T) {
}
timeDir := "2023-10-10"
- remoteFilePath := filepath.Join(timeDir, getCatalogName(commonv1.Catalog_CATALOG_STREAM), storage.DataDir, "test.txt")
+ remoteFilePath := filepath.Join(timeDir, snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM), storage.DataDir, "test.txt")
content := "hello"
err = fs.Upload(context.Background(), remoteFilePath, strings.NewReader(content))
if err != nil {
@@ -51,7 +52,7 @@ func TestRestoreDownload(t *testing.T) {
t.Fatalf("restoreCatalog failed: %v", err)
}
- localFilePath := filepath.Join(localRestoreDir, getCatalogName(commonv1.Catalog_CATALOG_STREAM), storage.DataDir, "test.txt")
+ localFilePath := filepath.Join(localRestoreDir, snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM), storage.DataDir, "test.txt")
got, err := os.ReadFile(localFilePath)
if err != nil {
t.Fatalf("failed to read local file: %v", err)
diff --git a/banyand/backup/snapshot/snapshot.go b/banyand/backup/snapshot/snapshot.go
new file mode 100644
index 00000000..edcd17b5
--- /dev/null
+++ b/banyand/backup/snapshot/snapshot.go
@@ -0,0 +1,100 @@
+// 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 snapshot provides the snapshot backup and restore functionality.
+package snapshot
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "path/filepath"
+ "time"
+
+ "google.golang.org/grpc"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// Conn connects to the gRPC server and executes the given function.
+func Conn[T any](gRPCAddr string, enableTLS, insecure bool, cert string, delFn func(conn *grpc.ClientConn) (T, error)) (T, error) {
+ opts, err := grpchelper.SecureOptions(nil, enableTLS, insecure, cert)
+ if err != nil {
+ var zero T
+ return zero, err
+ }
+ connection, err := grpchelper.Conn(gRPCAddr, 10*time.Second, opts...)
+ if err != nil {
+ var zero T
+ return zero, fmt.Errorf("failed to connect to gRPC server: %w", err)
+ }
+ defer connection.Close()
+ return delFn(connection)
+}
+
+// Get retrieves the snapshots from the gRPC server.
+func Get(gRPCAddr string, enableTLS, insecure bool, cert string, groups ...*databasev1.SnapshotRequest_Group) ([]*databasev1.Snapshot, error) {
+ return Conn(gRPCAddr, enableTLS, insecure, cert, func(conn *grpc.ClientConn) ([]*databasev1.Snapshot, error) {
+ ctx := context.Background()
+ client := databasev1.NewSnapshotServiceClient(conn)
+ snapshotResp, err := client.Snapshot(ctx, &databasev1.SnapshotRequest{Groups: groups})
+ if err != nil {
+ return nil, fmt.Errorf("failed to request snapshot: %w", err)
+ }
+ return snapshotResp.Snapshots, nil
+ })
+}
+
+// Dir returns the directory path of the snapshot.
+func Dir(snapshot *databasev1.Snapshot, streamRoot, measureRoot, propertyRoot string) (string, error) {
+ var baseDir string
+ switch snapshot.Catalog {
+ case commonv1.Catalog_CATALOG_STREAM:
+ baseDir = LocalDir(streamRoot, snapshot.Catalog)
+ case commonv1.Catalog_CATALOG_MEASURE:
+ baseDir = LocalDir(measureRoot, snapshot.Catalog)
+ case commonv1.Catalog_CATALOG_PROPERTY:
+ baseDir = LocalDir(propertyRoot, snapshot.Catalog)
+ default:
+ return "", errors.New("unknown catalog type")
+ }
+ return filepath.Join(baseDir, storage.SnapshotsDir, snapshot.Name), nil
+}
+
+// LocalDir returns the local directory path of the snapshot.
+func LocalDir(rootPath string, catalog commonv1.Catalog) string {
+ return filepath.Join(rootPath, CatalogName(catalog))
+}
+
+// CatalogName returns the name of the catalog.
+func CatalogName(catalog commonv1.Catalog) string {
+ switch catalog {
+ case commonv1.Catalog_CATALOG_STREAM:
+ return "stream"
+ case commonv1.Catalog_CATALOG_MEASURE:
+ return "measure"
+ case commonv1.Catalog_CATALOG_PROPERTY:
+ return "property"
+ default:
+ logger.Panicf("unknown catalog type: %v", catalog)
+ return ""
+ }
+}
diff --git a/banyand/internal/storage/index.go b/banyand/internal/storage/index.go
index 5a62bf83..f0ed91f0 100644
--- a/banyand/internal/storage/index.go
+++ b/banyand/internal/storage/index.go
@@ -91,6 +91,9 @@ func (s *seriesIndex) Update(docs index.Documents) error {
func (s *seriesIndex) filter(ctx context.Context, series []*pbv1.Series,
projection []index.FieldKey, secondaryQuery index.Query, timeRange *timestamp.TimeRange,
) (data SeriesData, err error) {
+ if len(series) == 0 && secondaryQuery == nil {
+ return data, nil
+ }
var seriesMatchers []index.SeriesMatcher
if len(series) > 0 {
seriesMatchers = make([]index.SeriesMatcher, len(series))
diff --git a/banyand/internal/storage/rotation.go b/banyand/internal/storage/rotation.go
index 81f242e7..e8c6df9f 100644
--- a/banyand/internal/storage/rotation.go
+++ b/banyand/internal/storage/rotation.go
@@ -44,14 +44,19 @@ func (d *database[T, O]) Tick(ts int64) {
func (d *database[T, O]) startRotationTask() error {
options := d.segmentController.getOptions()
- rt := newRetentionTask(d, options.TTL)
+ var rt *retentionTask[T, O]
+ if !d.disableRetention {
+ rt = newRetentionTask(d, options.TTL)
+ }
go func(rt *retentionTask[T, O]) {
for ts := range d.tsEventCh {
func(ts int64) {
d.rotationProcessOn.Store(true)
defer d.rotationProcessOn.Store(false)
t := time.Unix(0, ts)
- rt.run(t, d.logger)
+ if rt != nil {
+ rt.run(t, d.logger)
+ }
func() {
ss := d.segmentController.segments()
if len(ss) == 0 {
@@ -87,6 +92,9 @@ func (d *database[T, O]) startRotationTask() error {
}(ts)
}
}(rt)
+ if rt == nil {
+ return nil
+ }
return d.scheduler.Register("retention", rt.option, rt.expr, rt.run)
}
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index dfbae4dc..b96da252 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -462,6 +462,41 @@ func (sc *segmentController[T, O]) remove(deadline time.Time) (hasSegment bool,
return hasSegment, err
}
+func (sc *segmentController[T, O]) getExpiredSegmentsTimeRange() *timestamp.TimeRange {
+ deadline := time.Now().Local().Add(-sc.opts.TTL.estimatedDuration())
+ timeRange := ×tamp.TimeRange{
+ IncludeStart: true,
+ IncludeEnd: false,
+ }
+ for _, s := range sc.segments() {
+ if s.Before(deadline) {
+ if timeRange.Start.IsZero() {
+ timeRange.Start = s.Start
+ }
+ timeRange.End = s.End
+ }
+ s.DecRef()
+ }
+ return timeRange
+}
+
+func (sc *segmentController[T, O]) deleteExpiredSegments(timeRange timestamp.TimeRange) int64 {
+ deadline := time.Now().Local().Add(-sc.opts.TTL.estimatedDuration())
+ var count int64
+ for _, s := range sc.segments() {
+ if s.Before(deadline) && s.Overlapping(timeRange) {
+ s.delete()
+ sc.Lock()
+ sc.removeSeg(s.id)
+ sc.Unlock()
+ count++
+ }
+ s.DecRef()
+ }
+ sc.notifySegmentBoundaryUpdate()
+ return count
+}
+
func (sc *segmentController[T, O]) removeSeg(segID segmentID) {
for i, b := range sc.lst {
if b.id == segID {
diff --git a/banyand/internal/storage/storage.go b/banyand/internal/storage/storage.go
index b0c2c269..3e8c148a 100644
--- a/banyand/internal/storage/storage.go
+++ b/banyand/internal/storage/storage.go
@@ -109,6 +109,8 @@ type TSDB[T TSTable, O any] interface {
Tick(ts int64)
UpdateOptions(opts *commonv1.ResourceOpts)
TakeFileSnapshot(dst string) error
+ GetExpiredSegmentsTimeRange() *timestamp.TimeRange
+ DeleteExpiredSegments(timeRange timestamp.TimeRange) int64
}
// Segment is a time range of data.
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index 67ae1280..b850ec42 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -66,6 +66,7 @@ type TSDBOpts[T TSTable, O any] struct {
SeriesIndexFlushTimeoutSeconds int64
SeriesIndexCacheMaxBytes int
ShardNum uint32
+ DisableRetention bool
}
type (
@@ -89,6 +90,7 @@ type database[T TSTable, O any] struct {
sync.RWMutex
rotationProcessOn atomic.Bool
closed atomic.Bool
+ disableRetention bool
}
func (d *database[T, O]) Close() error {
@@ -136,7 +138,8 @@ func OpenTSDB[T TSTable, O any](ctx context.Context, opts TSDBOpts[T, O]) (TSDB[
p: p,
segmentController: newSegmentController[T](ctx, location,
l, opts, indexMetrics, opts.TableMetrics, opts.SegmentBoundaryUpdateFn),
- metrics: newMetrics(opts.StorageMetricsFactory),
+ metrics: newMetrics(opts.StorageMetricsFactory),
+ disableRetention: opts.DisableRetention,
}
db.logger.Info().Str("path", opts.Location).Msg("initialized")
lockPath := filepath.Join(opts.Location, lockFilename)
@@ -213,6 +216,14 @@ func (d *database[T, O]) TakeFileSnapshot(dst string) error {
return nil
}
+func (d *database[T, O]) GetExpiredSegmentsTimeRange() *timestamp.TimeRange {
+ return d.segmentController.getExpiredSegmentsTimeRange()
+}
+
+func (d *database[T, O]) DeleteExpiredSegments(timeRange timestamp.TimeRange) int64 {
+ return d.segmentController.deleteExpiredSegments(timeRange)
+}
+
func (d *database[T, O]) collect() {
if d.closed.Load() {
return
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 1c37d41e..c9a5f168 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -39,8 +39,10 @@ import (
"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/queue/pub"
"github.com/apache/skywalking-banyandb/pkg/logger"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
const (
@@ -79,7 +81,7 @@ type schemaRepo struct {
path string
}
-func newSchemaRepo(path string, svc *service) *schemaRepo {
+func newSchemaRepo(path string, svc *service, nodeLabels map[string]string) *schemaRepo {
sr := &schemaRepo{
path: path,
l: svc.l,
@@ -89,7 +91,7 @@ func newSchemaRepo(path string, svc *service) *schemaRepo {
sr.Repository = resourceSchema.NewRepository(
svc.metadata,
svc.l,
- newSupplier(path, svc, sr),
+ newSupplier(path, svc, sr, nodeLabels),
resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
)
sr.start()
@@ -127,6 +129,18 @@ func (sr *schemaRepo) Measure(metadata *commonv1.Metadata) (Measure, error) {
return sm, nil
}
+func (sr *schemaRepo) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ g, ok := sr.LoadGroup(group)
+ if !ok {
+ return nil
+ }
+ db := g.SupplyTSDB()
+ if db == nil {
+ return nil
+ }
+ return db.(storage.TSDB[*tsTable, option]).GetExpiredSegmentsTimeRange()
+}
+
func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
if len(kinds) != 5 {
logger.Panicf("unexpected kinds: %v", kinds)
@@ -325,11 +339,12 @@ type supplier struct {
l *logger.Logger
pm *protector.Memory
schemaRepo *schemaRepo
+ nodeLabels map[string]string
path string
option option
}
-func newSupplier(path string, svc *service, sr *schemaRepo) *supplier {
+func newSupplier(path string, svc *service, sr *schemaRepo, nodeLabels map[string]string) *supplier {
return &supplier{
path: path,
metadata: svc.metadata,
@@ -338,6 +353,7 @@ func newSupplier(path string, svc *service, sr *schemaRepo) *supplier {
omr: svc.omr,
pm: svc.pm,
schemaRepo: sr,
+ nodeLabels: nodeLabels,
}
}
@@ -361,13 +377,44 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
Database: name,
}
metrics, factory := s.newMetrics(p)
+ ro := groupSchema.ResourceOpts
+ if ro == nil {
+ return nil, fmt.Errorf("no resource opts in group %s", name)
+ }
+ shardNum := ro.ShardNum
+ ttl := ro.Ttl
+ segInterval := ro.SegmentInterval
+ if len(ro.Stages) > 0 && len(s.nodeLabels) > 0 {
+ matched := false
+ var ttlNum uint32
+ for _, st := range ro.Stages {
+ if st.Ttl.Unit != ro.Ttl.Unit {
+ return nil, fmt.Errorf("ttl unit %s is not consistent with stage %s", ro.Ttl.Unit, st.Ttl.Unit)
+ }
+ selector, err := pub.ParseLabelSelector(st.NodeSelector)
+ if err != nil {
+ return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
+ }
+ ttlNum += st.Ttl.Num
+ if !selector.Matches(s.nodeLabels) {
+ continue
+ }
+ matched = true
+ shardNum = st.ShardNum
+ segInterval = st.SegmentInterval
+ break
+ }
+ if matched {
+ ttl.Num += ttlNum
+ }
+ }
opts := storage.TSDBOpts[*tsTable, option]{
- ShardNum: groupSchema.ResourceOpts.ShardNum,
+ ShardNum: shardNum,
Location: path.Join(s.path, groupSchema.Metadata.Name),
TSTableCreator: newTSTable,
TableMetrics: metrics,
- SegmentInterval: storage.MustToIntervalRule(groupSchema.ResourceOpts.SegmentInterval),
- TTL: storage.MustToIntervalRule(groupSchema.ResourceOpts.Ttl),
+ SegmentInterval: storage.MustToIntervalRule(segInterval),
+ TTL: storage.MustToIntervalRule(ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index 8ba6b13a..9fc92260 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -38,6 +38,7 @@ import (
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/model"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
const (
@@ -51,6 +52,7 @@ var nilResult = model.MeasureQueryResult(nil)
type Query interface {
LoadGroup(name string) (resourceSchema.Group, bool)
Measure(measure *commonv1.Metadata) (Measure, error)
+ GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange
}
// Measure allows inspecting measure data points' details.
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index c67101f8..b6ac59b4 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -21,12 +21,15 @@ import (
"context"
"path"
"path/filepath"
+ "time"
"github.com/pkg/errors"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -37,6 +40,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
var (
@@ -85,6 +89,10 @@ func (s *service) LoadGroup(name string) (resourceSchema.Group, bool) {
return s.schemaRepo.LoadGroup(name)
}
+func (s *service) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ return s.schemaRepo.GetRemovalSegmentsTimeRange(group)
+}
+
func (s *service) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&s.root, "measure-root-path", "/tmp", "the root path of database")
@@ -126,7 +134,15 @@ func (s *service) PreRun(ctx context.Context) error {
s.snapshotDir = filepath.Join(path, storage.SnapshotsDir)
observability.UpdatePath(path)
s.localPipeline = queue.Local()
- s.schemaRepo = newSchemaRepo(filepath.Join(path, storage.DataDir), s)
+ val := ctx.Value(common.ContextNodeKey)
+ if val == nil {
+ return errors.New("node id is empty")
+ }
+ node := val.(common.Node)
+ s.schemaRepo = newSchemaRepo(filepath.Join(path, storage.DataDir), s, node.Labels)
+ if s.pipeline == nil {
+ return nil
+ }
if err := s.createNativeObservabilityGroup(ctx); err != nil {
return err
@@ -136,6 +152,10 @@ func (s *service) PreRun(ctx context.Context) error {
return err
}
+ if err := s.pipeline.Subscribe(data.TopicDeleteExpiredStreamSegments, &deleteStreamSegmentsListener{s: s}); err != nil {
+ return err
+ }
+
s.writeListener = setUpWriteCallback(s.l, s.schemaRepo, s.maxDiskUsagePercent)
// only subscribe metricPipeline for data node
if s.metricPipeline != nil {
@@ -156,8 +176,10 @@ func (s *service) Serve() run.StopNotify {
}
func (s *service) GracefulStop() {
- s.localPipeline.GracefulStop()
s.schemaRepo.Close()
+ if s.localPipeline != nil {
+ s.localPipeline.GracefulStop()
+ }
}
// NewService returns a new service.
@@ -170,3 +192,32 @@ func NewService(metadata metadata.Repo, pipeline queue.Server, metricPipeline qu
pm: pm,
}, nil
}
+
+// NewReadonlyService returns a new readonly service.
+func NewReadonlyService(metadata metadata.Repo, omr observability.MetricsRegistry, pm *protector.Memory) (Service, error) {
+ return &service{
+ metadata: metadata,
+ omr: omr,
+ pm: pm,
+ }, nil
+}
+
+type deleteStreamSegmentsListener struct {
+ *bus.UnImplementedHealthyListener
+ s *service
+}
+
+func (d *deleteStreamSegmentsListener) Rev(_ context.Context, message bus.Message) bus.Message {
+ req := message.Data().(*measurev1.DeleteExpiredSegmentsRequest)
+ if req == nil {
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
+ }
+
+ db, err := d.s.schemaRepo.loadTSDB(req.Group)
+ if err != nil {
+ d.s.l.Error().Err(err).Str("group", req.Group).Msg("failed to load tsdb")
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
+ }
+ deleted := db.DeleteExpiredSegments(timestamp.NewSectionTimeRange(req.TimeRange.Begin.AsTime(), req.TimeRange.End.AsTime()))
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), deleted)
+}
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index 1f46a0ff..a46752f4 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -132,7 +132,7 @@ func (s *clientService) PreRun(ctx context.Context) error {
schema.ConfigureEtcdTLSCertAndKey(s.etcdTLSCertFile, s.etcdTLSKeyFile),
schema.ConfigureWatchCheckInterval(s.etcdFullSyncInterval),
)
- if errors.Is(err, context.DeadlineExceeded) {
+ if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
select {
case <-stopCh:
return errors.New("pre-run interrupted")
@@ -248,6 +248,10 @@ func (s *clientService) TopNAggregationRegistry() schema.TopNAggregation {
return s.schemaRegistry
}
+func (s *clientService) NodeRegistry() schema.Node {
+ return s.schemaRegistry
+}
+
func (s *clientService) Name() string {
return "metadata"
}
diff --git a/banyand/metadata/metadata.go b/banyand/metadata/metadata.go
index e0b23fa0..9b1be0f9 100644
--- a/banyand/metadata/metadata.go
+++ b/banyand/metadata/metadata.go
@@ -48,6 +48,7 @@ type Repo interface {
TopNAggregationRegistry() schema.TopNAggregation
RegisterHandler(string, schema.Kind, schema.EventHandler)
UpdateSegmentsBoundary(stage, group string, boundary *modelv1.TimeRange)
+ NodeRegistry() schema.Node
}
// Service is the metadata repository.
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 20bbeb4a..b91a8eb0 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -78,6 +78,10 @@ func (l *local) Broadcast(timeout time.Duration, topic bus.Topic, message bus.Me
return []bus.Future{f}, nil
}
+func (l *local) OnAddOrUpdate(_ schema.Metadata) {
+ panic("unimplemented")
+}
+
func (l local) Name() string {
return "local-pipeline"
}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 98208dfc..a55995b7 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -110,11 +110,11 @@ func (p *pub) Broadcast(timeout time.Duration, topic bus.Topic, messages bus.Mes
if sel == "" {
matches = bypassMatches
} else {
- selector, err := parseLabelSelector(sel)
+ selector, err := ParseLabelSelector(sel)
if err != nil {
return nil, fmt.Errorf("failed to parse node selector: %w", err)
}
- matches = selector.matches
+ matches = selector.Matches
}
for _, n := range nodes {
tr := metadata.FindSegmentsBoundary(n, g)
@@ -225,6 +225,7 @@ func New(metadata metadata.Repo) queue.Client {
registered: make(map[string]*databasev1.Node),
handlers: make(map[bus.Topic]schema.EventHandler),
closer: run.NewCloser(1),
+ log: logger.GetLogger("server-queue-pub"),
}
}
@@ -233,7 +234,6 @@ func (*pub) Name() string {
}
func (p *pub) PreRun(context.Context) error {
- p.log = logger.GetLogger("server-queue-pub")
p.metadata.RegisterHandler("queue-client", schema.KindNode, p)
return nil
}
diff --git a/banyand/queue/pub/selector.go b/banyand/queue/pub/selector.go
index b49edac1..df6f2895 100644
--- a/banyand/queue/pub/selector.go
+++ b/banyand/queue/pub/selector.go
@@ -24,7 +24,8 @@ import (
"unicode"
)
-type labelSelector struct {
+// LabelSelector is a selector for labels.
+type LabelSelector struct {
criteria []condition
}
@@ -45,9 +46,10 @@ const (
opNotExists
)
-func parseLabelSelector(selector string) (*labelSelector, error) {
+// ParseLabelSelector parses a label selector string.
+func ParseLabelSelector(selector string) (*LabelSelector, error) {
if selector == "" {
- return &labelSelector{}, nil
+ return &LabelSelector{}, nil
}
var conditions []condition
@@ -64,7 +66,7 @@ func parseLabelSelector(selector string) (*labelSelector, error) {
conditions = append(conditions, cond)
}
- return &labelSelector{conditions}, nil
+ return &LabelSelector{conditions}, nil
}
func splitSelector(selector string) []string {
@@ -185,7 +187,8 @@ func validateLabelKey(key string) error {
return nil
}
-func (s *labelSelector) matches(labels map[string]string) bool {
+// Matches returns true if the labels match the selector.
+func (s *LabelSelector) Matches(labels map[string]string) bool {
for _, req := range s.criteria {
if !req.matches(labels) {
return false
diff --git a/banyand/queue/pub/selector_test.go b/banyand/queue/pub/selector_test.go
index 382f4886..104374e0 100644
--- a/banyand/queue/pub/selector_test.go
+++ b/banyand/queue/pub/selector_test.go
@@ -23,7 +23,7 @@ import (
func TestParseLabelSelector(t *testing.T) {
tests := []struct {
- want *labelSelector
+ want *LabelSelector
name string
input string
wantErr bool
@@ -31,12 +31,12 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "empty selector",
input: "",
- want: &labelSelector{},
+ want: &LabelSelector{},
},
{
name: "single equality",
input: "key=value",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opEquals, Values: []string{"value"}},
},
@@ -45,7 +45,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "not equals",
input: "key != value",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opNotEquals, Values: []string{"value"}},
},
@@ -54,7 +54,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "in clause",
input: "key in (v1, v2)",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opIn, Values: []string{"v1", "v2"}},
},
@@ -63,7 +63,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "not in clause",
input: "key notin (v1,v2)",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opNotIn, Values: []string{"v1", "v2"}},
},
@@ -72,7 +72,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "exists",
input: "key",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opExists},
},
@@ -81,7 +81,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "not exists",
input: "!key",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key", Op: opNotExists},
},
@@ -90,7 +90,7 @@ func TestParseLabelSelector(t *testing.T) {
{
name: "multiple conditions",
input: "key1=value1,key2 in (v1, v2),!key3",
- want: &labelSelector{
+ want: &LabelSelector{
criteria: []condition{
{Key: "key1", Op: opEquals, Values: []string{"value1"}},
{Key: "key2", Op: opIn, Values: []string{"v1", "v2"}},
@@ -122,7 +122,7 @@ func TestParseLabelSelector(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := parseLabelSelector(tt.input)
+ got, err := ParseLabelSelector(tt.input)
if (err != nil) != tt.wantErr {
t.Fatalf("parseLabelSelector() error = %v, wantErr %v", err, tt.wantErr)
}
@@ -293,14 +293,14 @@ func TestConditionMatches(t *testing.T) {
func TestLabelSelectorMatches(t *testing.T) {
tests := []struct {
- selector *labelSelector
+ selector *LabelSelector
labels map[string]string
name string
want bool
}{
{
name: "all conditions met",
- selector: &labelSelector{
+ selector: &LabelSelector{
criteria: []condition{
{Key: "k1", Op: opExists},
{Key: "k2", Op: opEquals, Values: []string{"v2"}},
@@ -316,7 +316,7 @@ func TestLabelSelectorMatches(t *testing.T) {
},
{
name: "one condition fails",
- selector: &labelSelector{
+ selector: &LabelSelector{
criteria: []condition{
{Key: "k1", Op: opExists},
{Key: "k2", Op: opEquals, Values: []string{"v2"}},
@@ -330,7 +330,7 @@ func TestLabelSelectorMatches(t *testing.T) {
},
{
name: "empty selector",
- selector: &labelSelector{},
+ selector: &LabelSelector{},
labels: map[string]string{"any": "value"},
want: true,
},
@@ -338,7 +338,7 @@ func TestLabelSelectorMatches(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got := tt.selector.matches(tt.labels)
+ got := tt.selector.Matches(tt.labels)
if got != tt.want {
t.Errorf("matches() = %v, want %v", got, tt.want)
}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index f447d6f3..c55bf0ec 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -44,6 +44,7 @@ type Client interface {
bus.Broadcaster
NewBatchPublisher(timeout time.Duration) BatchPublisher
Register(bus.Topic, schema.EventHandler)
+ OnAddOrUpdate(md schema.Metadata)
}
// Server is the interface for receiving data from the queue.
diff --git a/banyand/queue/sub/segments.go b/banyand/queue/sub/segments.go
new file mode 100644
index 00000000..02b27856
--- /dev/null
+++ b/banyand/queue/sub/segments.go
@@ -0,0 +1,82 @@
+// 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 sub
+
+import (
+ "context"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+type streamService struct {
+ streamv1.UnimplementedStreamServiceServer
+ ser *server
+}
+
+func (s *streamService) DeleteExpiredSegments(ctx context.Context, request *streamv1.DeleteExpiredSegmentsRequest) (*streamv1.DeleteExpiredSegmentsResponse, error) {
+ s.ser.listenersLock.RLock()
+ defer s.ser.listenersLock.RUnlock()
+ ll := s.ser.getListeners(data.TopicDeleteExpiredStreamSegments)
+ if len(ll) == 0 {
+ logger.Panicf("no listener found for topic %s", data.TopicDeleteExpiredStreamSegments)
+ }
+ var deleted int64
+ for _, l := range ll {
+ message := l.Rev(ctx, bus.NewMessage(bus.MessageID(0), request))
+ data := message.Data()
+ if data != nil {
+ d, ok := data.(int64)
+ if !ok {
+ logger.Panicf("invalid data type %T", data)
+ }
+ deleted += d
+ }
+ }
+ return &streamv1.DeleteExpiredSegmentsResponse{Deleted: deleted}, nil
+}
+
+type measureService struct {
+ measurev1.UnimplementedMeasureServiceServer
+ ser *server
+}
+
+func (s *measureService) DeleteExpiredSegments(ctx context.Context, request *measurev1.DeleteExpiredSegmentsRequest) (*measurev1.DeleteExpiredSegmentsResponse, error) {
+ s.ser.listenersLock.RLock()
+ defer s.ser.listenersLock.RUnlock()
+ ll := s.ser.getListeners(data.TopicMeasureDeleteExpiredSegments)
+ if len(ll) == 0 {
+ logger.Panicf("no listener found for topic %s", data.TopicMeasureDeleteExpiredSegments)
+ }
+ var deleted int64
+ for _, l := range ll {
+ message := l.Rev(ctx, bus.NewMessage(bus.MessageID(0), request))
+ data := message.Data()
+ if data != nil {
+ d, ok := data.(int64)
+ if !ok {
+ logger.Panicf("invalid data type %T", data)
+ }
+ deleted += d
+ }
+ }
+ return &measurev1.DeleteExpiredSegmentsResponse{Deleted: deleted}, nil
+}
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
index c0dcc12c..d669aea0 100644
--- a/banyand/queue/sub/server.go
+++ b/banyand/queue/sub/server.go
@@ -41,6 +41,8 @@ import (
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
@@ -65,6 +67,7 @@ var (
type server struct {
databasev1.UnimplementedSnapshotServiceServer
+ streamv1.UnimplementedStreamServiceServer
creds credentials.TransportCredentials
omr observability.MetricsRegistry
httpSrv *http.Server
@@ -180,6 +183,8 @@ func (s *server) Serve() run.StopNotify {
clusterv1.RegisterServiceServer(s.ser, s)
grpc_health_v1.RegisterHealthServer(s.ser, health.NewServer())
databasev1.RegisterSnapshotServiceServer(s.ser, s)
+ streamv1.RegisterStreamServiceServer(s.ser, &streamService{ser: s})
+ measurev1.RegisterMeasureServiceServer(s.ser, &measureService{ser: s})
var ctx context.Context
ctx, s.clientCloser = context.WithCancel(context.Background())
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 7f5bd596..4891b373 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -36,9 +36,11 @@ import (
"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/queue/pub"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/meter"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
var metadataScope = streamScope.SubScope("metadata")
@@ -55,7 +57,7 @@ type schemaRepo struct {
path string
}
-func newSchemaRepo(path string, svc *service) schemaRepo {
+func newSchemaRepo(path string, svc *service, nodeLabels map[string]string) schemaRepo {
sr := schemaRepo{
l: svc.l,
path: path,
@@ -63,7 +65,7 @@ func newSchemaRepo(path string, svc *service) schemaRepo {
Repository: resourceSchema.NewRepository(
svc.metadata,
svc.l,
- newSupplier(path, svc),
+ newSupplier(path, svc, nodeLabels),
resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
),
}
@@ -102,6 +104,18 @@ func (sr *schemaRepo) Stream(metadata *commonv1.Metadata) (Stream, error) {
return sm, nil
}
+func (sr *schemaRepo) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ g, ok := sr.LoadGroup(group)
+ if !ok {
+ return nil
+ }
+ db := g.SupplyTSDB()
+ if db == nil {
+ return nil
+ }
+ return db.(storage.TSDB[*tsTable, option]).GetExpiredSegmentsTimeRange()
+}
+
func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
if len(kinds) != 4 {
logger.Panicf("invalid kinds: %v", kinds)
@@ -237,11 +251,12 @@ type supplier struct {
l *logger.Logger
pm *protector.Memory
schemaRepo *schemaRepo
+ nodeLabels map[string]string
path string
option option
}
-func newSupplier(path string, svc *service) *supplier {
+func newSupplier(path string, svc *service, nodeLabels map[string]string) *supplier {
return &supplier{
path: path,
metadata: svc.metadata,
@@ -251,6 +266,7 @@ func newSupplier(path string, svc *service) *supplier {
omr: svc.omr,
pm: svc.pm,
schemaRepo: &svc.schemaRepo,
+ nodeLabels: nodeLabels,
}
}
@@ -273,14 +289,44 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
Module: "stream",
Database: name,
}
-
+ ro := groupSchema.ResourceOpts
+ if ro == nil {
+ return nil, fmt.Errorf("no resource opts in group %s", name)
+ }
+ shardNum := ro.ShardNum
+ ttl := ro.Ttl
+ segInterval := ro.SegmentInterval
+ if len(ro.Stages) > 0 && len(s.nodeLabels) > 0 {
+ matched := false
+ var ttlNum uint32
+ for _, st := range ro.Stages {
+ if st.Ttl.Unit != ro.Ttl.Unit {
+ return nil, fmt.Errorf("ttl unit %s is not consistent with stage %s", ro.Ttl.Unit, st.Ttl.Unit)
+ }
+ selector, err := pub.ParseLabelSelector(st.NodeSelector)
+ if err != nil {
+ return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
+ }
+ ttlNum += st.Ttl.Num
+ if !selector.Matches(s.nodeLabels) {
+ continue
+ }
+ matched = true
+ shardNum = st.ShardNum
+ segInterval = st.SegmentInterval
+ break
+ }
+ if matched {
+ ttl.Num += ttlNum
+ }
+ }
opts := storage.TSDBOpts[*tsTable, option]{
- ShardNum: groupSchema.ResourceOpts.ShardNum,
+ ShardNum: shardNum,
Location: path.Join(s.path, groupSchema.Metadata.Name),
TSTableCreator: newTSTable,
TableMetrics: s.newMetrics(p),
- SegmentInterval: storage.MustToIntervalRule(groupSchema.ResourceOpts.SegmentInterval),
- TTL: storage.MustToIntervalRule(groupSchema.ResourceOpts.Ttl),
+ SegmentInterval: storage.MustToIntervalRule(segInterval),
+ TTL: storage.MustToIntervalRule(ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index 0c4fd4cd..9f4b0506 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -21,12 +21,15 @@ import (
"context"
"path"
"path/filepath"
+ "time"
"github.com/pkg/errors"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -37,6 +40,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
var (
@@ -84,6 +88,10 @@ func (s *service) LoadGroup(name string) (resourceSchema.Group, bool) {
return s.schemaRepo.LoadGroup(name)
}
+func (s *service) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+ return s.schemaRepo.GetRemovalSegmentsTimeRange(group)
+}
+
func (s *service) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&s.root, "stream-root-path", "/tmp", "the root path of database")
@@ -119,19 +127,29 @@ func (s *service) Role() databasev1.Role {
return databasev1.Role_ROLE_DATA
}
-func (s *service) PreRun(_ context.Context) error {
+func (s *service) PreRun(ctx context.Context) error {
s.l = logger.GetLogger(s.Name())
s.lfs = fs.NewLocalFileSystemWithLogger(s.l)
path := path.Join(s.root, s.Name())
s.snapshotDir = filepath.Join(path, storage.SnapshotsDir)
observability.UpdatePath(path)
- s.localPipeline = queue.Local()
- s.schemaRepo = newSchemaRepo(filepath.Join(path, storage.DataDir), s)
- // run a serial watcher
+ val := ctx.Value(common.ContextNodeKey)
+ if val == nil {
+ return errors.New("node id is empty")
+ }
+ node := val.(common.Node)
+ s.schemaRepo = newSchemaRepo(filepath.Join(path, storage.DataDir), s, node.Labels)
+ if s.pipeline == nil {
+ return nil
+ }
+ s.localPipeline = queue.Local()
if err := s.pipeline.Subscribe(data.TopicSnapshot, &snapshotListener{s: s}); err != nil {
return err
}
+ if err := s.pipeline.Subscribe(data.TopicDeleteExpiredStreamSegments, &deleteStreamSegmentsListener{s: s}); err != nil {
+ return err
+ }
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent)
err := s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
if err != nil {
@@ -145,8 +163,10 @@ func (s *service) Serve() run.StopNotify {
}
func (s *service) GracefulStop() {
- s.localPipeline.GracefulStop()
s.schemaRepo.Close()
+ if s.localPipeline != nil {
+ s.localPipeline.GracefulStop()
+ }
}
// NewService returns a new service.
@@ -158,3 +178,32 @@ func NewService(metadata metadata.Repo, pipeline queue.Server, omr observability
pm: pm,
}, nil
}
+
+// NewReadonlyService returns a new readonly service.
+func NewReadonlyService(metadata metadata.Repo, omr observability.MetricsRegistry, pm *protector.Memory) (Service, error) {
+ return &service{
+ metadata: metadata,
+ omr: omr,
+ pm: pm,
+ }, nil
+}
+
+type deleteStreamSegmentsListener struct {
+ *bus.UnImplementedHealthyListener
+ s *service
+}
+
+func (d *deleteStreamSegmentsListener) Rev(_ context.Context, message bus.Message) bus.Message {
+ req := message.Data().(*streamv1.DeleteExpiredSegmentsRequest)
+ if req == nil {
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
+ }
+
+ db, err := d.s.schemaRepo.loadTSDB(req.Group)
+ if err != nil {
+ d.s.l.Error().Err(err).Str("group", req.Group).Msg("failed to load tsdb")
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
+ }
+ deleted := db.DeleteExpiredSegments(timestamp.NewSectionTimeRange(req.TimeRange.Begin.AsTime(), req.TimeRange.End.AsTime()))
+ return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), deleted)
+}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index b3a4756d..89bb65b3 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/query/model"
"github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/schema"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
const (
@@ -55,6 +56,7 @@ type option struct {
type Query interface {
LoadGroup(name string) (schema.Group, bool)
Stream(stream *commonv1.Metadata) (Stream, error)
+ GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange
}
// Stream allows inspecting elements' details.
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 09e8eabb..a1cd9e60 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -202,6 +202,9 @@
- [DataPointValue.Type](#banyandb-measure-v1-DataPointValue-Type)
- [banyandb/measure/v1/rpc.proto](#banyandb_measure_v1_rpc-proto)
+ - [DeleteExpiredSegmentsRequest](#banyandb-measure-v1-DeleteExpiredSegmentsRequest)
+ - [DeleteExpiredSegmentsResponse](#banyandb-measure-v1-DeleteExpiredSegmentsResponse)
+
- [MeasureService](#banyandb-measure-v1-MeasureService)
- [banyandb/property/v1/property.proto](#banyandb_property_v1_property-proto)
@@ -235,6 +238,9 @@
- [WriteResponse](#banyandb-stream-v1-WriteResponse)
- [banyandb/stream/v1/rpc.proto](#banyandb_stream_v1_rpc-proto)
+ - [DeleteExpiredSegmentsRequest](#banyandb-stream-v1-DeleteExpiredSegmentsRequest)
+ - [DeleteExpiredSegmentsResponse](#banyandb-stream-v1-DeleteExpiredSegmentsResponse)
+
- [StreamService](#banyandb-stream-v1-StreamService)
- [Scalar Value Types](#scalar-value-types)
@@ -3000,6 +3006,37 @@ the type of the data point cumulative or delta
## banyandb/measure/v1/rpc.proto
+
+<a name="banyandb-measure-v1-DeleteExpiredSegmentsRequest"></a>
+
+### DeleteExpiredSegmentsRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| group | [string](#string) | | |
+| time_range | [banyandb.model.v1.TimeRange](#banyandb-model-v1-TimeRange) | | |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-DeleteExpiredSegmentsResponse"></a>
+
+### DeleteExpiredSegmentsResponse
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| deleted | [int64](#int64) | | |
+
+
+
+
+
@@ -3017,6 +3054,7 @@ the type of the data point cumulative or delta
| Query | [QueryRequest](#banyandb-measure-v1-QueryRequest) | [QueryResponse](#banyandb-measure-v1-QueryResponse) | |
| Write | [WriteRequest](#banyandb-measure-v1-WriteRequest) stream | [WriteResponse](#banyandb-measure-v1-WriteResponse) stream | |
| TopN | [TopNRequest](#banyandb-measure-v1-TopNRequest) | [TopNResponse](#banyandb-measure-v1-TopNResponse) | |
+| DeleteExpiredSegments | [DeleteExpiredSegmentsRequest](#banyandb-measure-v1-DeleteExpiredSegmentsRequest) | [DeleteExpiredSegmentsResponse](#banyandb-measure-v1-DeleteExpiredSegmentsResponse) | |
@@ -3428,6 +3466,37 @@ QueryResponse is the response for a query to the Query module.
## banyandb/stream/v1/rpc.proto
+
+<a name="banyandb-stream-v1-DeleteExpiredSegmentsRequest"></a>
+
+### DeleteExpiredSegmentsRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| group | [string](#string) | | |
+| time_range | [banyandb.model.v1.TimeRange](#banyandb-model-v1-TimeRange) | | |
+
+
+
+
+
+
+<a name="banyandb-stream-v1-DeleteExpiredSegmentsResponse"></a>
+
+### DeleteExpiredSegmentsResponse
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| deleted | [int64](#int64) | | |
+
+
+
+
+
@@ -3444,6 +3513,7 @@ QueryResponse is the response for a query to the Query module.
| ----------- | ------------ | ------------- | ------------|
| Query | [QueryRequest](#banyandb-stream-v1-QueryRequest) | [QueryResponse](#banyandb-stream-v1-QueryResponse) | |
| Write | [WriteRequest](#banyandb-stream-v1-WriteRequest) stream | [WriteResponse](#banyandb-stream-v1-WriteResponse) stream | |
+| DeleteExpiredSegments | [DeleteExpiredSegmentsRequest](#banyandb-stream-v1-DeleteExpiredSegmentsRequest) | [DeleteExpiredSegmentsResponse](#banyandb-stream-v1-DeleteExpiredSegmentsResponse) | |
diff --git a/pkg/index/inverted/inverted_series.go b/pkg/index/inverted/inverted_series.go
index 649779ff..93c95c88 100644
--- a/pkg/index/inverted/inverted_series.go
+++ b/pkg/index/inverted/inverted_series.go
@@ -393,6 +393,7 @@ func (s *store) SeriesIterator(ctx context.Context) (index.FieldIterator[index.S
defer func() {
_ = reader.Close()
}()
+
dict, err := reader.DictionaryIterator(docIDField, nil, nil, nil)
if err != nil {
return nil, err