You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/10/09 20:55:59 UTC
(skywalking-banyandb) 01/01: Fix backup reporting success when an catalog upload failure
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch fix/backup-catalog-error-swallowed
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 454aaa5ceb30fc2a52b9f0e67deadec3a0dd1e35
Author: mrproliu <74...@qq.com>
AuthorDate: Fri Oct 9 22:54:03 2026 +0200
Fix backup reporting success when an catalog upload failure
---
CHANGES.md | 1 +
banyand/backup/backup.go | 15 ++--
banyand/backup/backup_test.go | 173 ++++++++++++++++++++++++++++++++++++++++++
3 files changed, 184 insertions(+), 5 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 9b0260287..47c75031f 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -57,6 +57,7 @@ Release Notes.
- Reject a `--trace-vectorized-batch-size` above 65,536 at startup instead of accepting it. The trace query path indexes a batch's active rows through a `uint16` selection, so a batch size above that bound silently wrapped -- reading early rows a second time and never reading the rest. The stream path already enforced this bound; measure now does too. A node configured above the bound will fail to start rather than return wrong results, and must lower the flag.
- Fix FODC proxy OOM kills on `/metrics`: stream the Prometheus text to the response, stop copying label maps, share one agent collection between concurrent scrapes, and install the same panic diagnostics as the agent.
- Stop the scheduled backup/lifecycle/retention scheduler from logging a false `action timed out` error for an action that runs longer than 5 minutes but ultimately succeeds. Crossing the internal 5-minute soft deadline now logs a lower-severity "still running" message and keeps waiting for the action's real outcome instead of abandoning it mid-flight; the eventual result is logged and the scheduler only starts the next scheduled run once the current one truly finishes.
+- Fix backup reporting success when an earlier catalog upload fails.
### Document
diff --git a/banyand/backup/backup.go b/banyand/backup/backup.go
index 95b4d6e2a..4ce4918bc 100644
--- a/banyand/backup/backup.go
+++ b/banyand/backup/backup.go
@@ -38,6 +38,7 @@ import (
"go.uber.org/multierr"
"golang.org/x/sync/errgroup"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
cfg "github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/fs/remote"
@@ -206,11 +207,15 @@ func backupAction(ctx context.Context, options backupOptions) error {
return err
}
- timeDir := getTimeDir(options.timeStyle)
+ return backupSnapshots(ctx, fs, snapshots, options, getTimeDir(options.timeStyle))
+}
+func backupSnapshots(ctx context.Context, fs remote.FS, snapshots []*databasev1.Snapshot, options backupOptions, timeDir string) error {
+ // errs accumulates per-catalog backup failures; it must not be reused for the
+ // per-iteration snapshot.Dir result, or a later catalog would reset it to nil.
+ var errs error
for _, snp := range snapshots {
- var snapshotDir string
- snapshotDir, err = snapshot.Dir(snp, options.streamRoot, options.measureRoot, options.propertyRoot, options.traceRoot, options.schemaRoot)
+ snapshotDir, err := snapshot.Dir(snp, options.streamRoot, options.measureRoot, options.propertyRoot, options.traceRoot, options.schemaRoot)
if err != nil {
logger.Warningf("Failed to get snapshot directory for %s: %v", snp.Name, err)
continue
@@ -219,9 +224,9 @@ func backupAction(ctx context.Context, options backupOptions) error {
if strings.HasPrefix(snp.Name, snapshot.SchemaPropertyCatalogName+"/") {
catalogName = snapshot.SchemaPropertyCatalogName
}
- multierr.AppendInto(&err, backupSnapshot(ctx, fs, snapshotDir, catalogName, timeDir, options.uploadConcurrency))
+ multierr.AppendInto(&errs, backupSnapshot(ctx, fs, snapshotDir, catalogName, timeDir, options.uploadConcurrency))
}
- return err
+ return errs
}
func newFS(dest string, config *remoteconfig.FsConfig) (remote.FS, error) {
diff --git a/banyand/backup/backup_test.go b/banyand/backup/backup_test.go
index e9b236276..ccad7c1b8 100644
--- a/banyand/backup/backup_test.go
+++ b/banyand/backup/backup_test.go
@@ -21,6 +21,7 @@ import (
"context"
"fmt"
"io"
+ "net"
"os"
"path"
"path/filepath"
@@ -29,6 +30,10 @@ import (
"testing"
"time"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/health"
+ "google.golang.org/grpc/health/grpc_health_v1"
+
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"
@@ -314,3 +319,171 @@ func TestContains(t *testing.T) {
}
}
}
+
+type fakeSnapshotServer struct {
+ databasev1.UnimplementedSnapshotServiceServer
+ snapshots []*databasev1.Snapshot
+}
+
+func (f *fakeSnapshotServer) Snapshot(context.Context, *databasev1.SnapshotRequest) (*databasev1.SnapshotResponse, error) {
+ return &databasev1.SnapshotResponse{Snapshots: f.snapshots}, nil
+}
+
+// TestBackupActionCatalogResults runs the full backup flow (snapshot RPC, local
+// snapshot directories, file:// destination) and verifies that a catalog upload
+// failure is reported regardless of the catalogs backed up after it, while a
+// snapshot whose directory cannot be resolved is skipped without failing the run.
+func TestBackupActionCatalogResults(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("unreadable files cannot be simulated when running as root")
+ }
+ const snapshotName = "snp"
+ type catalogData struct {
+ catalog commonv1.Catalog
+ unreadable bool
+ }
+ tests := []struct {
+ name string
+ catalogs []catalogData
+ wantErrs []string
+ wantBacked []string
+ extraUnknown bool
+ }{
+ {
+ name: "earlier catalog failure is not reset by a later success",
+ catalogs: []catalogData{
+ {catalog: commonv1.Catalog_CATALOG_STREAM, unreadable: true},
+ {catalog: commonv1.Catalog_CATALOG_MEASURE},
+ },
+ wantErrs: []string{"stream"},
+ wantBacked: []string{"measure"},
+ },
+ {
+ name: "failures of several catalogs are all reported",
+ catalogs: []catalogData{
+ {catalog: commonv1.Catalog_CATALOG_STREAM, unreadable: true},
+ {catalog: commonv1.Catalog_CATALOG_MEASURE, unreadable: true},
+ {catalog: commonv1.Catalog_CATALOG_TRACE},
+ },
+ wantErrs: []string{"stream", "measure"},
+ wantBacked: []string{"trace"},
+ },
+ {
+ name: "failure is reported when an unresolvable snapshot follows it",
+ catalogs: []catalogData{
+ {catalog: commonv1.Catalog_CATALOG_STREAM, unreadable: true},
+ {catalog: commonv1.Catalog_CATALOG_MEASURE},
+ },
+ extraUnknown: true,
+ wantErrs: []string{"stream"},
+ wantBacked: []string{"measure"},
+ },
+ {
+ name: "unresolvable last snapshot does not fail a successful run",
+ catalogs: []catalogData{
+ {catalog: commonv1.Catalog_CATALOG_STREAM},
+ {catalog: commonv1.Catalog_CATALOG_MEASURE},
+ },
+ extraUnknown: true,
+ wantBacked: []string{"stream", "measure"},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ dataRoot := t.TempDir()
+ destRoot := t.TempDir()
+ server := &fakeSnapshotServer{}
+ for _, c := range tt.catalogs {
+ catalogName := snapshot.CatalogName(c.catalog)
+ dir := filepath.Join(snapshot.LocalDir(dataRoot, c.catalog), storage.SnapshotsDir, snapshotName)
+ if err := os.MkdirAll(dir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ file := filepath.Join(dir, catalogName+".tm")
+ if err := os.WriteFile(file, []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if c.unreadable {
+ if err := os.Chmod(file, 0o000); err != nil {
+ t.Fatal(err)
+ }
+ }
+ server.snapshots = append(server.snapshots, &databasev1.Snapshot{Catalog: c.catalog, Name: snapshotName})
+ }
+ if tt.extraUnknown {
+ server.snapshots = append(server.snapshots, &databasev1.Snapshot{Catalog: commonv1.Catalog_CATALOG_UNSPECIFIED, Name: snapshotName})
+ }
+ addr := startSnapshotServer(t, server)
+
+ err := backupAction(context.Background(), backupOptions{
+ gRPCAddr: addr,
+ dest: "file://" + destRoot,
+ timeStyle: "daily",
+ streamRoot: dataRoot,
+ measureRoot: dataRoot,
+ propertyRoot: dataRoot,
+ traceRoot: dataRoot,
+ schemaRoot: dataRoot,
+ uploadConcurrency: 2,
+ })
+
+ if len(tt.wantErrs) == 0 {
+ if err != nil {
+ t.Fatalf("backupAction() error = %v, want nil", err)
+ }
+ } else {
+ if err == nil {
+ t.Fatalf("backupAction() error = nil, want failures of %v", tt.wantErrs)
+ }
+ for _, catalogName := range tt.wantErrs {
+ if !strings.Contains(err.Error(), catalogName+".tm") {
+ t.Errorf("backupAction() error = %v, want it to report the %s catalog", err, catalogName)
+ }
+ }
+ }
+ backed := backedUpCatalogs(t, destRoot)
+ for _, catalogName := range tt.wantBacked {
+ if _, ok := backed[catalogName]; !ok {
+ t.Errorf("catalog %s not backed up; backed up files = %v", catalogName, backed)
+ }
+ }
+ if len(backed) != len(tt.wantBacked) {
+ t.Errorf("backed up catalogs = %v, want %v", backed, tt.wantBacked)
+ }
+ })
+ }
+}
+
+func startSnapshotServer(t *testing.T, server databasev1.SnapshotServiceServer) string {
+ t.Helper()
+ lis, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ s := grpc.NewServer()
+ grpc_health_v1.RegisterHealthServer(s, health.NewServer())
+ databasev1.RegisterSnapshotServiceServer(s, server)
+ go func() {
+ _ = s.Serve(lis)
+ }()
+ t.Cleanup(s.Stop)
+ return lis.Addr().String()
+}
+
+// backedUpCatalogs maps each catalog found under the destination's <timeDir>/<catalog>/ to its uploaded files.
+func backedUpCatalogs(t *testing.T, destRoot string) map[string][]string {
+ t.Helper()
+ files, err := getAllFiles(destRoot)
+ if err != nil {
+ t.Fatal(err)
+ }
+ backed := make(map[string][]string)
+ for _, f := range files {
+ parts := strings.SplitN(f, "/", 3)
+ if len(parts) != 3 {
+ t.Fatalf("unexpected remote file layout: %s", f)
+ }
+ backed[parts[1]] = append(backed[parts[1]], parts[2])
+ }
+ return backed
+}