Backup cleanup: core implementation

pkg/restapi/backup.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
+	"strconv"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/render"
@@ -42,6 +43,7 @@ func newBackupHandler(services Services) *chi.Mux {
 		m.Get("/files", h.listFiles)
 	})
 	m.Get("/schema", h.describeSchema)
+	m.Delete("/cleanup", h.cleanup)
 
 	return m
 }
@@ -250,6 +252,32 @@ func (h backupHandler) describeSchema(w http.ResponseWriter, r *http.Request) {
 	render.Respond(w, r, convertSchema(cqlSchema))
 }
 
+func (h backupHandler) cleanup(w http.ResponseWriter, r *http.Request) {
+	cluster := mustClusterFromCtx(r)
+
+	var (
+		deleteDiskSnapshots bool
+		err                 error
+	)
+	if v := r.URL.Query().Get("delete_disk_snapshots"); v != "" {
+		deleteDiskSnapshots, err = strconv.ParseBool(v)
+		if err != nil {
+			respondBadRequest(w, r, util.ErrValidate(errors.Wrap(err, "parse delete_disk_snapshots")))
+			return
+		}
+	}
+	if !deleteDiskSnapshots {
+		respondBadRequest(w, r, util.ErrValidate(errors.New("no cleanup specified")))
+		return
+	}
+
+	if err := h.svc.Cleanup(r.Context(), cluster.ID); err != nil {
+		respondError(w, r, errors.Wrap(err, "perform cleanup"))
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
 func parseOptionalUUID(v string) (uuid.UUID, error) {
 	if v == "" {
 		return uuid.Nil, nil

pkg/restapi/services.go

@@ -69,6 +69,7 @@ type BackupService interface {
 	// GetProgress must work even when the cluster is no longer available.
 	GetProgress(ctx context.Context, clusterID, taskID, runID uuid.UUID) (backup.Progress, error)
 	DeleteSnapshot(ctx context.Context, clusterID uuid.UUID, locations []backupspec.Location, snapshotTags []string) error
+	Cleanup(ctx context.Context, clusterID uuid.UUID) error
 	GetValidationTarget(_ context.Context, clusterID uuid.UUID, properties json.RawMessage) (backup.ValidationTarget, error)
 	// GetValidationProgress must work even when the cluster is no longer available.
 	GetValidationProgress(ctx context.Context, clusterID, taskID, runID uuid.UUID) ([]backup.ValidationHostProgress, error)

pkg/service/backup/service.go

@@ -1168,6 +1168,16 @@ func (s *Service) DeleteSnapshot(ctx context.Context, clusterID uuid.UUID, locat
 	return nil
 }
 
+// Cleanup backup residues for specified cluster.
+// For now, it only removes SM snapshots from all nodes in the cluster.
+func (s *Service) Cleanup(ctx context.Context, clusterID uuid.UUID) error {
+	aw, err := s.newCleanupWorker(ctx, clusterID)
+	if err != nil {
+		return errors.Wrap(err, "create cleanup worker")
+	}
+	return errors.Wrap(aw.cleanupAll(ctx), "cleanup backup")
+}
+
 // NewSnapshotTag should be used instead of raw backupspec.NewSnapshotTag
 // for generating new snapshot tags for backups. It ensures that no two backups
 // (even for different clusters) generated by the same SM instance have the same

pkg/service/backup/service_backup_integration_test.go

@@ -2846,3 +2846,53 @@ func TestTGetDescribeSchemaIntegration(t *testing.T) {
 		})
 	}
 }
+
+func TestBackupCleanupIntegration(t *testing.T) {
+	const (
+		testBucket   = "backuptest-cleanup"
+		testKeyspace = "backuptest_cleanup"
+	)
+
+	var (
+		location       = s3Location(testBucket)
+		session        = CreateScyllaManagerDBSession(t)
+		h              = newBackupTestHelperWithUser(t, session, defaultConfig(), location, nil, "", "")
+		ctx            = context.Background()
+		clusterSession = CreateSessionAndDropAllKeyspaces(t, h.Client)
+	)
+
+	WriteData(t, clusterSession, testKeyspace, 1, "tab1", "tab2")
+
+	for i, host := range ManagedClusterHosts() {
+		// Make snapshots with different tags for more coverage
+		tag := backupspec.SnapshotTagAt(timeutc.Now().Add(-time.Duration(i) * time.Second))
+		if err := h.Client.TakeSnapshot(ctx, host, tag, testKeyspace); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	anySnapshotOnDisk := func() bool {
+		for _, host := range ManagedClusterHosts() {
+			tags, err := h.Client.Snapshots(ctx, host)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(tags) > 0 {
+				return true
+			}
+		}
+		return false
+	}
+
+	if !anySnapshotOnDisk() {
+		t.Fatal("Expected to find snapshot on disk after backup was stopped")
+	}
+
+	if err := h.service.Cleanup(ctx, h.ClusterID); err != nil {
+		t.Fatal(err)
+	}
+
+	if anySnapshotOnDisk() {
+		t.Fatal("Expected no snapshots on disk after cleanup")
+	}
+}

pkg/service/backup/worker_cleanup.go

@@ -0,0 +1,68 @@
+// Copyright (C) 2025 ScyllaDB
+
+package backup
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/scylladb/go-log"
+	"github.com/scylladb/scylla-manager/backupspec"
+	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
+	"github.com/scylladb/scylla-manager/v3/pkg/util/uuid"
+	"golang.org/x/sync/errgroup"
+)
+
+type cleanupWorker struct {
+	clusterID uuid.UUID
+
+	logger log.Logger
+	client *scyllaclient.Client
+
+	hosts []string
+}
+
+func (s *Service) newCleanupWorker(ctx context.Context, clusterID uuid.UUID) (*cleanupWorker, error) {
+	client, err := s.scyllaClient(ctx, clusterID)
+	if err != nil {
+		return nil, errors.Wrap(err, "get scylla client")
+	}
+	status, err := client.Status(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "get status")
+	}
+	return &cleanupWorker{
+		clusterID: clusterID,
+		logger:    s.logger.Named("cleanup_worker"),
+		client:    client,
+		hosts:     status.Up().Hosts(),
+	}, nil
+}
+
+func (w *cleanupWorker) cleanupAll(ctx context.Context) error {
+	eg, egCtx := errgroup.WithContext(ctx)
+	for _, host := range w.hosts {
+		eg.Go(func() error {
+			return errors.Wrapf(w.cleanupHost(egCtx, host), "host %s", host)
+		})
+	}
+	return eg.Wait()
+}
+
+func (w *cleanupWorker) cleanupHost(ctx context.Context, host string) error {
+	tags, err := w.client.Snapshots(ctx, host)
+	if err != nil {
+		return errors.Wrap(err, "list snapshots")
+	}
+	w.logger.Info(ctx, "Listed snapshots", "host", host, "tags", tags)
+
+	for _, tag := range tags {
+		if backupspec.IsSnapshotTag(tag) {
+			w.logger.Info(ctx, "Delete snapshot", "host", host, "tag", tag)
+			if err := w.client.DeleteSnapshot(ctx, host, tag); err != nil {
+				return errors.Wrapf(err, "delete snapshot %s", tag)
+			}
+		}
+	}
+	return nil
+}