feat: migration to single table SQL schema (#707)

This change adds a migration path from Keto version v0.6.x to the new persistence structure introduced by #638. Every namespace has to be migrated separately, or you can use the CLI to detect and migrate all namespaces at once. Have a look at `keto help namespace migrate legacy` for all details.
**Please make sure that you backup the database before running the migration command**. Please note that this migration might be a bit slower than usual, as we have to pull the data from the database, transcode it in Keto, and then write it to the new table structure.
Versions of Keto >v0.7 will not include this migration script, so you will first have to migrate to v0.7 and move on from there.

Author: zepatrik

.github/workflows/single-table-migration-e2e.yml

@@ -0,0 +1,24 @@
+name: Run full e2e test of the migration to single table persister (see https://github.com/ory/keto/issues/628)
+
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - feat/persistence-migration-path
+
+jobs:
+  test-migration:
+    runs-on: ubuntu-latest
+    name: Test Migration
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.16'
+      - name: Run test script
+        run: ./scripts/single-table-migration-e2e.sh
+      - uses: actions/upload-artifact@v2
+        if: failure()
+        with:
+          name: sqlite-db
+          path: migrate_e2e.sqlite

cmd/migrate/down.go

@@ -19,8 +19,7 @@ func newDownCmd() *cobra.Command {
 		Use:   "down <steps>",
 		Short: "Migrate the database down",
 		Long: "Migrate the database down a specific amount of steps.\n" +
-			"Pass 0 steps to fully migrate down.\n" +
-			"This does not affect namespaces. Use `keto namespace migrate down` for migrating namespaces.",
+			"Pass 0 steps to fully migrate down.",
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			steps, err := strconv.ParseInt(args[0], 0, 0)
@@ -39,7 +38,7 @@ func newDownCmd() *cobra.Command {
 				return err
 			}
 
-			return BoxDown(cmd, mb, int(steps), "")
+			return BoxDown(cmd, mb, int(steps))
 		},
 	}
 
@@ -49,27 +48,27 @@ func newDownCmd() *cobra.Command {
 	return cmd
 }
 
-func BoxDown(cmd *cobra.Command, mb *popx.MigrationBox, steps int, msgPrefix string) error {
+func BoxDown(cmd *cobra.Command, mb *popx.MigrationBox, steps int) error {
 	s, err := mb.Status(cmd.Context())
 	if err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould not get migration status: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not get migration status: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 	cmdx.PrintTable(cmd, s)
 
-	if !flagx.MustGetBool(cmd, FlagYes) && !cmdx.AskForConfirmation(msgPrefix+"Do you really want to migrate down? This will delete data.", cmd.InOrStdin(), cmd.OutOrStdout()) {
-		_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"Migration aborted.")
+	if !flagx.MustGetBool(cmd, FlagYes) && !cmdx.AskForConfirmation("Do you really want to migrate down? This will delete data.", cmd.InOrStdin(), cmd.OutOrStdout()) {
+		_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Migration aborted.")
 		return nil
 	}
 
 	if err := mb.Down(cmd.Context(), steps); err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould apply down migrations: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could apply down migrations: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 
 	s, err = mb.Status(cmd.Context())
 	if err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould not get migration status: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not get migration status: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 	cmdx.PrintTable(cmd, s)

cmd/migrate/up.go

@@ -16,18 +16,22 @@ import (
 )
 
 const (
-	FlagYes          = "yes"
-	FlagAllNamespace = "all-namespaces"
+	FlagYes = "yes"
 )
 
 func newUpCmd() *cobra.Command {
-	var allNamespaces bool
-
 	cmd := &cobra.Command{
 		Use:   "up",
 		Short: "Migrate the database up",
-		Long: "Migrate the database up.\n" +
-			"This does not affect namespaces. Use `keto namespace migrate up` for migrating namespaces.",
+		Long: `Run this command on a fresh SQL installation and when you upgrade Ory Keto from version v0.7.0 and later.
+
+It is recommended to run this command close to the SQL instance (e.g. same subnet) instead of over the public internet.
+This decreases risk of failure and decreases time required.
+
+### WARNING ###
+
+Before running this command on an existing database, create a back up!
+`,
 		Args: cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, _ []string) error {
 			ctx := cmd.Context()
@@ -42,7 +46,7 @@ func newUpCmd() *cobra.Command {
 				return err
 			}
 
-			if err := BoxUp(cmd, mb, ""); err != nil {
+			if err := BoxUp(cmd, mb); err != nil {
 				return err
 			}
 
@@ -51,7 +55,6 @@ func newUpCmd() *cobra.Command {
 	}
 
 	RegisterYesFlag(cmd.Flags())
-	cmd.Flags().BoolVar(&allNamespaces, FlagAllNamespace, false, "migrate all pending namespaces as well")
 
 	cmdx.RegisterFormatFlags(cmd.Flags())
 
@@ -62,38 +65,38 @@ func RegisterYesFlag(flags *pflag.FlagSet) {
 	flags.BoolP(FlagYes, "y", false, "yes to all questions, no user input required")
 }
 
-func BoxUp(cmd *cobra.Command, mb *popx.MigrationBox, msgPrefix string) error {
-	_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"Current status:")
+func BoxUp(cmd *cobra.Command, mb *popx.MigrationBox) error {
+	_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Current status:")
 
 	s, err := mb.Status(cmd.Context())
 	if err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould not get migration status: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not get migration status: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 	cmdx.PrintTable(cmd, s)
 
 	if !s.HasPending() {
-		_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"All migrations are already applied, there is nothing to do.")
+		_, _ = fmt.Fprintln(cmd.OutOrStdout(), "All migrations are already applied, there is nothing to do.")
 		return nil
 	}
 
-	if !flagx.MustGetBool(cmd, FlagYes) && !cmdx.AskForConfirmation(msgPrefix+"Are you sure that you want to apply this migration? Make sure to check the CHANGELOG.md for breaking changes beforehand.", cmd.InOrStdin(), cmd.OutOrStdout()) {
-		_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"Aborting")
+	if !flagx.MustGetBool(cmd, FlagYes) && !cmdx.AskForConfirmation("Are you sure that you want to apply this migration? Make sure to check the CHANGELOG.md for breaking changes beforehand.", cmd.InOrStdin(), cmd.OutOrStdout()) {
+		_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Aborting")
 		return nil
 	}
 
-	_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"Applying migrations...")
+	_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Applying migrations...")
 
 	if err := mb.Up(cmd.Context()); err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould not apply migrations: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not apply migrations: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 
-	_, _ = fmt.Fprintln(cmd.OutOrStdout(), msgPrefix+"Successfully applied all migrations:")
+	_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Successfully applied all migrations:")
 
 	s, err = mb.Status(cmd.Context())
 	if err != nil {
-		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%sCould not get migration status: %+v\n", msgPrefix, err)
+		_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not get migration status: %+v\n", err)
 		return cmdx.FailSilently(cmd)
 	}
 

cmd/namespace/migrate_legacy.go

@@ -0,0 +1,118 @@
+package namespace
+
+import (
+	"fmt"
+
+	"github.com/ory/x/cmdx"
+	"github.com/ory/x/flagx"
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/ory/keto/cmd/migrate"
+	"github.com/ory/keto/internal/driver"
+	"github.com/ory/keto/internal/namespace"
+	"github.com/ory/keto/internal/persistence"
+	"github.com/ory/keto/internal/persistence/sql/migrations"
+)
+
+func NewMigrateLegacyCmd() *cobra.Command {
+	downOnly := false
+
+	cmd := &cobra.Command{
+		Use:   "legacy [<namespace-name>]",
+		Short: "Migrate a namespace from v0.6.x to v0.7.x and later.",
+		Long: "Migrate a legacy namespaces from v0.6.x to the v0.7.x and later.\n" +
+			"This step only has to be executed once.\n" +
+			"If no namespace is specified, all legacy namespaces will be migrated.\n" +
+			"Please ensure that namespace IDs did not change in the config file and you have a backup in case something goes wrong!",
+		Args: cobra.MaximumNArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			reg, err := driver.NewDefaultRegistry(cmd.Context(), cmd.Flags(), false)
+			if errors.Is(err, persistence.ErrNetworkMigrationsMissing) {
+				_, _ = fmt.Fprintln(cmd.ErrOrStderr(), "Migrations were not applied yet, please apply them first using `keto migrate up`.")
+				return cmdx.FailSilently(cmd)
+			} else if err != nil {
+				return err
+			}
+
+			migrator := migrations.NewToSingleTableMigrator(reg)
+
+			var nn []*namespace.Namespace
+			if len(args) == 1 {
+				nm, err := reg.Config().NamespaceManager()
+				if err != nil {
+					_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "There seems to be a problem with the config: %s\n", err.Error())
+					return cmdx.FailSilently(cmd)
+				}
+				n, err := nm.GetNamespaceByName(cmd.Context(), args[0])
+				if err != nil {
+					_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "There seems to be a problem with the config: %s\n", err.Error())
+					return cmdx.FailSilently(cmd)
+				}
+
+				nn = []*namespace.Namespace{n}
+
+				if !flagx.MustGetBool(cmd, migrate.FlagYes) &&
+					!cmdx.AskForConfirmation(
+						fmt.Sprintf("Are you sure that you want to migrate the namespace '%s'?", args[0]),
+						cmd.InOrStdin(), cmd.OutOrStdout()) {
+					_, _ = fmt.Fprintln(cmd.OutOrStdout(), "OK, aborting.")
+					return nil
+				}
+			} else {
+				nn, err = migrator.LegacyNamespaces(cmd.Context())
+				if err != nil {
+					_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not get legacy namespaces: %s\n", err.Error())
+					return cmdx.FailSilently(cmd)
+				}
+
+				if len(nn) == 0 {
+					_, _ = fmt.Fprintln(cmd.OutOrStdout(), "Could not find legacy namespaces, there seems nothing to be done.")
+					return nil
+				}
+
+				var names string
+				for _, n := range nn {
+					names += "  " + n.Name + "\n"
+				}
+				if !flagx.MustGetBool(cmd, migrate.FlagYes) &&
+					!cmdx.AskForConfirmation(
+						fmt.Sprintf("I found the following legacy namespaces:\n%sDo you want to migrate all of them?", names),
+						cmd.InOrStdin(), cmd.OutOrStdout()) {
+					_, _ = fmt.Fprintln(cmd.OutOrStdout(), "OK, aborting.")
+					return nil
+				}
+			}
+
+			for _, n := range nn {
+				if !downOnly {
+					if err := migrator.MigrateNamespace(cmd.Context(), n); err != nil {
+						_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Encountered error while migrating: %s\nAborting.\n", err.Error())
+						if errors.Is(err, migrations.ErrInvalidTuples(nil)) {
+							_, _ = fmt.Fprintln(cmd.ErrOrStderr(), "Please see https://github.com/ory/keto/issues/661 for why this happens and how to resolve this.")
+						}
+						return cmdx.FailSilently(cmd)
+					}
+				}
+				if flagx.MustGetBool(cmd, migrate.FlagYes) ||
+					cmdx.AskForConfirmation(
+						fmt.Sprintf("Do you want to migrate namespace %s down? This will delete all data in the legacy table.", n.Name),
+						cmd.InOrStdin(), cmd.OutOrStdout()) {
+					if err := migrator.MigrateDown(cmd.Context(), n); err != nil {
+						_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Could not migrate down: %s\n", err.Error())
+						return cmdx.FailSilently(cmd)
+					}
+					_, _ = fmt.Fprintf(cmd.OutOrStdout(), "Successfully migrated down namespace %s.\n", n.Name)
+				}
+			}
+
+			return nil
+		},
+	}
+
+	migrate.RegisterYesFlag(cmd.Flags())
+	registerPackageFlags(cmd.Flags())
+	cmd.Flags().BoolVar(&downOnly, "down-only", false, "Migrate legacy namespace(s) only down.")
+
+	return cmd
+}

cmd/namespace/migrate_legacy_snapshot.sqlite

@@ -0,0 +1,90 @@
+package namespace
+
+import (
+	"context"
+	"io"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/ory/keto/internal/relationtuple"
+
+	"github.com/ory/x/cmdx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ory/keto/internal/driver"
+	"github.com/ory/keto/internal/driver/config"
+	"github.com/ory/keto/internal/namespace"
+	"github.com/ory/keto/internal/x/dbx"
+)
+
+func TestMigrateLegacy(t *testing.T) {
+	setup := func(t *testing.T) (*cmdx.CommandExecuter, *driver.RegistryDefault) {
+		fp := path.Join(t.TempDir(), "db.sqlite")
+		dst, err := os.Create(fp)
+		require.NoError(t, err)
+		defer dst.Close()
+		src, err := os.Open("migrate_legacy_snapshot.sqlite")
+		require.NoError(t, err)
+		defer src.Close()
+		_, err = io.Copy(dst, src)
+		require.NoError(t, err)
+
+		reg := driver.NewTestRegistry(t, &dbx.DsnT{
+			Name:        "sqlite",
+			Conn:        "sqlite://" + fp + "?_fk=true",
+			MigrateUp:   true,
+			MigrateDown: false,
+		})
+		nspaces := []*namespace.Namespace{
+			{
+				ID:   0,
+				Name: "a",
+			},
+			{
+				ID:   1,
+				Name: "b",
+			},
+		}
+		require.NoError(t, reg.Config().Set(config.KeyNamespaces, nspaces))
+
+		c := &cmdx.CommandExecuter{
+			New: NewMigrateLegacyCmd,
+			Ctx: context.WithValue(context.Background(), driver.RegistryContextKey, reg),
+		}
+
+		return c, reg
+	}
+
+	t.Run("case=invalid subject", func(t *testing.T) {
+		c, reg := setup(t)
+
+		conn, err := reg.PopConnection()
+		require.NoError(t, err)
+		require.NoError(t, conn.RawQuery("insert into keto_0000000000_relation_tuples (shard_id, object, relation, subject, commit_time) values ('foo', 'obj', 'rel', 'invalid#subject', 'now')").Exec())
+
+		stdErr := c.ExecExpectedErr(t, "-y", "a")
+		assert.Contains(t, stdErr, "found non-deserializable relationtuples")
+		assert.Contains(t, stdErr, "invalid#subject")
+
+		assert.Contains(t, c.ExecNoErr(t, "-y", "--down-only", "a"), "Successfully migrated down")
+	})
+
+	t.Run("case=migrates down only", func(t *testing.T) {
+		c, reg := setup(t)
+
+		conn, err := reg.PopConnection()
+		require.NoError(t, err)
+		require.NoError(t, conn.RawQuery("insert into keto_0000000000_relation_tuples (shard_id, object, relation, subject, commit_time) values ('foo', 'obj', 'rel', 'sub', 'now')").Exec())
+
+		c.ExecNoErr(t, "-y", "--down-only", "a")
+
+		rts, _, err := reg.RelationTupleManager().GetRelationTuples(context.Background(), &relationtuple.RelationQuery{
+			Namespace: "a",
+			Object:    "obj",
+		})
+		require.NoError(t, err)
+		assert.Len(t, rts, 0)
+	})
+}

cmd/namespace/migrate_legacy_test.go

@@ -25,7 +25,7 @@ func NewMigrateCmd() *cobra.Command {
 func RegisterCommandsRecursive(parent *cobra.Command) {
 	rootCmd := NewNamespaceCmd()
 	migrateCmd := NewMigrateCmd()
-	migrateCmd.AddCommand(NewMigrateUpCmd(), NewMigrateDownCmd(), NewMigrateStatusCmd())
+	migrateCmd.AddCommand(NewMigrateUpCmd(), NewMigrateDownCmd(), NewMigrateStatusCmd(), NewMigrateLegacyCmd())
 
 	rootCmd.AddCommand(migrateCmd, NewValidateCmd())