cloud/disk_manager/pkg/schema/dataplane.go - main - githubmirror/nbs

cloud/disk_manager/pkg/schema/dataplane.go
200 строк5 KB
Sergei
issue-1559: [Disk Manager] Add filesystem backup/restore tasks (#5628)
04 июн 2026, 14:40
04 июн 2026, 14:407ac2f26
package schema

import (
	"context"

	"github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/auth"
	server_config "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/configs/server/config"
	filesystem_config "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/dataplane/filesystem/config"
	nodes_schema "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/storage/schema"
	filesystem_traversal_schema "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/dataplane/filesystem/traversal/storage/schema"
	"github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/dataplane/snapshot/storage/schema"
	"github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/monitoring/metrics"
	"github.com/ydb-platform/nbs/cloud/tasks/persistence"
	tasks_storage "github.com/ydb-platform/nbs/cloud/tasks/storage"
)

/////////////////////////////////////////////////////////////////////////////////

func initDataplane(
	ctx context.Context,
	config *server_config.ServerConfig,
	creds auth.Credentials,
	db *persistence.YDBClient,
	dropUnusedColumns bool,
) error {

	err := tasks_storage.CreateYDBTables(
		ctx,
		config.GetTasksConfig(),
		db,
		dropUnusedColumns,
	)
	if err != nil {
		return err
	}

	snapshotConfig := config.GetDataplaneConfig().GetSnapshotConfig()

	snapshotDB, err := persistence.NewYDBClient(
		ctx,
		snapshotConfig.GetPersistenceConfig(),
		metrics.NewEmptyRegistry(),
		persistence.WithCredentials(creds),
	)
	if err != nil {
		return err
	}
	defer snapshotDB.Close(ctx)

	s3Config := snapshotConfig.GetPersistenceConfig().GetS3Config()
	var s3 *persistence.S3Client
	// TODO: remove when s3 will always be initialized.
	if s3Config != nil {
		s3, err = persistence.NewS3ClientFromConfig(
			s3Config,
			metrics.NewEmptyRegistry(),
			nil, // availabilityMonitoring
		)
		if err != nil {
			return err
		}
	}

	err = schema.Create(ctx, snapshotConfig, snapshotDB, s3, dropUnusedColumns)
	if err != nil {
		return err
	}

	migrationDstSnapshotConfig := config.GetDataplaneConfig().GetMigrationDstSnapshotConfig()
	if migrationDstSnapshotConfig == nil {
		return nil
	}

	migrationDstPersistenceConfig := migrationDstSnapshotConfig.GetPersistenceConfig()
	migrationDstDB, err := persistence.NewYDBClient(
		ctx,
		migrationDstPersistenceConfig,
		metrics.NewEmptyRegistry(),
		persistence.WithCredentials(creds),
	)
	if err != nil {
		return err
	}

	migrationDstS3Config := migrationDstPersistenceConfig.GetS3Config()
	var migrationDstS3 *persistence.S3Client
	if migrationDstS3Config != nil {
		migrationDstS3, err = persistence.NewS3ClientFromConfig(
			migrationDstS3Config,
			metrics.NewEmptyRegistry(),
			nil, // availabilityMonitoring
		)
		if err != nil {
			return err
		}
	}
	return schema.Create(
		ctx,
		migrationDstSnapshotConfig,
		migrationDstDB,
		migrationDstS3,
		dropUnusedColumns,
	)
}

func initFilesystemDataplane(
	ctx context.Context,
	config *server_config.ServerConfig,
	creds auth.Credentials,
	dropUnusedColumns bool,
) error {

	filesystemConfig := config.GetDataplaneConfig().GetFilesystemConfig()
	if filesystemConfig == nil {
		return nil
	}

	persistenceConfig := filesystemConfig.GetPersistenceConfig()
	if persistenceConfig == nil {
		return nil
	}

	filesystemDB, err := persistence.NewYDBClient(
		ctx,
		persistenceConfig,
		metrics.NewEmptyRegistry(),
		persistence.WithCredentials(creds),
	)
	if err != nil {
		return err
	}
	defer filesystemDB.Close(ctx)

	err = initFilesystemScrubbingSchema(ctx, filesystemConfig, filesystemDB, dropUnusedColumns)
	if err != nil {
		return err
	}

	return initFilesystemSnapshotSchema(ctx, filesystemConfig, filesystemDB, dropUnusedColumns)
}

func initFilesystemScrubbingSchema(
	ctx context.Context,
	filesystemConfig *filesystem_config.FilesystemDataplaneConfig,
	filesystemDB *persistence.YDBClient,
	dropUnusedColumns bool,
) error {

	scrubbingConfig := filesystemConfig.GetScrubbingConfig()
	if scrubbingConfig == nil {
		return nil
	}

	traversalConfig := scrubbingConfig.GetTraversalConfig()
	if traversalConfig == nil {
		return nil
	}

	return filesystem_traversal_schema.Create(
		ctx,
		traversalConfig.GetStorageFolder(),
		filesystemDB,
		dropUnusedColumns,
	)
}

func initFilesystemSnapshotSchema(
	ctx context.Context,
	filesystemConfig *filesystem_config.FilesystemDataplaneConfig,
	filesystemDB *persistence.YDBClient,
	dropUnusedColumns bool,
) error {

	snapshotConfig := filesystemConfig.GetSnapshotConfig()
	if snapshotConfig == nil {
		return nil
	}

	traversalConfig := snapshotConfig.GetTraversalConfig()
	if traversalConfig == nil {
		return nil
	}

	err := filesystem_traversal_schema.Create(
		ctx,
		traversalConfig.GetStorageFolder(),
		filesystemDB,
		dropUnusedColumns,
	)
	if err != nil {
		return err
	}

	return nodes_schema.Create(
		ctx,
		snapshotConfig.GetNodesStorageFolder(),
		filesystemDB,
		dropUnusedColumns,
	)
}