[CORE-2238] Loki in testpachd fast-follow

src/internal/dockertestenv/BUILD.bazel

@@ -12,7 +12,6 @@ go_library(
     visibility = ["//src:__subpackages__"],
     deps = [
         "//src/internal/backoff",
-        "//src/internal/cleanup",
         "//src/internal/dbutil",
         "//src/internal/errors",
         "//src/internal/log",

src/internal/dockertestenv/postgres.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/pachyderm/pachyderm/v2/src/internal/backoff"
-	"github.com/pachyderm/pachyderm/v2/src/internal/cleanup"
 	"github.com/pachyderm/pachyderm/v2/src/internal/dbutil"
 	"github.com/pachyderm/pachyderm/v2/src/internal/errors"
 	"github.com/pachyderm/pachyderm/v2/src/internal/log"
@@ -124,16 +123,39 @@ func NewTestDBConfig(t testing.TB) DBConfig {
 }
 
 // NewTestDBConfigCtx returns a DBConfig for an environment outside of tests.
-func NewTestDBConfigCtx(ctx context.Context) (config DBConfig, cleaner *cleanup.Cleaner, _ error) {
+func NewTestDBConfigCtx(ctx context.Context) (DBConfig, error) {
 	var (
 		dbName  = testutil.GenerateEphemeralDBName()
 		dexName = testutil.UniqueString("dex")
+		config  = DBConfig{
+			Direct: PostgresConfig{
+				Host:     postgresHost(),
+				Port:     postgresPort,
+				User:     DefaultPostgresUser,
+				Password: DefaultPostgresPassword,
+				DBName:   dbName,
+			},
+			PGBouncer: PostgresConfig{
+				Host:     PGBouncerHost(),
+				Port:     PGBouncerPort,
+				User:     DefaultPostgresUser,
+				Password: DefaultPostgresPassword,
+				DBName:   dbName,
+			},
+			Identity: PostgresConfig{
+				Host:     PGBouncerHost(),
+				Port:     PGBouncerPort,
+				User:     DefaultPostgresUser,
+				Password: DefaultPostgresPassword,
+				DBName:   dexName,
+			},
+		}
 	)
-	cleaner = new(cleanup.Cleaner)
+
 	if err := backoff.Retry(func() error {
 		return EnsureDBEnv(ctx)
 	}, backoff.NewConstantBackOff(time.Second*3)); err != nil {
-		return config, cleaner, errors.Wrap(err, "wait for database to be created")
+		return config, errors.Wrap(err, "wait for database to be created")
 	}
 
 	db, err := dbutil.NewDB(
@@ -142,45 +164,37 @@ func NewTestDBConfigCtx(ctx context.Context) (config DBConfig, cleaner *cleanup.
 		dbutil.WithHostPort(PGBouncerHost(), PGBouncerPort),
 		dbutil.WithDBName(DefaultPostgresDatabase))
 	if err != nil {
-		return config, cleaner, errors.Wrap(err, "NewDB")
+		return config, errors.Wrap(err, "NewDB")
 	}
-	cleaner.AddCleanup("close db connection", db.Close)
+	defer db.Close()
 
 	if err := testutil.CreateEphemeralDBNontest(ctx, db, dbName); err != nil {
-		return config, cleaner, errors.Wrapf(err, "CreateEphemeralDB(%v)", dbName)
+		return config, errors.Wrapf(err, "CreateEphemeralDB(%v)", dbName)
 	}
-	cleaner.AddCleanupCtx("cleanup pach database", func(ctx context.Context) error {
-		return errors.Wrapf(testutil.CleanupEphemeralDB(ctx, db, dbName), "cleanup database %v", dbName)
-	})
 	if err := testutil.CreateEphemeralDBNontest(ctx, db, dexName); err != nil {
-		return config, cleaner, errors.Wrapf(err, "CreateEphemeralDB(%v)", dexName)
+		return config, errors.Wrapf(err, "CreateEphemeralDB(%v)", dexName)
 	}
-	cleaner.AddCleanupCtx("cleanup dex database", func(ctx context.Context) error {
-		return errors.Wrapf(testutil.CleanupEphemeralDB(ctx, db, dexName), "cleanup database %v", dbName)
-	})
-	return DBConfig{
-		Direct: PostgresConfig{
-			Host:     postgresHost(),
-			Port:     postgresPort,
-			User:     DefaultPostgresUser,
-			Password: DefaultPostgresPassword,
-			DBName:   dbName,
-		},
-		PGBouncer: PostgresConfig{
-			Host:     PGBouncerHost(),
-			Port:     PGBouncerPort,
-			User:     DefaultPostgresUser,
-			Password: DefaultPostgresPassword,
-			DBName:   dbName,
-		},
-		Identity: PostgresConfig{
-			Host:     PGBouncerHost(),
-			Port:     PGBouncerPort,
-			User:     DefaultPostgresUser,
-			Password: DefaultPostgresPassword,
-			DBName:   dexName,
-		},
-	}, cleaner, nil
+	return config, nil
+}
+
+func (config DBConfig) Cleanup(ctx context.Context) error {
+	db, err := dbutil.NewDB(
+		dbutil.WithMaxOpenConns(1),
+		dbutil.WithUserPassword(DefaultPostgresUser, DefaultPostgresPassword),
+		dbutil.WithHostPort(PGBouncerHost(), PGBouncerPort),
+		dbutil.WithDBName(DefaultPostgresDatabase))
+	for _, c := range []PostgresConfig{config.Direct, config.Identity} {
+		if err != nil {
+			log.Error(ctx, "opening", zap.String("db", c.DBName), zap.Error(err))
+			continue
+		}
+		if err := testutil.CleanupEphemeralDB(ctx, db, c.DBName); err != nil {
+			log.Error(ctx, "deleting", zap.String("db", c.DBName), zap.Error(err))
+			continue
+		}
+
+	}
+	return errors.Wrap(db.Close(), "closing database")
 }
 
 // NewTestDB creates a new database connection scoped to the test.

src/internal/lokiutil/testloki/BUILD.bazel

@@ -1,9 +1,13 @@
 load("@rules_go//go:def.bzl", "go_library", "go_test")
 
+# gazelle:go_test file
+
 go_library(
     name = "testloki",
-    testonly = 1,
-    srcs = ["testloki.go"],
+    srcs = [
+        "logger.go",
+        "testloki.go",
+    ],
     data = [
         "config.yaml",
         "//tools/loki",
@@ -14,10 +18,14 @@ go_library(
         "//src/internal/errors",
         "//src/internal/log",
         "//src/internal/lokiutil/client",
+        "//src/internal/pachconfig",
+        "//src/internal/pachd",
         "//src/internal/pctx",
         "//src/internal/promutil",
+        "//src/internal/randutil",
         "@in_gopkg_yaml_v3//:yaml_v3",
         "@org_uber_go_zap//:zap",
+        "@org_uber_go_zap//zapcore",
         "@rules_go//go/runfiles:go_default_library",
         "@rules_go//go/tools/bazel:go_default_library",
     ],
@@ -34,3 +42,16 @@ go_test(
         "@com_github_google_go_cmp//cmp",
     ],
 )
+
+go_test(
+    name = "logger_test",
+    size = "small",
+    srcs = ["logger_test.go"],
+    deps = [
+        ":testloki",
+        "//src/internal/pachd",
+        "//src/internal/pctx",
+        "//src/logs",
+        "@org_golang_google_protobuf//encoding/protojson",
+    ],
+)