feat(make_backup): add root_path for backup

Author: ulya-sidorina

.github/workflows/unit-test.yml

@@ -93,3 +93,20 @@ jobs:
       - name: docker compose down
         run: |
           docker compose down
+      - name: docker compose up     
+        if: ${{ matrix.enable_new_paths_format }}
+        run: |
+          docker compose up -d
+      - name: run new_paths_format tests
+        if: ${{ matrix.enable_new_paths_format }}
+        run: |
+          while [ "$(docker inspect -f {{.State.Health.Status}} local-ydbcp)" != "healthy" ]; do
+            echo "Waiting for container to become healthy..."
+            sleep 1
+          done
+          echo "Starting new_paths_format tests!"
+          docker exec local-ydbcp sh -c './test_new_paths_format'
+      - name: docker compose down
+        if: ${{ matrix.enable_new_paths_format }}
+        run: |
+          docker compose down

Dockerfile

@@ -20,6 +20,7 @@ RUN go build -o . ./cmd/ydbcp/main.go
 RUN go build -o ./make_backup ./cmd/integration/make_backup/main.go
 RUN go build -o ./list_entities ./cmd/integration/list_entities/main.go
 RUN go build -o ./orm ./cmd/integration/orm/main.go
+RUN go build -o ./test_new_paths_format ./cmd/integration/new_paths_format/main.go
 
 # Command to run the executable
 CMD ["./main", "--config=local_config.yaml"]

cmd/integration/new_paths_format/main.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+	"ydbcp/cmd/integration/common"
+	pb "ydbcp/pkg/proto/ydbcp/v1alpha1"
+
+	"google.golang.org/grpc"
+)
+
+const (
+	containerID      = "abcde"
+	databaseName     = "/local"
+	ydbcpEndpoint    = "0.0.0.0:50051"
+	databaseEndpoint = "grpcs://local-ydb:2135"
+)
+
+type backupScenario struct {
+	name                string
+	request             *pb.MakeBackupRequest
+	expectedRootPath    string
+	expectedSourcePaths []string
+}
+
+func newMakeBackupRequest(rootPath string, sourcePaths []string) *pb.MakeBackupRequest {
+	return &pb.MakeBackupRequest{
+		ContainerId:      containerID,
+		DatabaseName:     databaseName,
+		DatabaseEndpoint: databaseEndpoint,
+		RootPath:         rootPath,
+		SourcePaths:      sourcePaths,
+	}
+}
+
+func runBackupScenario(ctx context.Context, backupClient pb.BackupServiceClient, opClient pb.OperationServiceClient, scenario backupScenario) {
+	tbwr, err := backupClient.MakeBackup(ctx, scenario.request)
+	if err != nil {
+		log.Panicf("scenario %s: failed to make backup: %v", scenario.name, err)
+	}
+	op, err := opClient.GetOperation(
+		ctx, &pb.GetOperationRequest{
+			Id: tbwr.Id,
+		},
+	)
+	if err != nil {
+		log.Panicf("scenario %s: failed to get operation: %v", scenario.name, err)
+	}
+
+	if op.GetRootPath() != scenario.expectedRootPath {
+		log.Panicf("scenario %s: expected root path %q, got %q", scenario.name, scenario.expectedRootPath, op.GetRootPath())
+	}
+
+	if !equalStringSlices(op.GetSourcePaths(), scenario.expectedSourcePaths) {
+		log.Panicf("scenario %s: expected source paths %v, got %v", scenario.name, scenario.expectedSourcePaths, op.GetSourcePaths())
+	}
+
+	log.Printf("scenario %s: passed", scenario.name)
+}
+
+func equalStringSlices(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func main() {
+	conn := common.CreateGRPCClient(ydbcpEndpoint)
+	defer func(conn *grpc.ClientConn) {
+		err := conn.Close()
+		if err != nil {
+			log.Panicln("failed to close connection")
+		}
+	}(conn)
+	backupClient := pb.NewBackupServiceClient(conn)
+	opClient := pb.NewOperationServiceClient(conn)
+
+	ctx := context.Background()
+
+	scenarios := []backupScenario{
+		{
+			name:                "full backup",
+			request:             newMakeBackupRequest("", nil),
+			expectedRootPath:    "",
+			expectedSourcePaths: []string{},
+		},
+		{
+			name:                "full backup with specified root path",
+			request:             newMakeBackupRequest("stocks", nil),
+			expectedRootPath:    "stocks",
+			expectedSourcePaths: []string{},
+		},
+		{
+			name:                "partial backup",
+			request:             newMakeBackupRequest("", []string{"kv_test"}),
+			expectedRootPath:    "",
+			expectedSourcePaths: []string{"kv_test"},
+		},
+		{
+			name:                "partial backup with specified root path",
+			request:             newMakeBackupRequest("stocks", []string{"orders", "orderLines"}),
+			expectedRootPath:    "stocks",
+			expectedSourcePaths: []string{"orders", "orderLines"},
+		},
+	}
+
+	for _, scenario := range scenarios {
+		runBackupScenario(ctx, backupClient, opClient, scenario)
+		time.Sleep(2 * time.Second)
+	}
+}

init_db/create_tables.sh

@@ -3,4 +3,16 @@
 
 # create and fill user table kv_test
 ./ydb -e ${YDB_ENDPOINT} -d /local workload kv init
-./ydb -e ${YDB_ENDPOINT} -d /local workload kv run upsert --rows 100
+./ydb -e ${YDB_ENDPOINT} -d /local workload kv run upsert --rows 100
+
+# create and fill user tables: stock, orders, orderLines
+./ydb -e ${YDB_ENDPOINT} -d /local workload stock init -p 10 -q 10 -o 10
+
+# create directory for user tables
+./ydb -e ${YDB_ENDPOINT} -d /local scheme mkdir stocks
+
+# move user tables (stock, orders, orderLines)  to separate directory
+./ydb -e ${YDB_ENDPOINT} -d /local tools rename \
+  --item source=/local/stock,destination=/local/stocks/stock \
+  --item source=/local/orders,destination=/local/stocks/orders \
+  --item source=/local/orderLines,destination=/local/stocks/orderLines

internal/backup_operations/make_backup.go

@@ -30,6 +30,7 @@ type MakeBackupInternalRequest struct {
 	ContainerID          string
 	DatabaseEndpoint     string
 	DatabaseName         string
+	RootPath             string
 	SourcePaths          []string
 	SourcePathsToExclude []string
 	ScheduleID           *string
@@ -42,6 +43,7 @@ func FromBackupSchedule(schedule *types.BackupSchedule) MakeBackupInternalReques
 		ContainerID:          schedule.ContainerID,
 		DatabaseEndpoint:     schedule.DatabaseEndpoint,
 		DatabaseName:         schedule.DatabaseName,
+		RootPath:             schedule.RootPath,
 		SourcePaths:          schedule.SourcePaths,
 		SourcePathsToExclude: schedule.SourcePathsToExclude,
 		ScheduleID:           &schedule.ID,
@@ -57,6 +59,7 @@ func FromTBWROperation(tbwr *types.TakeBackupWithRetryOperation) MakeBackupInter
 		ContainerID:          tbwr.ContainerID,
 		DatabaseEndpoint:     tbwr.YdbConnectionParams.Endpoint,
 		DatabaseName:         tbwr.YdbConnectionParams.DatabaseName,
+		RootPath:             tbwr.RootPath,
 		SourcePaths:          tbwr.SourcePaths,
 		SourcePathsToExclude: tbwr.SourcePathsToExclude,
 		ScheduleID:           tbwr.ScheduleID,
@@ -141,9 +144,14 @@ func ValidateSourcePaths(
 	if req.ScheduleID != nil {
 		ctx = xlog.With(ctx, zap.String("ScheduleID", *req.ScheduleID))
 	}
+	basePath, ok := SafePathJoin(req.DatabaseName, req.RootPath)
+	if !ok {
+		xlog.Error(ctx, "incorrect root path", zap.String("path", req.RootPath))
+		return nil, status.Errorf(codes.InvalidArgument, "incorrect root path %s", req.RootPath)
+	}
 	sourcePaths := make([]string, 0, len(req.SourcePaths))
 	for _, p := range req.SourcePaths {
-		fullPath, ok := SafePathJoin(req.DatabaseName, p)
+		fullPath, ok := SafePathJoin(basePath, p)
 		if !ok {
 			xlog.Error(ctx, "incorrect source path", zap.String("path", p))
 			return nil, status.Errorf(codes.InvalidArgument, "incorrect source path %s", p)
@@ -345,6 +353,7 @@ func MakeBackup(
 		SecretKey:         secretKey,
 		Description:       "ydbcp backup", // TODO: the description shoud be better
 		NumberOfRetries:   10,             // TODO: get it from configuration
+		RootPath:          req.RootPath,
 		SourcePaths:       pathsForExport,
 		DestinationPrefix: destinationPrefix,
 		S3ForcePathStyle:  s3.S3ForcePathStyle,
@@ -393,6 +402,7 @@ func MakeBackup(
 			Endpoint:     req.DatabaseEndpoint,
 			DatabaseName: req.DatabaseName,
 		},
+		RootPath:             req.RootPath,
 		SourcePaths:          req.SourcePaths,
 		SourcePathsToExclude: req.SourcePathsToExclude,
 		Audit: &pb.AuditInfo{

internal/connectors/client/connector.go

@@ -268,7 +268,7 @@ func (d *ClientYdbConnector) ExportToS3(
 	}
 
 	if featureFlags.EnableNewPathsFormat {
-		exportRequest.Settings.SourcePath = clientDb.Name()
+		exportRequest.Settings.SourcePath = path.Join(clientDb.Name(), s3Settings.RootPath)
 		exportRequest.Settings.DestinationPrefix = s3Settings.DestinationPrefix
 	}
 

internal/connectors/db/process_result_set.go

@@ -143,6 +143,7 @@ func ReadOperationFromResultSet(res query.Row) (types.Operation, error) {
 		ydbOperationId       *string
 		operationStateBuf    *string
 		message              *string
+		rootPath             *string
 		sourcePaths          *string
 		sourcePathsToExclude *string
 		creator              *string
@@ -168,6 +169,7 @@ func ReadOperationFromResultSet(res query.Row) (types.Operation, error) {
 		query.Named("operation_id", &ydbOperationId),
 		query.Named("status", &operationStateBuf),
 		query.Named("message", &message),
+		query.Named("root_path", &rootPath),
 		query.Named("paths", &sourcePaths),
 		query.Named("paths_to_exclude", &sourcePathsToExclude),
 
@@ -223,6 +225,7 @@ func ReadOperationFromResultSet(res query.Row) (types.Operation, error) {
 				Endpoint:     databaseEndpoint,
 				DatabaseName: databaseName,
 			},
+			RootPath:             StringOrEmpty(rootPath),
 			SourcePaths:          sourcePathsSlice,
 			SourcePathsToExclude: sourcePathsToExcludeSlice,
 			YdbOperationId:       StringOrEmpty(ydbOperationId),
@@ -301,6 +304,7 @@ func ReadOperationFromResultSet(res query.Row) (types.Operation, error) {
 					Endpoint:     databaseEndpoint,
 					DatabaseName: databaseName,
 				},
+				RootPath:             StringOrEmpty(rootPath),
 				SourcePaths:          sourcePathsSlice,
 				SourcePathsToExclude: sourcePathsToExcludeSlice,
 				Audit:                auditFromDb(creator, createdAt, completedAt),
@@ -330,6 +334,7 @@ func ReadBackupScheduleFromResultSet(res query.Row, withRPOInfo bool) (*types.Ba
 		createdAt              *time.Time
 		name                   *string
 		ttl                    *time.Duration
+		rootPath               *string
 		sourcePaths            *string
 		sourcePathsToExclude   *string
 		recoveryPointObjective *time.Duration
@@ -351,6 +356,7 @@ func ReadBackupScheduleFromResultSet(res query.Row, withRPOInfo bool) (*types.Ba
 		query.Named("created_at", &createdAt),
 		query.Named("name", &name),
 		query.Named("ttl", &ttl),
+		query.Named("root_path", &rootPath),
 		query.Named("paths", &sourcePaths),
 		query.Named("paths_to_exclude", &sourcePathsToExclude),
 		query.Named("recovery_point_objective", &recoveryPointObjective),
@@ -400,6 +406,7 @@ func ReadBackupScheduleFromResultSet(res query.Row, withRPOInfo bool) (*types.Ba
 		ContainerID:          containerID,
 		DatabaseName:         databaseName,
 		DatabaseEndpoint:     databaseEndpoint,
+		RootPath:             StringOrEmpty(rootPath),
 		SourcePaths:          sourcePathsSlice,
 		SourcePathsToExclude: sourcePathsToExcludeSlice,
 		Audit:                auditFromDb(initiated, createdAt, nil),

internal/connectors/db/yql/queries/write.go

@@ -117,6 +117,9 @@ func BuildCreateOperationQuery(operation types.Operation, index int) WriteSingle
 			"$operation_id",
 			table_types.StringValueFromString(tb.YdbOperationId),
 		)
+		if len(tb.RootPath) > 0 {
+			d.AddValueParam("$root_path", table_types.StringValueFromString(tb.RootPath))
+		}
 		if len(tb.SourcePaths) > 0 {
 			d.AddValueParam("$paths", table_types.StringValueFromString(types.SerializeSourcePaths(tb.SourcePaths)))
 		}
@@ -146,6 +149,9 @@ func BuildCreateOperationQuery(operation types.Operation, index int) WriteSingle
 			"$endpoint",
 			table_types.StringValueFromString(tbwr.YdbConnectionParams.Endpoint),
 		)
+		if len(tbwr.RootPath) > 0 {
+			d.AddValueParam("$root_path", table_types.StringValueFromString(tbwr.RootPath))
+		}
 		if len(tbwr.SourcePaths) > 0 {
 			d.AddValueParam("$paths", table_types.StringValueFromString(types.SerializeSourcePaths(tbwr.SourcePaths)))
 		}
@@ -357,6 +363,9 @@ func BuildCreateBackupScheduleQuery(schedule types.BackupSchedule, index int) Wr
 	if schedule.ScheduleSettings.Ttl != nil {
 		d.AddValueParam("$ttl", table_types.IntervalValueFromDuration(schedule.ScheduleSettings.Ttl.AsDuration()))
 	}
+	if len(schedule.RootPath) > 0 {
+		d.AddValueParam("$root_path", table_types.StringValueFromString(schedule.RootPath))
+	}
 	if len(schedule.SourcePaths) > 0 {
 		d.AddValueParam("$paths", table_types.StringValueFromString(types.SerializeSourcePaths(schedule.SourcePaths)))
 	}

internal/connectors/db/yql/schema/create_tables.yql

@@ -51,6 +51,7 @@ CREATE TABLE Operations (
     status String,
     message String,
 
+    root_path String,
     paths String,
     paths_to_exclude String,
     operation_id String,
@@ -79,6 +80,7 @@ CREATE TABLE BackupSchedules (
 
     crontab String NOT NULL,
     ttl Interval,
+    root_path String,
     paths String,
     paths_to_exclude String,
 

internal/handlers/schedule_backup.go

@@ -67,6 +67,7 @@ func BackupScheduleHandler(
 					Endpoint:     schedule.DatabaseEndpoint,
 					DatabaseName: schedule.DatabaseName,
 				},
+				RootPath:             schedule.RootPath,
 				SourcePaths:          schedule.SourcePaths,
 				SourcePathsToExclude: schedule.SourcePathsToExclude,
 				Audit: &pb.AuditInfo{