Use Praefect's RemoveRepository from remove-repository subcommand

The RemoveRepository subcommand is working around the lack of atomicity
in Praefect's repository removals. It manually cleaned up database state
and issued deletes Gitalys to ensure deletes go through even if there
is no metadata for the repository. This ensured there would be no conflicts
when recreating a repository after running the command.

These workarounds are no longer necessary as Praefect is now handling
deletions. As soon as the repository's metadata is deleted, it stops
existing from Praefects point of view. New repository created in the same
virtual storage and relative path lands in a different directory on the
Gitalys. As such, we can simplify the command and simply call Praefect's
RemoveRepository to delete the repository's metadata and to delete the
known replicas.

The command was changed to error if the repository does not exist as it
can't do anything if so. Likewise, the output was changed to not mention
the nodes the repository exists on as it's mostly irrelevant. The deletion
may succeed as far as the user cares but some replicas may still be left
on the disk if Praefect was unable to remove them.

Changelog: changed

3 files changed, 38 insertions, 187 deletions

diff --git a/cmd/praefect/subcmd_remove_repository.go b/cmd/praefect/subcmd_remove_repository.go
index 49b049615..5298c713d 100644
--- a/cmd/praefect/subcmd_remove_repository.go
+++ b/cmd/praefect/subcmd_remove_repository.go
@@ -7,7 +7,6 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"strings"
 	"sync"
 	"time"
 
@@ -19,7 +18,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/labkit/correlation"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/status"
 )
 
 const (
@@ -61,15 +59,11 @@ func (cmd *removeRepository) FlagSet() *flag.FlagSet {
 			"	This command removes all state associated with a given repository from the Gitaly Cluster.\n" +
 			"	This includes both on-disk repositories on all relevant Gitaly nodes as well as any potential\n" +
 			"	database state as tracked by Praefect.\n" +
-			"	Runs in dry-run mode by default and lists the gitaly nodes from which the repository will be removed from " +
+			"	Runs in dry-run mode by default checks whether the repository exists" +
 			"	without actually removing it from the database and disk.\n" +
 			"	When -apply is used, the repository will be removed from the database as well as\n" +
 			"	the individual gitaly nodes on which they exist.\n")
 		fs.PrintDefaults()
-		printfErr("NOTE:\n" +
-			"	It may happen that parts of the repository continue to exist after this command, either because\n" +
-			"	of an error that happened during deletion or because of in-flight RPC calls targeting the repository.\n" +
-			"	It is safe and recommended to re-run this command in such a case.\n")
 	}
 	return fs
 }
@@ -116,21 +110,10 @@ func (cmd *removeRepository) exec(ctx context.Context, logger logrus.FieldLogger
 		if exists {
 			fmt.Fprintln(cmd.w, "Repository found in the database.")
 		} else {
-			fmt.Fprintln(cmd.w, "Repository is not being tracked in Praefect.")
+			return errors.New("repository is not being tracked in Praefect")
 		}
 	}
 
-	storagesOnDisk, err := cmd.getStoragesFromNodes(ctx, cfg)
-	if err != nil {
-		return err
-	}
-
-	if len(storagesOnDisk) == 0 {
-		fmt.Fprintln(cmd.w, "Repository not found on any gitaly nodes.")
-	} else {
-		fmt.Fprintf(cmd.w, "Repository found on the following gitaly nodes: %s.\n", strings.Join(storagesOnDisk, ", "))
-	}
-
 	if !cmd.apply {
 		fmt.Fprintln(cmd.w, "Re-run the command with -apply to remove repositories from the database and disk.")
 		return nil
@@ -138,18 +121,21 @@ func (cmd *removeRepository) exec(ctx context.Context, logger logrus.FieldLogger
 
 	fmt.Fprintf(cmd.w, "Attempting to remove %s from the database, and delete it from all gitaly nodes...\n", cmd.relativePath)
 
-	fmt.Fprintln(cmd.w, "Removing repository from the database...")
-	removed, err := cmd.removeRepositoryFromDatabase(ctx, db)
+	addr, err := getNodeAddress(cfg)
 	if err != nil {
-		return fmt.Errorf("remove repository info from praefect database: %w", err)
+		return fmt.Errorf("get node address: %w", err)
 	}
 
-	if !removed {
-		fmt.Fprintln(cmd.w, "The database has no information about this repository.")
-	} else {
-		fmt.Fprintln(cmd.w, "Removed repository metadata from the database.")
+	_, err = cmd.removeRepository(ctx, &gitalypb.Repository{
+		StorageName:  cmd.virtualStorage,
+		RelativePath: cmd.relativePath,
+	}, addr, cfg.Auth.Token)
+	if err != nil {
+		return fmt.Errorf("repository removal failed: %w", err)
 	}
 
+	fmt.Fprintln(cmd.w, "Repository removal completed.")
+
 	fmt.Fprintln(cmd.w, "Removing replication events...")
 	ticker := helper.NewTimerTicker(time.Second)
 	defer ticker.Stop()
@@ -157,80 +143,9 @@ func (cmd *removeRepository) exec(ctx context.Context, logger logrus.FieldLogger
 		return fmt.Errorf("remove scheduled replication events: %w", err)
 	}
 	fmt.Fprintln(cmd.w, "Replication event removal completed.")
-
-	// We should try to remove repository from each of gitaly nodes.
-	fmt.Fprintln(cmd.w, "Removing repository directly from gitaly nodes...")
-	cmd.removeRepositoryForEachGitaly(ctx, cfg, logger)
-	fmt.Fprintln(cmd.w, "Finished removing repository from gitaly nodes.")
-
 	return nil
 }
 
-// getStoragesFromNodes looks on disk to finds the storages this repository exists for
-func (cmd *removeRepository) getStoragesFromNodes(ctx context.Context, cfg config.Config) ([]string, error) {
-	var storages []string
-	for _, vs := range cfg.VirtualStorages {
-		if vs.Name != cmd.virtualStorage {
-			continue
-		}
-
-		storages = make([]string, len(vs.Nodes))
-		var wg sync.WaitGroup
-
-		for i := 0; i < len(vs.Nodes); i++ {
-			wg.Add(1)
-			go func(node *config.Node, i int) {
-				defer wg.Done()
-				repo := &gitalypb.Repository{
-					StorageName:  node.Storage,
-					RelativePath: cmd.relativePath,
-				}
-				exists, err := repositoryExists(ctx, repo, node.Address, node.Token)
-				if err != nil {
-					cmd.logger.WithError(err).Errorf("checking if repository exists on %q", node.Storage)
-				}
-				if exists {
-					storages[i] = node.Storage
-				}
-			}(vs.Nodes[i], i)
-		}
-
-		wg.Wait()
-		break
-	}
-
-	var storagesFound []string
-	for _, storage := range storages {
-		if storage != "" {
-			storagesFound = append(storagesFound, storage)
-		}
-	}
-
-	return storagesFound, nil
-}
-
-func (cmd *removeRepository) removeRepositoryFromDatabase(ctx context.Context, db *sql.DB) (bool, error) {
-	var removed bool
-	if err := db.QueryRowContext(
-		ctx,
-		`WITH remove_storages_info AS (
-				DELETE FROM storage_repositories
-				WHERE virtual_storage = $1 AND relative_path = $2
-			)
-			DELETE FROM repositories
-			WHERE virtual_storage = $1 AND relative_path = $2
-			RETURNING TRUE`,
-		cmd.virtualStorage,
-		cmd.relativePath,
-	).Scan(&removed); err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
-			return false, nil
-		}
-		return false, fmt.Errorf("query row: %w", err)
-	}
-	return removed, nil
-}
-
 func (cmd *removeRepository) removeRepository(ctx context.Context, repo *gitalypb.Repository, addr, token string) (bool, error) {
 	conn, err := subCmdDial(ctx, addr, token, cmd.dialTimeout)
 	if err != nil {
@@ -241,10 +156,7 @@ func (cmd *removeRepository) removeRepository(ctx context.Context, repo *gitalyp
 	ctx = metadata.AppendToOutgoingContext(ctx, "client_name", removeRepositoryCmdName)
 	repositoryClient := gitalypb.NewRepositoryServiceClient(conn)
 	if _, err := repositoryClient.RemoveRepository(ctx, &gitalypb.RemoveRepositoryRequest{Repository: repo}); err != nil {
-		if _, ok := status.FromError(err); !ok {
-			return false, fmt.Errorf("RemoveRepository: %w", err)
-		}
-		return false, nil
+		return false, err
 	}
 	return true, nil
 }
@@ -291,43 +203,3 @@ func (cmd *removeRepository) removeReplicationEvents(ctx context.Context, logger
 	}
 	return nil
 }
-
-func (cmd *removeRepository) removeNode(
-	ctx context.Context,
-	logger logrus.FieldLogger,
-	node *config.Node,
-) {
-	logger.Debugf("remove repository with gitaly %q at %q", node.Storage, node.Address)
-	repo := &gitalypb.Repository{
-		StorageName:  node.Storage,
-		RelativePath: cmd.relativePath,
-	}
-	removed, err := cmd.removeRepository(ctx, repo, node.Address, node.Token)
-	if err != nil {
-		logger.WithError(err).Warnf("repository removal failed for %q", node.Storage)
-	} else {
-		if removed {
-			fmt.Fprintf(cmd.w, "Successfully removed %s from %s\n", cmd.relativePath, node.Storage)
-		} else {
-			fmt.Fprintf(cmd.w, "Did not remove %s from %s\n", cmd.relativePath, node.Storage)
-		}
-	}
-	logger.Debugf("repository removal call to gitaly %q completed", node.Storage)
-}
-
-func (cmd *removeRepository) removeRepositoryForEachGitaly(ctx context.Context, cfg config.Config, logger logrus.FieldLogger) {
-	for _, vs := range cfg.VirtualStorages {
-		if vs.Name == cmd.virtualStorage {
-			var wg sync.WaitGroup
-			for i := 0; i < len(vs.Nodes); i++ {
-				wg.Add(1)
-				go func(i int) {
-					defer wg.Done()
-					cmd.removeNode(ctx, logger, vs.Nodes[i])
-				}(i)
-			}
-			wg.Wait()
-			break
-		}
-	}
-}
diff --git a/cmd/praefect/subcmd_remove_repository_test.go b/cmd/praefect/subcmd_remove_repository_test.go
index 040dd40e5..f00be8d1d 100644
--- a/cmd/praefect/subcmd_remove_repository_test.go
+++ b/cmd/praefect/subcmd_remove_repository_test.go
@@ -6,15 +6,15 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/sirupsen/logrus"
-	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/v14/client"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/git/gittest"
+	gitalycfg "gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/config"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/service/setup"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/praefect/config"
@@ -107,6 +107,8 @@ func TestRemoveRepository_Exec(t *testing.T) {
 	t.Run("dry run", func(t *testing.T) {
 		var out bytes.Buffer
 		repo := createRepo(t, ctx, repoClient, praefectStorage, t.Name())
+		replicaPath := gittest.GetReplicaPath(ctx, t, gitalycfg.Cfg{SocketPath: praefectServer.Address()}, repo)
+
 		cmd := &removeRepository{
 			logger:         testhelper.NewDiscardingLogger(t),
 			virtualStorage: repo.StorageName,
@@ -117,10 +119,9 @@ func TestRemoveRepository_Exec(t *testing.T) {
 		}
 		require.NoError(t, cmd.Exec(flag.NewFlagSet("", flag.PanicOnError), conf))
 
-		require.DirExists(t, filepath.Join(g1Cfg.Storages[0].Path, repo.RelativePath))
-		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, repo.RelativePath))
+		require.DirExists(t, filepath.Join(g1Cfg.Storages[0].Path, replicaPath))
+		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, replicaPath))
 		assert.Contains(t, out.String(), "Repository found in the database.\n")
-		assert.Contains(t, out.String(), "Repository found on the following gitaly nodes:")
 		assert.Contains(t, out.String(), "Re-run the command with -apply to remove repositories from the database and disk.\n")
 
 		repositoryRowExists, err := datastore.NewPostgresRepositoryStore(db, nil).RepositoryExists(ctx, cmd.virtualStorage, cmd.relativePath)
@@ -144,10 +145,8 @@ func TestRemoveRepository_Exec(t *testing.T) {
 		require.NoDirExists(t, filepath.Join(g1Cfg.Storages[0].Path, repo.RelativePath))
 		require.NoDirExists(t, filepath.Join(g2Cfg.Storages[0].Path, repo.RelativePath))
 		assert.Contains(t, out.String(), "Repository found in the database.\n")
-		assert.Contains(t, out.String(), "Repository found on the following gitaly nodes:")
 		assert.Contains(t, out.String(), fmt.Sprintf("Attempting to remove %s from the database, and delete it from all gitaly nodes...\n", repo.RelativePath))
-		assert.Contains(t, out.String(), fmt.Sprintf("Successfully removed %s from", repo.RelativePath))
-		assert.NotContains(t, out.String(), fmt.Sprintf("Did not remove %s from", repo.RelativePath))
+		assert.Contains(t, out.String(), "Repository removal completed.")
 
 		var repositoryRowExists bool
 		require.NoError(t, db.QueryRow(
@@ -176,9 +175,8 @@ func TestRemoveRepository_Exec(t *testing.T) {
 		require.NoDirExists(t, filepath.Join(g1Cfg.Storages[0].Path, repo.RelativePath))
 		require.NoDirExists(t, filepath.Join(g2Cfg.Storages[0].Path, repo.RelativePath))
 		assert.Contains(t, out.String(), "Repository found in the database.\n")
-		assert.Contains(t, out.String(), "Repository found on the following gitaly nodes: gitaly-1")
 		assert.Contains(t, out.String(), fmt.Sprintf("Attempting to remove %s from the database, and delete it from all gitaly nodes...\n", repo.RelativePath))
-		assert.Contains(t, out.String(), fmt.Sprintf("Successfully removed %s from gitaly-1", repo.RelativePath))
+		assert.Contains(t, out.String(), "Repository removal completed.")
 
 		var repositoryRowExists bool
 		require.NoError(t, db.QueryRow(
@@ -191,8 +189,10 @@ func TestRemoveRepository_Exec(t *testing.T) {
 	t.Run("no info about repository on praefect", func(t *testing.T) {
 		var out bytes.Buffer
 		repo := createRepo(t, ctx, repoClient, praefectStorage, t.Name())
+		replicaPath := gittest.GetReplicaPath(ctx, t, gitalycfg.Cfg{SocketPath: praefectServer.Address()}, repo)
+
 		repoStore := datastore.NewPostgresRepositoryStore(db.DB, nil)
-		_, _, err := repoStore.DeleteRepository(ctx, repo.StorageName, repo.RelativePath)
+		_, _, err = repoStore.DeleteRepository(ctx, repo.StorageName, repo.RelativePath)
 		require.NoError(t, err)
 
 		cmd := &removeRepository{
@@ -203,12 +203,9 @@ func TestRemoveRepository_Exec(t *testing.T) {
 			apply:          true,
 			w:              &writer{w: &out},
 		}
-		require.NoError(t, cmd.Exec(flag.NewFlagSet("", flag.PanicOnError), conf))
-		assert.Contains(t, out.String(), "Repository is not being tracked in Praefect.\n")
-		assert.Contains(t, out.String(), "Repository found on the following gitaly nodes:")
-		assert.Contains(t, out.String(), "The database has no information about this repository.\n")
-		require.NoDirExists(t, filepath.Join(g1Cfg.Storages[0].Path, repo.RelativePath))
-		require.NoDirExists(t, filepath.Join(g2Cfg.Storages[0].Path, repo.RelativePath))
+		require.Error(t, cmd.Exec(flag.NewFlagSet("", flag.PanicOnError), conf), "repository is not being tracked in Praefect")
+		require.DirExists(t, filepath.Join(g1Cfg.Storages[0].Path, replicaPath))
+		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, replicaPath))
 
 		requireNoDatabaseInfo(t, db, cmd)
 	})
@@ -218,11 +215,12 @@ func TestRemoveRepository_Exec(t *testing.T) {
 		repo := createRepo(t, ctx, repoClient, praefectStorage, t.Name())
 		g2Srv.Shutdown()
 
-		logger := testhelper.NewDiscardingLogger(t)
-		loggerHook := test.NewLocal(logger)
+		replicaPath := gittest.GetReplicaPath(ctx, t, gitalycfg.Cfg{SocketPath: praefectServer.Address()}, repo)
+		require.DirExists(t, filepath.Join(g1Cfg.Storages[0].Path, replicaPath))
+		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, replicaPath))
 
 		cmd := &removeRepository{
-			logger:         logrus.NewEntry(logger),
+			logger:         logrus.NewEntry(testhelper.NewDiscardingLogger(t)),
 			virtualStorage: praefectStorage,
 			relativePath:   repo.RelativePath,
 			dialTimeout:    100 * time.Millisecond,
@@ -231,22 +229,10 @@ func TestRemoveRepository_Exec(t *testing.T) {
 		}
 
 		require.NoError(t, cmd.Exec(flag.NewFlagSet("", flag.PanicOnError), conf))
+		assert.Contains(t, out.String(), "Repository removal completed.")
 
-		var checkExistsOnNodeErrFound, removeRepoFromDiskErrFound bool
-		for _, entry := range loggerHook.AllEntries() {
-			if strings.Contains(entry.Message, `checking if repository exists on "gitaly-2"`) {
-				checkExistsOnNodeErrFound = true
-			}
-
-			if strings.Contains(entry.Message, `repository removal failed for "gitaly-2"`) {
-				removeRepoFromDiskErrFound = true
-			}
-		}
-		require.True(t, checkExistsOnNodeErrFound)
-		require.True(t, removeRepoFromDiskErrFound)
-
-		require.NoDirExists(t, filepath.Join(g1Cfg.Storages[0].Path, repo.RelativePath))
-		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, repo.RelativePath))
+		require.NoDirExists(t, filepath.Join(g1Cfg.Storages[0].Path, replicaPath))
+		require.DirExists(t, filepath.Join(g2Cfg.Storages[0].Path, replicaPath))
 
 		requireNoDatabaseInfo(t, db, cmd)
 	})
diff --git a/cmd/praefect/subcmd_track_repository_test.go b/cmd/praefect/subcmd_track_repository_test.go
index 2e3c807bf..3e8750f42 100644
--- a/cmd/praefect/subcmd_track_repository_test.go
+++ b/cmd/praefect/subcmd_track_repository_test.go
@@ -3,7 +3,6 @@ package main
 import (
 	"bytes"
 	"flag"
-	"io"
 	"path/filepath"
 	"testing"
 	"time"
@@ -170,15 +169,9 @@ func TestAddRepository_Exec(t *testing.T) {
 				defer nodeMgr.Stop()
 
 				repoDS := datastore.NewPostgresRepositoryStore(db, conf.StorageNames())
-
-				rmRepoCmd := &removeRepository{
-					logger:         logger,
-					virtualStorage: virtualStorageName,
-					relativePath:   tc.relativePath,
-					w:              io.Discard,
-					apply:          true,
-				}
-				require.NoError(t, rmRepoCmd.Exec(flag.NewFlagSet("", flag.PanicOnError), conf))
+				exists, err := repoDS.RepositoryExists(ctx, virtualStorageName, tc.relativePath)
+				require.NoError(t, err)
+				require.False(t, exists)
 
 				// create the repo on Gitaly without Praefect knowing
 				require.NoError(t, createRepoThroughGitaly1(tc.relativePath))
@@ -207,7 +200,7 @@ func TestAddRepository_Exec(t *testing.T) {
 				assert.Contains(t, assignments, g1Cfg.Storages[0].Name)
 				assert.Contains(t, assignments, g2Cfg.Storages[0].Name)
 
-				exists, err := repoDS.RepositoryExists(ctx, virtualStorageName, tc.relativePath)
+				exists, err = repoDS.RepositoryExists(ctx, virtualStorageName, tc.relativePath)
 				require.NoError(t, err)
 				assert.True(t, exists)
 				assert.Contains(t, stdout.String(), tc.expectedOutput)