Initial commit

cmd/transferctl/cobraaux/cobraaux.go

@@ -0,0 +1,33 @@
+package cobraaux
+
+import (
+	"github.com/doublecloud/tross/internal/core/xerrors"
+	"github.com/spf13/cobra"
+)
+
+// RegisterCommand is like parent.AddCommand(child), but also
+// makes chaining of PersistentPreRunE and PersistentPreRun
+func RegisterCommand(parent, child *cobra.Command) {
+	parentPpre := parent.PersistentPreRunE
+	childPpre := child.PersistentPreRunE
+	if child.PersistentPreRunE == nil && child.PersistentPreRun != nil {
+		childPpre = func(cmd *cobra.Command, args []string) error {
+			child.PersistentPreRun(cmd, args)
+			return nil
+		}
+	}
+	if childPpre != nil {
+		child.PersistentPreRunE = func(cmd *cobra.Command, args []string) error {
+			if parentPpre != nil {
+				err := parentPpre(cmd, args)
+				if err != nil {
+					return xerrors.Errorf("cannot process parent PersistentPreRunE: %w", err)
+				}
+			}
+			return childPpre(cmd, args)
+		}
+	} else if parentPpre != nil {
+		child.PersistentPreRunE = parentPpre
+	}
+	parent.AddCommand(child)
+}

cmd/trcli/activate/activate.go

@@ -0,0 +1,45 @@
+package activate
+
+import (
+	"context"
+
+	"github.com/doublecloud/tross/cmd/trcli/config"
+	"github.com/doublecloud/tross/internal/core/metrics/solomon"
+	"github.com/doublecloud/tross/internal/core/xerrors"
+	"github.com/doublecloud/tross/pkg/abstract/coordinator"
+	server "github.com/doublecloud/tross/pkg/abstract/model"
+	"github.com/doublecloud/tross/pkg/worker/tasks"
+	"github.com/spf13/cobra"
+)
+
+func ActivateCommand() *cobra.Command {
+	var transferParams string
+	activationCommand := &cobra.Command{
+		Use:   "activate",
+		Short: "Activate transfer locally",
+		Args:  cobra.MatchAll(cobra.ExactArgs(0)),
+		RunE:  activate(&transferParams),
+	}
+	activationCommand.Flags().StringVar(&transferParams, "transfer", "./transfer.yaml", "path to yaml file with transfer configuration")
+	return activationCommand
+}
+
+func activate(transferYaml *string) func(cmd *cobra.Command, args []string) error {
+	return func(cmd *cobra.Command, args []string) error {
+		transfer, err := config.TransferFromYaml(transferYaml)
+		if err != nil {
+			return xerrors.Errorf("unable to load transfer: %w", err)
+		}
+		return RunActivate(transfer)
+	}
+}
+
+func RunActivate(transfer *server.Transfer) error {
+	return tasks.ActivateDelivery(
+		context.Background(),
+		nil,
+		coordinator.NewFakeClient(),
+		*transfer,
+		solomon.NewRegistry(solomon.NewRegistryOpts()),
+	)
+}

cmd/trcli/activate/tests/ch_init.sql

@@ -0,0 +1 @@
+CREATE DATABASE trcli_activate_test_ch;

cmd/trcli/activate/tests/dump/pg_init.sql

@@ -0,0 +1,5 @@
+CREATE TABLE public.t2(i INT PRIMARY KEY, f REAL);
+INSERT INTO public.t2(i, f) VALUES (1, 1.0), (2, 4.0);
+
+CREATE TABLE public.t3(i INT PRIMARY KEY, f REAL);
+INSERT INTO public.t3(i, f) VALUES (1, 1.0), (2, 4.0), (3, 5.0), (4, 4.0), (5, 6.0);

cmd/trcli/activate/tests/pg2ch_test.go

@@ -0,0 +1,41 @@
+package tests
+
+import (
+	_ "embed"
+	"testing"
+	"time"
+
+	"github.com/doublecloud/tross/cmd/trcli/activate"
+	"github.com/doublecloud/tross/cmd/trcli/config"
+	chrecipe "github.com/doublecloud/tross/pkg/providers/clickhouse/recipe"
+	"github.com/doublecloud/tross/pkg/providers/postgres/pgrecipe"
+	"github.com/doublecloud/tross/tests/helpers"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed transfer.yaml
+var transferYaml []byte
+
+func TestActivate(t *testing.T) {
+	src := pgrecipe.RecipeSource(
+		pgrecipe.WithPrefix("SRC"),
+		pgrecipe.WithFiles("dump/pg_init.sql"),
+	)
+
+	dst, err := chrecipe.Target(
+		chrecipe.WithInitFile("ch_init.sql"),
+		chrecipe.WithDatabase("trcli_activate_test_ch"),
+	)
+	require.NoError(t, err)
+
+	transfer, err := config.ParseTransfer(transferYaml)
+	require.NoError(t, err)
+
+	transfer.Src = src
+	transfer.Dst = dst
+
+	require.NoError(t, activate.RunActivate(transfer))
+
+	require.NoError(t, helpers.WaitDestinationEqualRowsCount(dst.Database, "t2", helpers.GetSampleableStorageByModel(t, dst), 60*time.Second, 2))
+	require.NoError(t, helpers.WaitDestinationEqualRowsCount(dst.Database, "t3", helpers.GetSampleableStorageByModel(t, dst), 60*time.Second, 5))
+}

cmd/trcli/activate/tests/transfer.yaml

@@ -0,0 +1,18 @@
+id: dtttest
+transfername: pg-to-ch-test
+description: ""
+type: SNAPSHOT_ONLY
+src:
+  id: dtetestsource
+  name: pg-source
+  type: pg
+  params: |
+    {
+    }
+dst:
+  id: dtetesttarget
+  name: ch-target
+  type: ch
+  params: |
+    {
+    }

cmd/trcli/check/check.go

@@ -0,0 +1,82 @@
+package check
+
+import (
+	"context"
+
+	"github.com/doublecloud/tross/cmd/trcli/config"
+	"github.com/doublecloud/tross/internal/core/xerrors"
+	"github.com/doublecloud/tross/internal/logger"
+	server "github.com/doublecloud/tross/pkg/abstract/model"
+	"github.com/doublecloud/tross/pkg/worker/tasks"
+	"github.com/spf13/cobra"
+	"gopkg.in/yaml.v2"
+)
+
+func CheckCommand() *cobra.Command {
+	var transferParams string
+	checkCommand := &cobra.Command{
+		Use:   "check",
+		Short: "Check transfer locally",
+		Args:  cobra.MatchAll(cobra.ExactArgs(0)),
+		RunE:  check(&transferParams),
+	}
+	checkCommand.Flags().StringVar(&transferParams, "transfer", "./transfer.yaml", "path to yaml file with transfer configuration")
+	return checkCommand
+}
+
+func check(transferYaml *string) func(cmd *cobra.Command, args []string) error {
+	return func(cmd *cobra.Command, args []string) error {
+		transfer, err := config.TransferFromYaml(transferYaml)
+		if err != nil {
+			return xerrors.Errorf("unable to load transfer: %w", err)
+		}
+		return RunCheck(transfer)
+	}
+}
+
+func testSource(transfer server.Transfer) error {
+	transformationJSON, err := transfer.TransformationJSON()
+	if err != nil {
+		return xerrors.Errorf("unable to serialize transformation: %w", err)
+	}
+	res := tasks.TestEndpoint(context.Background(), &tasks.TestEndpointParams{
+		Transfer:             &transfer,
+		IsSource:             true,
+		EndpointType:         transfer.SrcType(),
+		EndpointConfig:       transfer.SrcJSON(),
+		TransformationConfig: transformationJSON,
+	})
+
+	prettyRes, err := yaml.Marshal(res)
+	if err != nil {
+		return err
+	}
+
+	logger.Log.Infof("check source done: found %d objects:\n\n%s", len(res.Schema), string(prettyRes))
+	return nil
+}
+
+func testTarget(transfer server.Transfer) error {
+	if res := tasks.TestEndpoint(context.Background(), &tasks.TestEndpointParams{
+		Transfer:             &transfer,
+		IsSource:             false,
+		EndpointType:         transfer.DstType(),
+		EndpointConfig:       transfer.DstJSON(),
+		TransformationConfig: nil,
+	}); res.Err() != nil {
+		return xerrors.Errorf("unable to check target: %w", res.Err())
+	}
+	logger.Log.Infof("check target done")
+	return nil
+}
+
+func RunCheck(transfer *server.Transfer) error {
+	if err := testSource(*transfer); err != nil {
+		return xerrors.Errorf("failed testing source: %w", err)
+	}
+
+	if err := testTarget(*transfer); err != nil {
+		return xerrors.Errorf("failed testing target: %w", err)
+	}
+	return nil
+}

cmd/trcli/check/tests/dump/pg_init.sql

@@ -0,0 +1,5 @@
+CREATE TABLE public.t2(i INT PRIMARY KEY, f REAL);
+INSERT INTO public.t2(i, f) VALUES (1, 1.0), (2, 4.0);
+
+CREATE TABLE public.t3(i INT PRIMARY KEY, f REAL);
+INSERT INTO public.t3(i, f) VALUES (1, 1.0), (2, 4.0), (3, 5.0), (4, 4.0), (5, 6.0);

cmd/trcli/check/tests/pg2ch_test.go

@@ -0,0 +1,34 @@
+package tests
+
+import (
+	_ "embed"
+	"testing"
+
+	"github.com/doublecloud/tross/cmd/trcli/check"
+	"github.com/doublecloud/tross/cmd/trcli/config"
+	"github.com/doublecloud/tross/pkg/providers/postgres/pgrecipe"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed transfer.yaml
+var transferYaml []byte
+
+func TestCheck(t *testing.T) {
+	src := pgrecipe.RecipeSource(
+		pgrecipe.WithPrefix("SRC"),
+		pgrecipe.WithFiles("dump/pg_init.sql"),
+	)
+
+	dst := pgrecipe.RecipeTarget(
+		pgrecipe.WithPrefix("DST"),
+	)
+	dst.MaintainTables = true // forces table creation on push
+
+	transfer, err := config.ParseTransfer(transferYaml)
+	require.NoError(t, err)
+
+	transfer.Src = src
+	transfer.Dst = dst
+
+	require.NoError(t, check.RunCheck(transfer))
+}

cmd/trcli/check/tests/transfer.yaml

@@ -0,0 +1,18 @@
+id: dtttest
+transfername: pg-to-pg-test
+description: ""
+type: SNAPSHOT_ONLY
+src:
+  id: dtetestsource
+  name: pg-source
+  type: pg
+  params: |
+    {
+    }
+dst:
+  id: dtetesttarget
+  name: pg-target
+  type: pg
+  params: |
+    {
+    }