Start export s3 and fixes for demo

.github/workflows/go-test.yml

@@ -22,7 +22,7 @@ jobs:
           go install google.golang.org/grpc/cmd/[email protected]
 
       - name: Compile protos
-        run: make
+        run: make proto
 
       - name: Build
         run: |

Makefile

@@ -1,9 +1,26 @@
-TOPTARGETS := all
+FILES ?= $(shell find . -type f -name '*.go')
+PACKAGES ?= $(shell go list ./...)
 
-SUBDIRS := pkg/proto
+.PHONY: all
 
-$(TOPTARGETS): $(SUBDIRS)
-$(SUBDIRS):
-	$(MAKE) -C $@ $(MAKECMDGOALS)
+all: test fmt lint vet proto build
 
-.PHONY: $(TOPTARGETS) $(SUBDIRS)
+proto:
+	$(MAKE) -C pkg/proto
+
+test:
+	go test -v ./... -short
+
+fmt:
+	go fmt ./...
+	goimports -w $(FILES)
+
+lint:
+	golint $(PACKAGES)
+
+vet:
+	go vet ./...
+
+build: ydbcp
+ydbcp:
+	go build -C cmd/ydbcp -o ydbcp

cmd/ydbcp/config.yaml

@@ -1,2 +1,18 @@
-ydbcp_db_connection_string: "grpc://localhost:2136/local"
 operation_ttl_seconds: 86400 # 24 hours
+
+db_connection:
+  connection_string: "grpcs://localhost:2135/domain/database"
+  insecure: true
+  discovery: false
+
+client_connection:
+  insecure: true
+  discovery: false
+
+s3:
+  endpoint: s3.endpoint.com
+  region: s3-region
+  bucket: s3-bucket
+  path_prefix: cluster-domain
+  access_key_id_path: path-to-s3-key
+  secret_access_key_path: path-to-s3-sec

cmd/ydbcp/main.go

@@ -8,9 +8,12 @@ import (
 	"net"
 	"os"
 	"os/signal"
+	"path"
 	"strconv"
+	"strings"
 	"sync"
 	"syscall"
+	"time"
 
 	table_types "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
 	_ "go.uber.org/automaxprocs"
@@ -38,8 +41,9 @@ var (
 type server struct {
 	pb.UnimplementedBackupServiceServer
 	pb.UnimplementedOperationServiceServer
-	driver db.DBConnector
-	s3     config.S3Config
+	driver     db.DBConnector
+	clientConn client.ClientConnector
+	s3         config.S3Config
 }
 
 func (s *server) GetBackup(ctx context.Context, request *pb.GetBackupRequest) (*pb.Backup, error) {
@@ -74,13 +78,65 @@ func (s *server) GetBackup(ctx context.Context, request *pb.GetBackupRequest) (*
 func (s *server) MakeBackup(ctx context.Context, req *pb.MakeBackupRequest) (*pb.Operation, error) {
 	xlog.Info(ctx, "MakeBackup", zap.String("request", req.String()))
 
+	clientConnectionParams := types.YdbConnectionParams{
+		Endpoint:     req.GetDatabaseEndpoint(),
+		DatabaseName: req.GetDatabaseName(),
+	}
+	dsn := types.MakeYdbConnectionString(clientConnectionParams)
+	client, err := s.clientConn.Open(ctx, dsn)
+	if err != nil {
+		// xlog.Error(ctx, "can't open client connection", zap.Error(err), zap.String("dsn", dsn))
+		return nil, fmt.Errorf("can't open client connection, dsn %s: %w", dsn, err)
+	}
+	defer func() {
+		if err := s.clientConn.Close(ctx, client); err != nil {
+			xlog.Error(ctx, "can't close client connection", zap.Error(err))
+		}
+	}()
+
+	accessKey, err := s.s3.AccessKey()
+	if err != nil {
+		xlog.Error(ctx, "can't get S3AccessKey", zap.Error(err))
+		return nil, fmt.Errorf("can't get S3AccessKey: %w", err)
+	}
+	secretKey, err := s.s3.SecretKey()
+	if err != nil {
+		xlog.Error(ctx, "can't get S3SecretKey", zap.Error(err))
+		return nil, fmt.Errorf("can't get S3SecretKey: %w", err)
+	}
+
+	dbNamePath := strings.Replace(req.DatabaseName, "/", "_", -1) // TODO: checking user imput
+	dbNamePath = strings.Trim(dbNamePath, "_")
+	dstPrefix := path.Join(s.s3.PathPrefix, dbNamePath)
+
+	s3Settings := types.ExportSettings{
+		Endpoint:            s.s3.Endpoint,
+		Region:              s.s3.Region,
+		Bucket:              s.s3.Bucket,
+		AccessKey:           accessKey,
+		SecretKey:           secretKey,
+		Description:         "ydbcp backup", // TODO: the description shoud be better
+		NumberOfRetries:     10,             // TODO: get it from configuration
+		SourcePaths:         req.GetSourcePaths(),
+		SourcePathToExclude: req.GetSourcePathsToExclude(),
+		DestinationPrefix:   s.s3.PathPrefix,
+		BackupID:            types.GenerateObjectID(), // TODO: do we need backup id?
+	}
+
+	clientOperationID, err := s.clientConn.ExportToS3(ctx, client, s3Settings)
+	if err != nil {
+		xlog.Error(ctx, "can't start export operation", zap.Error(err), zap.String("dns", dsn))
+		return nil, fmt.Errorf("can't start export operation, dsn %s: %w", dsn, err)
+	}
+	xlog.Debug(ctx, "export operation started", zap.String("clientOperationID", clientOperationID), zap.String("dsn", dsn))
+
 	backup := types.Backup{
 		ContainerID:  req.GetContainerId(),
 		DatabaseName: req.GetDatabaseName(),
 		S3Endpoint:   s.s3.Endpoint,
 		S3Region:     s.s3.Region,
 		S3Bucket:     s.s3.Bucket,
-		S3PathPrefix: s.s3.PathPrefix,
+		S3PathPrefix: dstPrefix,
 		Status:       types.BackupStatePending,
 	}
 	backupID, err := s.driver.CreateBackup(ctx, backup)
@@ -98,11 +154,13 @@ func (s *server) MakeBackup(ctx context.Context, req *pb.MakeBackupRequest) (*pb
 		ContainerID: req.ContainerId,
 		State:       types.OperationStatePending,
 		YdbConnectionParams: types.YdbConnectionParams{
-			Endpoint:     req.GetEndpoint(),
+			Endpoint:     req.GetDatabaseEndpoint(),
 			DatabaseName: req.GetDatabaseName(),
 		},
 		SourcePaths:         req.GetSourcePaths(),
 		SourcePathToExclude: req.GetSourcePathsToExclude(),
+		CreatedAt:           time.Now(),
+		YdbOperationId:      clientOperationID,
 	}
 
 	operationID, err := s.driver.CreateOperation(ctx, op)
@@ -248,10 +306,19 @@ func main() {
 	s := grpc.NewServer()
 	reflection.Register(s)
 
-	dbConnector := db.NewYdbConnector(configInstance)
+	dbConnector, err := db.NewYdbConnector(ctx, configInstance.DBConnection)
+	if err != nil {
+		xlog.Error(ctx, "Error init DBConnector", zap.Error(err))
+		os.Exit(1)
+	}
+	clientConnector := client.NewClientYdbConnector(configInstance.ClientConnection)
 
-	server := server{driver: dbConnector}
-	defer server.driver.Close()
+	server := server{
+		driver:     dbConnector,
+		clientConn: clientConnector,
+		s3:         configInstance.S3,
+	}
+	defer server.driver.Close(ctx)
 
 	pb.RegisterBackupServiceServer(s, &server)
 	pb.RegisterOperationServiceServer(s, &server)
@@ -271,9 +338,7 @@ func main() {
 	handlersRegistry := processor.NewOperationHandlerRegistry()
 	err = handlersRegistry.Add(
 		types.OperationTypeTB,
-		handlers.NewTBOperationHandler(
-			dbConnector, client.NewClientYdbConnector(), configInstance, queries.NewWriteTableQuery,
-		),
+		handlers.NewTBOperationHandler(dbConnector, clientConnector, configInstance, queries.NewWriteTableQuery),
 	)
 	if err != nil {
 		xlog.Error(ctx, "failed to register TB handler", zap.Error(err))
@@ -282,7 +347,7 @@ func main() {
 
 	err = handlersRegistry.Add(
 		types.OperationTypeRB,
-		handlers.NewRBOperationHandler(dbConnector, client.NewClientYdbConnector(), configInstance),
+		handlers.NewRBOperationHandler(dbConnector, clientConnector, configInstance),
 	)
 
 	if err != nil {

internal/config/config.go

@@ -3,7 +3,9 @@ package config
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
+	"strings"
 	"ydbcp/internal/util/xlog"
 
 	"go.uber.org/zap"
@@ -19,10 +21,24 @@ type S3Config struct {
 	SecretAccessKeyPath string `yaml:"secret_access_key_path"`
 }
 
+type YDBConnectionConfig struct {
+	ConnectionString   string `yaml:"connection_string"`
+	Insecure           bool   `yaml:"insecure"`
+	Discovery          bool   `yaml:"discovery" default:"true"`
+	DialTimeoutSeconds uint32 `yaml:"dial_timeout_seconds" default:"5"`
+}
+
+type ClientConnectionConfig struct {
+	Insecure           bool   `yaml:"insecure"`
+	Discovery          bool   `yaml:"discovery" default:"true"`
+	DialTimeoutSeconds uint32 `yaml:"dial_timeout_seconds" default:"5"`
+}
+
 type Config struct {
-	YdbcpDbConnectionString string   `yaml:"ydbcp_db_connection_string"`
-	S3                      S3Config `yaml:"s3"`
-	OperationTtlSeconds     int64    `yaml:"operation_ttl_seconds"`
+	DBConnection        YDBConnectionConfig    `yaml:"db_connection"`
+	ClientConnection    ClientConnectionConfig `yaml:"client_connection"`
+	S3                  S3Config               `yaml:"s3"`
+	OperationTtlSeconds int64                  `yaml:"operation_ttl_seconds"`
 }
 
 func (config Config) ToString() (string, error) {
@@ -54,3 +70,20 @@ func InitConfig(ctx context.Context, confPath string) (Config, error) {
 	}
 	return Config{}, errors.New("configuration file path is empty")
 }
+
+func readSecret(filename string) (string, error) {
+	rawSecret, err := os.ReadFile(filename)
+	if err != nil {
+		return "", fmt.Errorf("can't read file %s: %w", filename, err)
+	}
+	return strings.TrimSpace(string(rawSecret)), nil
+}
+
+func (c *S3Config) AccessKey() (string, error) {
+	return readSecret(c.AccessKeyIDPath)
+}
+
+func (c *S3Config) SecretKey() (string, error) {
+	return readSecret(c.SecretAccessKeyPath)
+
+}

internal/connectors/client/connector.go

@@ -3,15 +3,17 @@ package client
 import (
 	"context"
 	"fmt"
-	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
-	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
 	"path"
 	"regexp"
 	"strings"
 	"time"
+	"ydbcp/internal/config"
 	"ydbcp/internal/types"
 	"ydbcp/internal/util/xlog"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
+	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
+
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Export_V1"
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Import_V1"
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Operation_V1"
@@ -20,6 +22,7 @@ import (
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Import"
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Operations"
 	"github.com/ydb-platform/ydb-go-sdk/v3"
+	"github.com/ydb-platform/ydb-go-sdk/v3/balancers"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/types/known/durationpb"
 )
@@ -36,18 +39,32 @@ type ClientConnector interface {
 }
 
 type ClientYdbConnector struct {
+	config config.ClientConnectionConfig
 }
 
-func NewClientYdbConnector() *ClientYdbConnector {
-	return &ClientYdbConnector{}
+func NewClientYdbConnector(config config.ClientConnectionConfig) *ClientYdbConnector {
+	return &ClientYdbConnector{
+		config: config,
+	}
 }
 
 func (d *ClientYdbConnector) Open(ctx context.Context, dsn string) (*ydb.Driver, error) {
 	xlog.Info(ctx, "Connecting to client db", zap.String("dsn", dsn))
-	db, connErr := ydb.Open(ctx, dsn, ydb.WithAnonymousCredentials())
 
-	if connErr != nil {
-		return nil, fmt.Errorf("error connecting to client db: %s", connErr.Error())
+	opts := []ydb.Option{
+		ydb.WithAnonymousCredentials(),
+		ydb.WithDialTimeout(time.Second * time.Duration(d.config.DialTimeoutSeconds)),
+	}
+	if d.config.Insecure {
+		opts = append(opts, ydb.WithTLSSInsecureSkipVerify())
+	}
+	if !d.config.Discovery {
+		opts = append(opts, ydb.WithBalancer(balancers.SingleConn()))
+	}
+
+	db, err := ydb.Open(ctx, dsn, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("error connecting to client db: %w", err)
 	}
 
 	return db, nil
@@ -60,9 +77,8 @@ func (d *ClientYdbConnector) Close(ctx context.Context, clientDb *ydb.Driver) er
 
 	xlog.Info(ctx, "Closing client db driver")
 	err := clientDb.Close(ctx)
-
 	if err != nil {
-		return fmt.Errorf("error closing client db driver: %s", err.Error())
+		return fmt.Errorf("error closing client db driver: %w", err)
 	}
 
 	return nil
@@ -206,7 +222,7 @@ func (d *ClientYdbConnector) ExportToS3(ctx context.Context, clientDb *ydb.Drive
 		zap.String("description", s3Settings.Description),
 	)
 
-	response, exportErr := exportClient.ExportToS3(
+	response, err := exportClient.ExportToS3(
 		ctx,
 		&Ydb_Export.ExportToS3Request{
 			OperationParams: &Ydb_Operations.OperationParams{
@@ -225,13 +241,12 @@ func (d *ClientYdbConnector) ExportToS3(ctx context.Context, clientDb *ydb.Drive
 		},
 	)
 
-	if exportErr != nil {
-		return "", fmt.Errorf("error exporting to S3: %s", exportErr.Error())
+	if err != nil {
+		return "", fmt.Errorf("error exporting to S3: %w", err)
 	}
 
 	if response.GetOperation().GetStatus() != Ydb.StatusIds_SUCCESS {
-		return "", fmt.Errorf("exporting to S3 was failed: %v",
-			response.GetOperation().GetIssues())
+		return "", fmt.Errorf("exporting to S3 was failed: %v", response.GetOperation().GetIssues())
 	}
 
 	return response.GetOperation().GetId(), nil
@@ -249,7 +264,7 @@ func (d *ClientYdbConnector) ImportFromS3(ctx context.Context, clientDb *ydb.Dri
 		zap.String("description", s3Settings.Description),
 	)
 
-	response, importErr := importClient.ImportFromS3(
+	response, err := importClient.ImportFromS3(
 		ctx,
 		&Ydb_Import.ImportFromS3Request{
 			OperationParams: &Ydb_Operations.OperationParams{
@@ -260,8 +275,8 @@ func (d *ClientYdbConnector) ImportFromS3(ctx context.Context, clientDb *ydb.Dri
 		},
 	)
 
-	if importErr != nil {
-		return "", fmt.Errorf("error importing from s3: %s", importErr.Error())
+	if err != nil {
+		return "", fmt.Errorf("error importing from s3: %w", err)
 	}
 
 	if response.GetOperation().GetStatus() != Ydb.StatusIds_SUCCESS {