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/02/11 00:08:52 UTC
(skywalking-banyandb) branch backup created (now 29920fb0)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch backup
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 29920fb0 Add backup CLI tool
This branch includes the following new commits:
new 3715a8e3 Add backup command line tool
new 29920fb0 Add backup CLI tool
The 2 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/02: Add backup command line tool
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch backup
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 3715a8e37dd51956599a3b49169170f76ea87904
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Feb 10 20:30:34 2025 +0800
Add backup command line tool
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
Makefile | 2 +-
banyand/internal/storage/snapshot.go | 44 +++++
banyand/internal/storage/snapshot_test.go | 62 +++++++
banyand/measure/service.go | 9 +-
banyand/measure/snapshot.go | 1 +
banyand/property/listener.go | 2 +
banyand/property/service.go | 5 +
banyand/stream/service.go | 7 +-
banyand/stream/snapshot.go | 1 +
bydbbackup/Makefile | 34 ++++
bydbbackup/cmd/backup/main.go | 34 ++++
bydbbackup/internal/backup/backup.go | 292 ++++++++++++++++++++++++++++++
bydbbackup/internal/backup/backup_test.go | 193 ++++++++++++++++++++
bydbctl/internal/cmd/health_check.go | 30 +--
pkg/fs/remote/local/local.go | 98 ++++++++++
pkg/fs/remote/remote.go | 49 +++++
pkg/grpchelper/client.go | 32 ++++
18 files changed, 866 insertions(+), 30 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 220e5d65..062785e0 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -26,6 +26,7 @@ Release Notes.
- Test: Add the e2e test for zipkin.
- Test: Limit the CPU and memory usage of the e2e test.
- Add taking the snapshot of data files.
+- Add backup command line tool to backup the data files.
### Bug Fixes
diff --git a/Makefile b/Makefile
index 0d6a1747..66974133 100644
--- a/Makefile
+++ b/Makefile
@@ -27,7 +27,7 @@ endif
include scripts/build/version.mk
-PROJECTS := ui banyand bydbctl
+PROJECTS := ui banyand bydbctl bydbbackup
TEST_CI_OPTS ?=
diff --git a/banyand/internal/storage/snapshot.go b/banyand/internal/storage/snapshot.go
new file mode 100644
index 00000000..4cd3e28c
--- /dev/null
+++ b/banyand/internal/storage/snapshot.go
@@ -0,0 +1,44 @@
+// 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 storage
+
+import (
+ "path/filepath"
+ "sort"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// DeleteStaleSnapshots deletes the stale snapshots in the root directory.
+func DeleteStaleSnapshots(root string, maxNum int, lfs fs.FileSystem) {
+ if maxNum <= 0 {
+ return
+ }
+ lfs.MkdirIfNotExist(root, dirPerm)
+ snapshots := lfs.ReadDir(root)
+ if len(snapshots) <= maxNum {
+ return
+ }
+ // sort by snapshot name whose format is "20060102150405-00000000"
+ sort.Slice(snapshots, func(i, j int) bool {
+ return snapshots[i].Name() < snapshots[j].Name()
+ })
+ for i := 0; i < len(snapshots)-maxNum; i++ {
+ lfs.MustRMAll(filepath.Join(root, snapshots[i].Name()))
+ }
+}
diff --git a/banyand/internal/storage/snapshot_test.go b/banyand/internal/storage/snapshot_test.go
new file mode 100644
index 00000000..b3b4c0da
--- /dev/null
+++ b/banyand/internal/storage/snapshot_test.go
@@ -0,0 +1,62 @@
+// 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 storage
+
+import (
+ "path/filepath"
+ "sort"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+func TestDeleteStaleSnapshots(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ snapshotsRoot := filepath.Join(tmpPath, "snapshots")
+ fileSystem.MkdirIfNotExist(snapshotsRoot, 0o755)
+
+ snapshotNames := []string{
+ "20201010101010-00000001",
+ "20201010101010-00000002",
+ "20201010101010-00000003",
+ "20201010101010-00000004",
+ }
+ for _, name := range snapshotNames {
+ dirPath := filepath.Join(snapshotsRoot, name)
+ fileSystem.MkdirIfNotExist(dirPath, 0o755)
+ }
+
+ DeleteStaleSnapshots(snapshotsRoot, 2, fileSystem)
+
+ remaining := fileSystem.ReadDir(snapshotsRoot)
+ require.Equal(t, 2, len(remaining))
+
+ var names []string
+ for _, info := range remaining {
+ names = append(names, info.Name())
+ }
+ sort.Strings(names)
+ assert.Equal(t, []string{"20201010101010-00000003", "20201010101010-00000004"}, names)
+}
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 1735e80b..5d6eb933 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
@@ -53,18 +54,20 @@ type Service interface {
var _ Service = (*service)(nil)
type service struct {
- writeListener bus.MessageListener
+ lfs fs.FileSystem
metadata metadata.Repo
pipeline queue.Server
localPipeline queue.Queue
metricPipeline queue.Server
omr observability.MetricsRegistry
+ writeListener bus.MessageListener
schemaRepo *schemaRepo
- l *logger.Logger
pm *protector.Memory
+ l *logger.Logger
root string
option option
maxDiskUsagePercent int
+ maxFileSnapshotNum int
}
func (s *service) Measure(metadata *commonv1.Metadata) (Measure, error) {
@@ -88,6 +91,7 @@ func (s *service) FlagSet() *run.FlagSet {
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
flagS.VarP(&s.option.seriesCacheMaxSize, "measure-series-cache-max-size", "", "the max size of series cache in each group")
flagS.IntVar(&s.maxDiskUsagePercent, "measure-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
+ flagS.IntVar(&s.maxFileSnapshotNum, "measure-max-file-snapshot-num", 10, "the maximum number of file snapshots allowed")
return flagS
}
@@ -114,6 +118,7 @@ func (s *service) Role() databasev1.Role {
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())
observability.UpdatePath(path)
s.localPipeline = queue.Local()
diff --git a/banyand/measure/snapshot.go b/banyand/measure/snapshot.go
index c4c33876..6b6ea56f 100644
--- a/banyand/measure/snapshot.go
+++ b/banyand/measure/snapshot.go
@@ -219,6 +219,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
}
s.snapshotMux.Lock()
defer s.snapshotMux.Unlock()
+ storage.DeleteStaleSnapshots(filepath.Join(s.s.schemaRepo.path, snapshotsDir), s.s.maxFileSnapshotNum, s.s.lfs)
sn := s.snapshotName()
var err error
for _, g := range gg {
diff --git a/banyand/property/listener.go b/banyand/property/listener.go
index 82f02791..2aee3a0a 100644
--- a/banyand/property/listener.go
+++ b/banyand/property/listener.go
@@ -33,6 +33,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -229,6 +230,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
}
s.snapshotMux.Lock()
defer s.snapshotMux.Unlock()
+ storage.DeleteStaleSnapshots(filepath.Join(s.s.db.location, snapshotsDir), s.s.maxFileSnapshotNum, s.s.lfs)
sn := s.snapshotName()
shardsRef := s.s.db.sLst.Load()
if shardsRef == nil {
diff --git a/banyand/property/service.go b/banyand/property/service.go
index 27fab914..6afd7198 100644
--- a/banyand/property/service.go
+++ b/banyand/property/service.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -46,6 +47,7 @@ type service struct {
metadata metadata.Repo
pipeline queue.Server
omr observability.MetricsRegistry
+ lfs fs.FileSystem
l *logger.Logger
db *database
close chan struct{}
@@ -53,6 +55,7 @@ type service struct {
nodeID string
flushTimeout time.Duration
maxDiskUsagePercent int
+ maxFileSnapshotNum int
}
func (s *service) FlagSet() *run.FlagSet {
@@ -60,6 +63,7 @@ func (s *service) FlagSet() *run.FlagSet {
flagS.StringVar(&s.root, "property-root-path", "/tmp", "the root path of database")
flagS.DurationVar(&s.flushTimeout, "property-flush-timeout", defaultFlushTimeout, "the memory data timeout of measure")
flagS.IntVar(&s.maxDiskUsagePercent, "property-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
+ flagS.IntVar(&s.maxFileSnapshotNum, "property-max-file-snapshot-num", 2, "the maximum number of file snapshots allowed")
return flagS
}
@@ -86,6 +90,7 @@ func (s *service) Role() databasev1.Role {
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())
observability.UpdatePath(path)
val := ctx.Value(common.ContextNodeKey)
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index 3da45a69..3a591df9 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
@@ -58,12 +59,14 @@ type service struct {
pipeline queue.Server
localPipeline queue.Queue
omr observability.MetricsRegistry
- l *logger.Logger
+ lfs fs.FileSystem
pm *protector.Memory
+ l *logger.Logger
schemaRepo schemaRepo
root string
option option
maxDiskUsagePercent int
+ maxFileSnapshotNum int
}
func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
@@ -88,6 +91,7 @@ func (s *service) FlagSet() *run.FlagSet {
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
flagS.VarP(&s.option.seriesCacheMaxSize, "stream-series-cache-max-size", "", "the max size of series cache in each group")
flagS.IntVar(&s.maxDiskUsagePercent, "stream-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
+ flagS.IntVar(&s.maxFileSnapshotNum, "stream-max-file-snapshot-num", 2, "the maximum number of file snapshots allowed")
return flagS
}
@@ -114,6 +118,7 @@ func (s *service) Role() databasev1.Role {
func (s *service) PreRun(_ context.Context) error {
s.l = logger.GetLogger(s.Name())
+ s.lfs = fs.NewLocalFileSystemWithLogger(s.l)
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
s.localPipeline = queue.Local()
diff --git a/banyand/stream/snapshot.go b/banyand/stream/snapshot.go
index d10999b6..f0bacb31 100644
--- a/banyand/stream/snapshot.go
+++ b/banyand/stream/snapshot.go
@@ -225,6 +225,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
}
s.snapshotMux.Lock()
defer s.snapshotMux.Unlock()
+ storage.DeleteStaleSnapshots(filepath.Join(s.s.schemaRepo.path, snapshotsDir), s.s.maxFileSnapshotNum, s.s.lfs)
sn := s.snapshotName()
var err error
for _, g := range gg {
diff --git a/bydbbackup/Makefile b/bydbbackup/Makefile
new file mode 100644
index 00000000..699e660a
--- /dev/null
+++ b/bydbbackup/Makefile
@@ -0,0 +1,34 @@
+# 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.
+#
+
+NAME := bydbbackup
+BINARIES := $(NAME)-backup
+
+IMG_NAME := skywalking-bydbbackup
+
+include ../scripts/build/version.mk
+include ../scripts/build/base.mk
+include ../scripts/build/generate_go.mk
+include ../scripts/build/build.mk
+include ../scripts/build/docker.mk
+include ../scripts/build/test.mk
+include ../scripts/build/lint.mk
+include ../scripts/build/vendor.mk
+include ../scripts/build/help.mk
+
+prepare-build: generate
diff --git a/bydbbackup/cmd/backup/main.go b/bydbbackup/cmd/backup/main.go
new file mode 100644
index 00000000..4c823973
--- /dev/null
+++ b/bydbbackup/cmd/backup/main.go
@@ -0,0 +1,34 @@
+// 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 main provides main entry for the backup command-line tool.
+package main
+
+import (
+ "fmt"
+ "os"
+
+ "github.com/apache/skywalking-banyandb/bydbbackup/internal/backup"
+)
+
+func main() {
+ cmd := backup.NewBackupCommand()
+ if err := cmd.Execute(); err != nil {
+ fmt.Fprintln(os.Stderr, err)
+ os.Exit(1)
+ }
+}
diff --git a/bydbbackup/internal/backup/backup.go b/bydbbackup/internal/backup/backup.go
new file mode 100644
index 00000000..ae1042b9
--- /dev/null
+++ b/bydbbackup/internal/backup/backup.go
@@ -0,0 +1,292 @@
+// 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 backup provides the backup command-line tool.
+package backup
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/url"
+ "os"
+ "os/signal"
+ "path"
+ "path/filepath"
+ "syscall"
+ "time"
+
+ "github.com/benbjohnson/clock"
+ "github.com/robfig/cron/v3"
+ "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/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"
+)
+
+const snapshotDir = "snapshots"
+
+var scheduleExprMap = map[string]string{
+ "hourly": "5 * * * *",
+ "daily": "5 0 * * *",
+}
+
+// NewBackupCommand creates a new backup command.
+func NewBackupCommand() *cobra.Command {
+ var (
+ gRPCAddr string
+ enableTLS bool
+ insecure bool
+ cert string
+ streamRoot string
+ measureRoot string
+ propertyRoot string
+ dest string
+ timeStyle string
+ scheduleStyle string
+ )
+
+ cmd := &cobra.Command{
+ Use: "backup",
+ Short: "Backup BanyanDB snapshots to remote storage",
+ RunE: func(_ *cobra.Command, _ []string) error {
+ if scheduleStyle == "" {
+ return backupAction(dest, gRPCAddr, enableTLS, insecure, cert,
+ streamRoot, measureRoot, propertyRoot, timeStyle)
+ }
+ expr, ok := scheduleExprMap[scheduleStyle]
+ if !ok {
+ return fmt.Errorf("unsupported schedule style: %s. Only support: hourly or daily", scheduleStyle)
+ }
+ schedLogger := logger.GetLogger().Named("backup-scheduler")
+ clockInstance := clock.New()
+ sch := timestamp.NewScheduler(schedLogger, clockInstance)
+ cronOptions := cron.Minute | cron.Hour
+ err := sch.Register("backup", cronOptions, expr, func(_ time.Time, l *logger.Logger) bool {
+ err := backupAction(dest, gRPCAddr, enableTLS, insecure, cert,
+ streamRoot, measureRoot, propertyRoot, timeStyle)
+ if err != nil {
+ l.Error().Err(err).Msg("backup failed")
+ } else {
+ l.Info().Msg("backup 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(&gRPCAddr, "grpc-addr", "127.0.0.1:17912", "gRPC address of the data node")
+ cmd.Flags().BoolVar(&enableTLS, "enable-tls", false, "Enable TLS for gRPC connection")
+ cmd.Flags().BoolVar(&insecure, "insecure", false, "Skip server certificate verification")
+ cmd.Flags().StringVar(&cert, "cert", "", "Path to the gRPC server certificate")
+ cmd.Flags().StringVar(&streamRoot, "stream-root-path", "/tmp", "Root directory for stream catalog")
+ cmd.Flags().StringVar(&measureRoot, "measure-root-path", "/tmp", "Root directory for measure catalog")
+ cmd.Flags().StringVar(&propertyRoot, "property-root-path", "/tmp", "Root directory for property catalog")
+ cmd.Flags().StringVar(&dest, "dest", "", "Destination URL (e.g., file:///backups)")
+ cmd.Flags().StringVar(&timeStyle, "time-style", "daily", "Time directory style (daily|hourly)")
+ cmd.Flags().StringVar(&scheduleStyle, "schedule", "", "Schedule expression for periodic backup. The format is a cron expression \"<minute> <hour>\"")
+
+ return cmd
+}
+
+func backupAction(dest, gRPCAddr string, enableTLS, insecure bool, cert,
+ streamRoot, measureRoot, propertyRoot, timeStyle string,
+) error {
+ if dest == "" {
+ return errors.New("dest is required")
+ }
+
+ fs, err := newFS(dest)
+ if err != nil {
+ return err
+ }
+ defer fs.Close()
+
+ snapshots, err := getSnapshots(gRPCAddr, enableTLS, insecure, cert)
+ if err != nil {
+ return err
+ }
+
+ timeDir := getTimeDir(timeStyle)
+
+ for _, snapshot := range snapshots {
+ var snapshotDir string
+ snapshotDir, err = getSnapshotDir(snapshot, streamRoot, measureRoot, propertyRoot)
+ if err != nil {
+ logger.Warningf("Failed to get snapshot directory for %s: %v", snapshot.Name, err)
+ continue
+ }
+ multierr.AppendInto(&err, backupSnapshot(fs, snapshotDir, snapshot.Name, timeDir))
+ }
+ return err
+}
+
+func newFS(dest string) (remote.FS, error) {
+ u, err := url.Parse(dest)
+ if err != nil {
+ return nil, fmt.Errorf("invalid dest URL: %w", err)
+ }
+
+ switch u.Scheme {
+ case "file":
+ return local.NewFS(u.Path)
+ default:
+ return nil, fmt.Errorf("unsupported scheme: %s", u.Scheme)
+ }
+}
+
+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 = streamRoot
+ case commonv1.Catalog_CATALOG_MEASURE:
+ baseDir = measureRoot
+ case commonv1.Catalog_CATALOG_PROPERTY:
+ baseDir = propertyRoot
+ default:
+ return "", errors.New("unknown catalog type")
+ }
+
+ return filepath.Join(baseDir, snapshotDir, snapshot.Name), nil
+}
+
+func getTimeDir(style string) string {
+ now := time.Now()
+ switch style {
+ case "hourly":
+ return now.Format("2006-01-02-15")
+ default:
+ return now.Format("2006-01-02")
+ }
+}
+
+func backupSnapshot(fs remote.FS, snapshotDir, snapshotName, timeDir string) error {
+ localFiles, err := getAllFiles(snapshotDir)
+ if err != nil {
+ return err
+ }
+
+ ctx := context.Background()
+ remotePrefix := path.Join(timeDir, snapshotName) + "/"
+
+ remoteFiles, err := fs.List(ctx, remotePrefix)
+ if err != nil {
+ return err
+ }
+
+ for _, relPath := range localFiles {
+ remotePath := path.Join(timeDir, snapshotName, relPath)
+ if !contains(remoteFiles, remotePath) {
+ if err := uploadFile(ctx, fs, snapshotDir, relPath, remotePath); err != nil {
+ return err
+ }
+ }
+ }
+
+ deleteOrphanedFiles(ctx, fs, localFiles, remoteFiles, timeDir, snapshotName)
+ return nil
+}
+
+func getAllFiles(root string) ([]string, error) {
+ var files []string
+ err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ if !info.IsDir() {
+ relPath, err := filepath.Rel(root, path)
+ if err != nil {
+ return err
+ }
+ files = append(files, filepath.ToSlash(relPath))
+ }
+ return nil
+ })
+ return files, err
+}
+
+func uploadFile(ctx context.Context, fs remote.FS, snapshotDir, relPath, remotePath string) error {
+ localPath := filepath.Join(snapshotDir, relPath)
+ file, err := os.Open(localPath)
+ if err != nil {
+ return err
+ }
+ defer file.Close()
+
+ return fs.Upload(ctx, remotePath, file)
+}
+
+func deleteOrphanedFiles(ctx context.Context, fs remote.FS, localFiles, remoteFiles []string, timeDir, snapshotName string) {
+ expected := make(map[string]struct{})
+ for _, f := range localFiles {
+ expected[path.Join(timeDir, snapshotName, f)] = struct{}{}
+ }
+
+ for _, remoteFile := range remoteFiles {
+ if _, exists := expected[remoteFile]; !exists {
+ if err := fs.Delete(ctx, remoteFile); err != nil {
+ logger.Warningf("Warning: failed to delete orphaned file %s: %v\n", remoteFile, err)
+ }
+ }
+ }
+}
+
+func contains(slice []string, s string) bool {
+ for _, item := range slice {
+ if item == s {
+ return true
+ }
+ }
+ return false
+}
diff --git a/bydbbackup/internal/backup/backup_test.go b/bydbbackup/internal/backup/backup_test.go
new file mode 100644
index 00000000..ef3a6247
--- /dev/null
+++ b/bydbbackup/internal/backup/backup_test.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 backup
+
+import (
+ "context"
+ "io"
+ "os"
+ "path"
+ "path/filepath"
+ "testing"
+ "time"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+func TestNewFS(t *testing.T) {
+ tests := []struct {
+ name string
+ dest string
+ wantErr bool
+ }{
+ {"valid file scheme", "file:///tmp", false},
+ {"invalid scheme", "s3://bucket", true},
+ {"malformed URL", ":invalid", true},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ _, err := newFS(tt.dest)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("newFS() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ })
+ }
+}
+
+func TestGetSnapshotDir(t *testing.T) {
+ tests := []struct {
+ name string
+ snapshot *databasev1.Snapshot
+ streamRoot string
+ measureRoot string
+ propRoot string
+ want string
+ wantErr bool
+ }{
+ {
+ "stream catalog",
+ &databasev1.Snapshot{Catalog: commonv1.Catalog_CATALOG_STREAM, Name: "test"},
+ "/stream", "/measure", "/prop",
+ filepath.Join("/stream", snapshotDir, "test"),
+ false,
+ },
+ {
+ "unknown catalog",
+ &databasev1.Snapshot{Catalog: commonv1.Catalog_CATALOG_UNSPECIFIED, Name: "test"},
+ "", "", "",
+ "",
+ true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := getSnapshotDir(tt.snapshot, tt.streamRoot, tt.measureRoot, tt.propRoot)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("getSnapshotDir() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if got != tt.want {
+ t.Errorf("getSnapshotDir() = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
+
+func TestGetTimeDir(t *testing.T) {
+ now := time.Now()
+ tests := []struct {
+ name string
+ style string
+ want string
+ }{
+ {"hourly", "hourly", now.Format("2006-01-02-15")},
+ {"daily (default)", "invalid", now.Format("2006-01-02")},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got := getTimeDir(tt.style)
+ if got != tt.want {
+ t.Errorf("getTimeDir() = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
+
+func TestGetAllFiles(t *testing.T) {
+ tmpDir := t.TempDir()
+ // Create test files and subdirectory.
+ os.WriteFile(filepath.Join(tmpDir, "file1"), nil, 0o600)
+ os.Mkdir(filepath.Join(tmpDir, "sub"), 0o755)
+ os.WriteFile(filepath.Join(tmpDir, "sub/file2"), nil, 0o600)
+
+ want := []string{"file1", "sub/file2"}
+ files, err := getAllFiles(tmpDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(files) != len(want) {
+ t.Fatalf("got %d files, want %d", len(files), len(want))
+ }
+ for i, f := range want {
+ if files[i] != f {
+ t.Errorf("file[%d] = %v, want %v", i, files[i], f)
+ }
+ }
+}
+
+type mockFS struct {
+ uploaded []string
+ deleted []string
+}
+
+func (m *mockFS) List(_ context.Context, prefix string) ([]string, error) {
+ return []string{path.Join(prefix, "existing.txt")}, nil // Simulate existing remote file.
+}
+
+func (m *mockFS) Upload(_ context.Context, p string, _ io.Reader) error {
+ m.uploaded = append(m.uploaded, p)
+ return nil
+}
+
+func (m *mockFS) Delete(_ context.Context, p string) error {
+ m.deleted = append(m.deleted, p)
+ return nil
+}
+
+func (m *mockFS) Download(_ context.Context, _ string) (io.ReadCloser, error) { return nil, nil }
+
+func (m *mockFS) Close() error { return nil }
+
+func TestBackupSnapshot(t *testing.T) {
+ tmpDir := t.TempDir()
+ os.WriteFile(filepath.Join(tmpDir, "newfile.txt"), nil, 0o600)
+
+ m := &mockFS{}
+ err := backupSnapshot(m, tmpDir, "test-snapshot", "daily")
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ wantUpload := "daily/test-snapshot/newfile.txt"
+ if len(m.uploaded) != 1 || m.uploaded[0] != wantUpload {
+ t.Errorf("uploaded = %v, want %v", m.uploaded, wantUpload)
+ }
+
+ wantDelete := "daily/test-snapshot/existing.txt"
+ if len(m.deleted) != 1 || m.deleted[0] != wantDelete {
+ t.Errorf("deleted = %v, want %v", m.deleted, wantDelete)
+ }
+}
+
+func TestContains(t *testing.T) {
+ tests := []struct {
+ s string
+ slice []string
+ want bool
+ }{
+ {slice: []string{"a", "b"}, s: "a", want: true},
+ {slice: []string{"a", "b"}, s: "c", want: false},
+ }
+ for _, tt := range tests {
+ got := contains(tt.slice, tt.s)
+ if got != tt.want {
+ t.Errorf("contains(%v, %s) = %v, want %v", tt.slice, tt.s, got, tt.want)
+ }
+ }
+}
diff --git a/bydbctl/internal/cmd/health_check.go b/bydbctl/internal/cmd/health_check.go
index ce54b6e6..2fe2cd0e 100644
--- a/bydbctl/internal/cmd/health_check.go
+++ b/bydbctl/internal/cmd/health_check.go
@@ -18,19 +18,14 @@
package cmd
import (
- "crypto/tls"
- "crypto/x509"
"fmt"
- "os"
"time"
"github.com/go-resty/resty/v2"
- "github.com/pkg/errors"
"github.com/spf13/cobra"
"google.golang.org/grpc"
- "google.golang.org/grpc/credentials"
- ins "google.golang.org/grpc/credentials/insecure"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/version"
)
@@ -51,26 +46,9 @@ func newHealthCheckCmd() *cobra.Command {
return request.req.Get(getPath("/api/healthz"))
}, yamlPrinter, enableTLS, insecure, cert)
}
- if enableTLS {
- config := &tls.Config{
- // #nosec G402
- InsecureSkipVerify: insecure,
- }
- if cert != "" {
- cert, errRead := os.ReadFile(cert)
- if errRead != nil {
- return errRead
- }
- certPool := x509.NewCertPool()
- if !certPool.AppendCertsFromPEM(cert) {
- return errors.New("failed to add server's certificate")
- }
- config.RootCAs = certPool
- }
- creds := credentials.NewTLS(config)
- opts = append(opts, grpc.WithTransportCredentials(creds))
- } else {
- opts = append(opts, grpc.WithTransportCredentials(ins.NewCredentials()))
+ opts, err = grpchelper.SecureOptions(opts, enableTLS, insecure, cert)
+ if err != nil {
+ return err
}
err = helpers.HealthCheck(grpcAddr, 10*time.Second, 10*time.Second, opts...)()
if err == nil {
diff --git a/pkg/fs/remote/local/local.go b/pkg/fs/remote/local/local.go
new file mode 100644
index 00000000..01f20455
--- /dev/null
+++ b/pkg/fs/remote/local/local.go
@@ -0,0 +1,98 @@
+// 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 local provides a local file system implementation.
+package local
+
+import (
+ "context"
+ "io"
+ "os"
+ "path/filepath"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs/remote"
+)
+
+const dirPerm = 0o755
+
+var _ remote.FS = (*fs)(nil)
+
+type fs struct {
+ baseDir string
+}
+
+// NewFS creates a new local file system.
+func NewFS(baseDir string) (remote.FS, error) {
+ if err := os.MkdirAll(baseDir, dirPerm); err != nil {
+ return nil, err
+ }
+ return &fs{baseDir: baseDir}, nil
+}
+
+func (l *fs) Upload(_ context.Context, path string, data io.Reader) error {
+ fullPath := filepath.Join(l.baseDir, path)
+ if err := os.MkdirAll(filepath.Dir(fullPath), dirPerm); err != nil {
+ return err
+ }
+
+ file, err := os.Create(fullPath)
+ if err != nil {
+ return err
+ }
+ defer file.Close()
+
+ _, err = io.Copy(file, data)
+ return err
+}
+
+func (l *fs) Download(_ context.Context, path string) (io.ReadCloser, error) {
+ fullPath := filepath.Join(l.baseDir, path)
+ return os.Open(fullPath)
+}
+
+func (l *fs) List(_ context.Context, prefix string) ([]string, error) {
+ var files []string
+ fullPath := filepath.Join(l.baseDir, prefix)
+
+ err := filepath.Walk(fullPath, func(path string, info os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ if !info.IsDir() {
+ relPath, err := filepath.Rel(l.baseDir, path)
+ if err != nil {
+ return err
+ }
+ files = append(files, filepath.ToSlash(relPath))
+ }
+ return nil
+ })
+
+ if os.IsNotExist(err) {
+ return nil, nil
+ }
+ return files, err
+}
+
+func (l *fs) Delete(_ context.Context, path string) error {
+ fullPath := filepath.Join(l.baseDir, path)
+ return os.Remove(fullPath)
+}
+
+func (l *fs) Close() error {
+ return nil
+}
diff --git a/pkg/fs/remote/remote.go b/pkg/fs/remote/remote.go
new file mode 100644
index 00000000..3d9d7b21
--- /dev/null
+++ b/pkg/fs/remote/remote.go
@@ -0,0 +1,49 @@
+// 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 remote provides an interface for interacting with various remote file systems such as S3, GCS, and distributed file systems for backup/restore operations.
+package remote
+
+import (
+ "context"
+ "io"
+)
+
+// FS defines the interface for interacting with various remote file systems
+// such as S3, GCS, and distributed file systems for backup/restore operations.
+type FS interface {
+ // Upload writes a file to the remote file system at the specified path.
+ // The path should include the snapshot directory (e.g. "snapshot123/file.data").
+ // Implementations must handle both new file creation and overwrite scenarios.
+ Upload(ctx context.Context, path string, data io.Reader) error
+
+ // Download retrieves a file from the remote file system at the specified path.
+ // Returns a ReadCloser that must be closed by the caller after consumption.
+ Download(ctx context.Context, path string) (io.ReadCloser, error)
+
+ // List returns all file paths under a given prefix (typically a snapshot directory).
+ // Used to check existing files during backup/restore operations.
+ List(ctx context.Context, prefix string) ([]string, error)
+
+ // Delete removes a specific file from the remote file system.
+ // Path must include the snapshot directory and file name.
+ Delete(ctx context.Context, path string) error
+
+ // Close releases any resources or connections associated with the remote FS.
+ // Must be called when the client is no longer needed.
+ Close() error
+}
diff --git a/pkg/grpchelper/client.go b/pkg/grpchelper/client.go
index 34ea546d..db583cfe 100644
--- a/pkg/grpchelper/client.go
+++ b/pkg/grpchelper/client.go
@@ -20,10 +20,16 @@ package grpchelper
import (
"context"
+ "crypto/tls"
+ "crypto/x509"
+ "errors"
+ "os"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ ins "google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
@@ -31,6 +37,32 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
+// SecureOptions returns gRPC dial options with secure connection settings.
+func SecureOptions(dest []grpc.DialOption, enabled, insecure bool, cert string) ([]grpc.DialOption, error) {
+ if !enabled {
+ dest = append(dest, grpc.WithTransportCredentials(ins.NewCredentials()))
+ return dest, nil
+ }
+ config := &tls.Config{
+ // #nosec G402
+ InsecureSkipVerify: insecure,
+ }
+ if cert != "" {
+ cert, errRead := os.ReadFile(cert)
+ if errRead != nil {
+ return nil, errRead
+ }
+ certPool := x509.NewCertPool()
+ if !certPool.AppendCertsFromPEM(cert) {
+ return nil, errors.New("failed to add server's certificate")
+ }
+ config.RootCAs = certPool
+ }
+ creds := credentials.NewTLS(config)
+ dest = append(dest, grpc.WithTransportCredentials(creds))
+ return dest, nil
+}
+
// Conn returns a gRPC client connection once connecting the server.
func Conn(addr string, healthCheckTimeout time.Duration, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
l := logger.GetLogger("grpc-helper")
(skywalking-banyandb) 02/02: Add backup CLI tool
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch backup
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 29920fb089da1a5ca0a1ea1b1ccd304b933bb414
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Feb 11 08:07:49 2025 +0800
Add backup CLI tool
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/internal/storage/version.go | 2 +-
banyand/internal/storage/versions.yml | 2 +-
banyand/measure/service.go | 13 +++++--
banyand/measure/snapshot.go | 4 +-
banyand/stream/service.go | 7 +++-
banyand/stream/snapshot.go | 4 +-
docs/operation/backup.md | 73 +++++++++++++++++++++++++++++++++++
7 files changed, 94 insertions(+), 11 deletions(-)
diff --git a/banyand/internal/storage/version.go b/banyand/internal/storage/version.go
index dc8e33c8..8225629e 100644
--- a/banyand/internal/storage/version.go
+++ b/banyand/internal/storage/version.go
@@ -28,7 +28,7 @@ import (
const (
metadataFilename = "metadata"
- currentVersion = "1.1.0"
+ currentVersion = "1.2.0"
compatibleVersionsKey = "versions"
compatibleVersionsFilename = "versions.yml"
)
diff --git a/banyand/internal/storage/versions.yml b/banyand/internal/storage/versions.yml
index 5458e2d7..eb98c0e6 100644
--- a/banyand/internal/storage/versions.yml
+++ b/banyand/internal/storage/versions.yml
@@ -14,4 +14,4 @@
# limitations under the License.
versions:
-- 1.1.0
+- 1.2.0
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 5d6eb933..35f1aac2 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -20,6 +20,7 @@ package measure
import (
"context"
"path"
+ "path/filepath"
"github.com/pkg/errors"
@@ -37,6 +38,8 @@ import (
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
)
+const dataDir = "data"
+
var (
errEmptyRootPath = errors.New("root path is empty")
// ErrMeasureNotExist denotes a measure doesn't exist in the metadata repo.
@@ -54,17 +57,18 @@ type Service interface {
var _ Service = (*service)(nil)
type service struct {
+ writeListener bus.MessageListener
lfs fs.FileSystem
- metadata metadata.Repo
pipeline queue.Server
localPipeline queue.Queue
metricPipeline queue.Server
omr observability.MetricsRegistry
- writeListener bus.MessageListener
- schemaRepo *schemaRepo
+ metadata metadata.Repo
pm *protector.Memory
+ schemaRepo *schemaRepo
l *logger.Logger
root string
+ snapshotDir string
option option
maxDiskUsagePercent int
maxFileSnapshotNum int
@@ -120,9 +124,10 @@ 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, snapshotsDir)
observability.UpdatePath(path)
s.localPipeline = queue.Local()
- s.schemaRepo = newSchemaRepo(path, s)
+ s.schemaRepo = newSchemaRepo(filepath.Join(path, dataDir), s)
if err := s.createNativeObservabilityGroup(ctx); err != nil {
return err
diff --git a/banyand/measure/snapshot.go b/banyand/measure/snapshot.go
index 6b6ea56f..a873b19b 100644
--- a/banyand/measure/snapshot.go
+++ b/banyand/measure/snapshot.go
@@ -219,7 +219,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
}
s.snapshotMux.Lock()
defer s.snapshotMux.Unlock()
- storage.DeleteStaleSnapshots(filepath.Join(s.s.schemaRepo.path, snapshotsDir), s.s.maxFileSnapshotNum, s.s.lfs)
+ storage.DeleteStaleSnapshots(s.s.snapshotDir, s.s.maxFileSnapshotNum, s.s.lfs)
sn := s.snapshotName()
var err error
for _, g := range gg {
@@ -228,7 +228,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), nil)
default:
}
- if errGroup := s.s.takeGroupSnapshot(filepath.Join(s.s.schemaRepo.path, snapshotsDir, sn), g.GetSchema().Metadata.Name); err != nil {
+ if errGroup := s.s.takeGroupSnapshot(filepath.Join(s.s.snapshotDir, sn), g.GetSchema().Metadata.Name); err != nil {
s.s.l.Error().Err(errGroup).Str("group", g.GetSchema().Metadata.Name).Msg("fail to take group snapshot")
err = multierr.Append(err, errGroup)
continue
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index 3a591df9..e1e50e70 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -20,6 +20,7 @@ package stream
import (
"context"
"path"
+ "path/filepath"
"github.com/pkg/errors"
@@ -37,6 +38,8 @@ import (
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
)
+const dataDir = "data"
+
var (
errEmptyRootPath = errors.New("root path is empty")
// ErrStreamNotExist denotes a stream doesn't exist in the metadata repo.
@@ -64,6 +67,7 @@ type service struct {
l *logger.Logger
schemaRepo schemaRepo
root string
+ snapshotDir string
option option
maxDiskUsagePercent int
maxFileSnapshotNum int
@@ -120,9 +124,10 @@ func (s *service) PreRun(_ 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, snapshotsDir)
observability.UpdatePath(path)
s.localPipeline = queue.Local()
- s.schemaRepo = newSchemaRepo(path, s)
+ s.schemaRepo = newSchemaRepo(filepath.Join(path, dataDir), s)
// run a serial watcher
if err := s.pipeline.Subscribe(data.TopicSnapshot, &snapshotListener{s: s}); err != nil {
diff --git a/banyand/stream/snapshot.go b/banyand/stream/snapshot.go
index f0bacb31..1cee3f2a 100644
--- a/banyand/stream/snapshot.go
+++ b/banyand/stream/snapshot.go
@@ -225,7 +225,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
}
s.snapshotMux.Lock()
defer s.snapshotMux.Unlock()
- storage.DeleteStaleSnapshots(filepath.Join(s.s.schemaRepo.path, snapshotsDir), s.s.maxFileSnapshotNum, s.s.lfs)
+ storage.DeleteStaleSnapshots(s.s.snapshotDir, s.s.maxFileSnapshotNum, s.s.lfs)
sn := s.snapshotName()
var err error
for _, g := range gg {
@@ -234,7 +234,7 @@ func (s *snapshotListener) Rev(ctx context.Context, message bus.Message) bus.Mes
return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), nil)
default:
}
- if errGroup := s.s.takeGroupSnapshot(filepath.Join(s.s.schemaRepo.path, snapshotsDir, sn), g.GetSchema().Metadata.Name); err != nil {
+ if errGroup := s.s.takeGroupSnapshot(filepath.Join(s.s.snapshotDir, sn), g.GetSchema().Metadata.Name); err != nil {
s.s.l.Error().Err(errGroup).Str("group", g.GetSchema().Metadata.Name).Msg("fail to take group snapshot")
err = multierr.Append(err, errGroup)
continue
diff --git a/docs/operation/backup.md b/docs/operation/backup.md
new file mode 100644
index 00000000..4162c258
--- /dev/null
+++ b/docs/operation/backup.md
@@ -0,0 +1,73 @@
+# Backup Data Using BanyanDB Backup Tool
+
+The backup command-line tool is used to backup BanyanDB snapshots to remote storage. The tool allows both one-time and scheduled backups.
+
+## Overview
+
+The backup tool performs the following operations:
+
+- Connects to a BanyanDB data node via gRPC to request the latest snapshots.
+- Determines the local snapshot directories based on the catalog type:
+ - **Stream Catalog:** Uses the `stream-root-path`.
+ - **Measure Catalog:** Uses the `measure-root-path`.
+ - **Property Catalog:** Uses the `property-root-path`.
+- Computes a time-based directory name (formatted as daily or hourly).
+- Uploads files that are not found in the remote storage to the specified destination.
+- Deletes orphaned files in the remote destination that no longer exist locally.
+- Optionally schedules periodic backups using cron-style expressions.
+
+## Prerequisites
+
+Before running the backup tool, ensure you have:
+
+- A running BanyanDB data node exposing a gRPC service (by default at `127.0.0.1:17912`).
+- Network connectivity to the data node.
+- A valid destination URL specified using the `file:///` scheme (or any supported scheme in future releases).
+- Necessary access rights for reading snapshot data and writing to the destination.
+
+## Command-Line Usage
+
+### One-Time Backup
+
+To perform a one-time backup, run the backup command with the required flags. At a minimum, you must specify the destination URL.
+
+**Example Command:**
+```bash
+./bydbbackup backup --dest "file:///backups"
+```
+
+### Scheduled Backup
+
+To enable periodic backups, provide the `--schedule` flag with a schedule style (e.g., `hourly` or `daily`) and set your preferred time style using `--time-style`. The supported schedule expressions based on the tool’s internal map are:
+
+- **Hourly:** Schedules at "5 * * * *" (runs on the 5th minute of every hour).
+- **Daily:** Schedules at "5 0 * * *" (runs at 00:05 every day).
+
+**Example Command:**
+
+```bash
+./bydbbackup backup --dest "file:///backups" --schedule daily --time-style daily
+```
+
+When a schedule is provided, the tool:
+
+- Registers a cron job using an internal scheduler.
+- Runs the backup action periodically according to the scheduled expression.
+- Waits for termination signals (`SIGINT` or `SIGTERM`) to gracefully shut down.
+
+## Detailed Options
+
+| Flag | Description | Default Value |
+| ------------------- | ----------------------------------------------------------------------------------------- | --------------------- |
+| `--grpc-addr` | gRPC address of the data node. | `127.0.0.1:17912` |
+| `--enable-tls` | Enable TLS for the gRPC connection. | `false` |
+| `--insecure` | Skip server certificate verification. | `false` |
+| `--cert` | Path to the gRPC server certificate. | _empty_ |
+| `--stream-root-path`| Root directory for the stream catalog snapshots. | `/tmp` |
+| `--measure-root-path`| Root directory for the measure catalog snapshots. | `/tmp` |
+| `--property-root-path`| Root directory for the property catalog snapshots. | `/tmp` |
+| `--dest` | Destination URL for backup data. (e.g., `file:///backups`) | _required_ |
+| `--time-style` | Directory naming style based on time. Supports `daily` or `hourly`. | `daily` |
+| `--schedule` | Schedule style for periodic backup. If not set, backup is performed once. Options: `hourly` or `daily`. | _empty_ |
+
+This guide should provide you with the necessary steps and information to effectively use the backup tool for your data backup operations.