Tilt local env: Expand the set of tests that can run locally by adding Toxyproxy service and separating PG connections in tests (#4156)

This PR:

1. Adds `Toxiproxy` as Tilt resource, optionally initialized as other
ancillary services.
2. Separates PG access in several tests that assumed PG DB for PeerDB
catalog and Source DB to be the same.
3. Changes error matching from `==` to `errors.Is()`
4. Cleans-up Toxiproxy configuration when the tests using it complete.
5. Sets `PEERDB_VERSION_SHA_SHORT` to non-empty strings in Tilt
environment.
6. Improves reliability of `TestResyncFailed` by:
  a. Cancelling the pipeline before tearing down PSQL.
b. Limiting backend PID kill to `peer` db application as we want to test
how the pipeline fails without affecting the client pool at test code
side.
c. Separating the catalog checks connection application from the one
being killed.
7. Adds `ssh` as Tilt resource, optionally initialized as other
ancillary services.

This make the following e2e pass locally:

- `env -f ../.envgo test ./e2e/... -v -test.run TestApiPg`
- `env -f ../.envgo test ./e2e/... -v -test.run TestApiMy`
- `env -f ../.envgo test ./e2e/... -v -test.run TestApiMongo`
- `env -f ../.envgo test ./e2e/... -v -test.run
^TestPeerFlowE2ETestSuitePG_CH$`
- `env -f ../.envgo test ./e2e/... -v -test.run
^TestPeerFlowE2ETestSuiteMySQL_CH$`
- `env -f ../.envgo test ./e2e/... -v -test.run TestGenericCH_PG`
- `env -f ../.envgo test ./e2e/... -v -test.run TestGenericCH_MySQL`
- `env -f ../.envgo test ./e2e/... -v -test.run
TestMongoClickhouseSuite`
- `env -f ../.envgo test ./e2e/... -v -test.run TestSwitchboardMongo`
- `env -f ../.envgo test ./e2e/... -v -test.run TestSwitchboardMySQL`
- `env -f ../.envgo test ./e2e/... -v -test.run TestSwitchboardPostgres`
- `env -f ../.envgo test ./connectors/postgres/... -v`
- `env -f ../.envgo test ./connectors/clickhouse/... -v` -> ⚠️
Passing all but:
- `env -f ../.envTestIAMRoleCanIssueSelectFromS3` This test uses AWS
access credentials, I don't think it matches minio features, skipping
when `FLOW_TESTS_AWS_S3_BUCKET_NAME` is not present.

📝 This PR doesn't address the env var injection with `godotenv` so
`enf -f` doesn't need to be used. It doesn't expand the "Run through
Tilt UI clicks" addition of all the supported cases, I am adding both in
a follow-up PR.

Closes: https://linear.app/clickhouse/issue/DBI-640

---------

Co-authored-by: Ilia Demianenko <ilia.demianenko@clickhouse.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

Tiltfile

@@ -6,7 +6,7 @@ docker_build('flow-api', '.',
     dockerfile='stacks/flow.Dockerfile',
     target='flow-api',
     only=['flow/', 'stacks/flow.Dockerfile'],
-    build_args={'PEERDB_VERSION_SHA_SHORT': os.getenv('PEERDB_VERSION_SHA_SHORT', '')},
+    build_args={'PEERDB_VERSION_SHA_SHORT': os.getenv('PEERDB_VERSION_SHA_SHORT', 'unknown')},
 )
 
 docker_build('flow-worker', '.',
@@ -149,6 +149,11 @@ dc_resource('dozzle', labels=['Monitoring'], links=[
     link('http://localhost:8118', 'Dozzle Container Monitor'),
 ])
 
+# Test services: Services supporting test execution that are not data stores, like proxies, mock servers, etc.
+
+dc_resource('toxiproxy', labels=['TestInfra'], auto_init=False)
+dc_resource('openssh', labels=['TestInfra'], auto_init=False)
+
 # Cleanup
 
 load('ext://uibutton', 'cmd_button', 'location')
@@ -178,6 +183,8 @@ def e2e_test(name, test_run, extra_deps=[], vars_overrides={}):
         resource_deps=['flow-api', 'flow-worker', 'catalog', 'provision-clickhouse'] + extra_deps,
     )
 
+# Generic e2e tests
+
 # Postgres to ClickHouse generic tests
 e2e_test('postgres', 'TestGenericCH_PG', ['provision-postgres'])
 
@@ -191,4 +198,8 @@ e2e_test('mysql-pos', 'TestGenericCH_MySQL', ['provision-mysql-pos'], vars_overr
 e2e_test('mariadb', 'TestGenericCH_MySQL', ['provision-mariadb'], vars_overrides={'CI_MYSQL_PORT': '$CI_MARIADB_PORT'})
 
 # MongoDB to ClickHouse test suite
-e2e_test('mongodb', 'TestMongoClickhouseSuite', ['provision-mongodb'])
+e2e_test('mongodb', 'TestMongoClickhouseSuite', ['provision-mongodb'])
+
+# API e2e tests
+
+e2e_test('api-postgres', 'TestApiPg', ['provision-postgres'])

ancillary-docker-compose.yml

@@ -129,6 +129,53 @@ services:
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
 
+  toxiproxy:
+    container_name: peerdb-test-toxiproxy
+    # Same image as CI flow
+    image: ghcr.io/shopify/toxiproxy:2.12.0@sha256:9378ed52a28bc50edc1350f936f518f31fa95f0d15917d6eb40b8e376d1a214e
+    restart: unless-stopped
+    ports:
+      - 18474:8474
+      - 9902:9902
+      - 9904:9904
+      - 9903:9903
+      - 42001:42001
+      - 42002:42002
+      - 42003:42003
+      - 49001:49001
+      - 49002:49002
+      - 49003:49003
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    healthcheck:
+      test: ["CMD", "/toxiproxy-cli", "list"]
+      interval: 2s
+      timeout: 10s
+      retries: 5
+
+  openssh:
+    container_name: peerdb-test-ssh
+    # Same image as CI flow
+    image: linuxserver/openssh-server:latest@sha256:57c13941ef385d735cca18ee969eeba18e42ab95480c97af2182505aacdfe307
+    restart: unless-stopped
+    ports:
+      - "2222:2222"
+    environment:
+      - PUID=1000
+      - PGID=1000
+      - TZ=Etc/UTC
+      - USER_NAME=testuser
+      - USER_PASSWORD=testpass
+      - PASSWORD_ACCESS=true
+      - DOCKER_MODS=linuxserver/mods:openssh-server-ssh-tunnel
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    healthcheck:
+      test: ["CMD", "nc", "-z", "localhost", "2222"]
+      interval: 2s
+      timeout: 10s
+      retries: 5
+
 volumes:
   postgres_data:
   clickhouse_data:
@@ -140,3 +187,4 @@ volumes:
 networks:
   default:
     name: peerdb_network
+    external: true

flow/connectors/clickhouse/s3_iam_role_test.go

@@ -3,6 +3,7 @@ package connclickhouse
 import (
 	"fmt"
 	"os"
+	"strconv"
 	"testing"
 	"time"
 
@@ -13,6 +14,11 @@ import (
 )
 
 func TestIAMRoleCanIssueSelectFromS3(t *testing.T) {
+	const bucketNameEnvVar = "FLOW_TESTS_AWS_S3_BUCKET_NAME"
+	if os.Getenv(bucketNameEnvVar) == "" {
+		t.Skipf("skipping test since %s is not set", bucketNameEnvVar)
+	}
+
 	for _, envVar := range []string{
 		"AWS_ACCESS_KEY_ID",
 		"AWS_SECRET_ACCESS_KEY",
@@ -34,13 +40,26 @@ func TestIAMRoleCanIssueSelectFromS3(t *testing.T) {
 	} {
 		t.Setenv(envVar, os.Getenv("FLOW_TESTS_"+envVar))
 	}
-	t.Setenv("PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME", os.Getenv("FLOW_TESTS_AWS_S3_BUCKET_NAME"))
+	t.Setenv("PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME", os.Getenv(bucketNameEnvVar))
 	ctx := t.Context()
 
+	chHost := "localhost"
+	if host := os.Getenv("CI_CLICKHOUSE_HOST"); host != "" {
+		chHost = host
+	}
+	var chPort uint32 = 9000
+	if port := os.Getenv("CI_CLICKHOUSE_NATIVE_PORT"); port != "" {
+		p, err := strconv.ParseUint(port, 10, 32)
+		if err != nil {
+			t.Fatalf("invalid CI_CLICKHOUSE_NATIVE_PORT %q: %v", port, err)
+		}
+		chPort = uint32(p)
+	}
+
 	conn, err := NewClickHouseConnector(ctx, nil,
 		&protos.ClickhouseConfig{
-			Host:       "localhost",
-			Port:       9000,
+			Host:       chHost,
+			Port:       chPort,
 			Database:   "default",
 			DisableTls: true,
 		})

flow/connectors/mysql/ssh_keepalive_test.go

@@ -3,6 +3,8 @@ package connmysql
 import (
 	"context"
 	"os"
+	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -24,11 +26,28 @@ func setupMySQLConnectorWithSSH(ctx context.Context, t *testing.T, proxyName str
 	toxiproxyClient := utils.NewToxiproxyClient(t)
 	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
 
+	mysqlHost := "mysql"
+	if envHost := os.Getenv("CI_MYSQL_HOST"); envHost != "" {
+		mysqlHost = envHost
+	}
+
+	var mysqlPort uint32 = 3306
+	if envPortStr := os.Getenv("CI_MYSQL_PORT"); envPortStr != "" {
+		envPort, err := strconv.ParseUint(strings.TrimSpace(strings.Split(envPortStr, "#")[0]), 10, 32)
+		require.NoError(t, err, "Failed to parse CI_MYSQL_PORT")
+		mysqlPort = uint32(envPort)
+	}
+
+	mysqlRootPass := "cipass"
+	if envPass := os.Getenv("CI_MYSQL_ROOT_PASSWORD"); envPass != "" {
+		mysqlRootPass = envPass
+	}
+
 	connector, err := NewMySqlConnector(ctx, &protos.MySqlConfig{
-		Host:     "mysql",
-		Port:     3306,
+		Host:     mysqlHost,
+		Port:     mysqlPort,
 		User:     "root",
-		Password: "cipass",
+		Password: mysqlRootPass,
 		Database: "mysql",
 		SshConfig: &protos.SSHConfig{
 			Host:     "localhost",

flow/connectors/postgres/ssh_keepalive_test.go

@@ -2,6 +2,9 @@ package connpostgres
 
 import (
 	"context"
+	"os"
+	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -23,9 +26,30 @@ func setupPostgresConnectorWithSSH(ctx context.Context, t *testing.T, proxyName
 	toxiproxyClient := utils.NewToxiproxyClient(t)
 	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
 
+	pgHost := "localhost" // Default for local environments
+
+	// In local set-up environments like Tilt, when a non catalog
+	// instance of Postgres is present, we use it instead of the catalog.
+	if envHost := os.Getenv("PG_HOST"); envHost != "" {
+		pgHost = envHost
+	}
+
+	// In CI, TOXIPROXY_POSTGRES_HOST is set, usually pointing to 'catalog'
+	// network.
+	if toxiproxyPgHost := os.Getenv("TOXIPROXY_POSTGRES_HOST"); toxiproxyPgHost != "" {
+		pgHost = toxiproxyPgHost
+	}
+
+	var pgPort uint32 = 5432
+	if envPortStr := os.Getenv("PG_PORT"); envPortStr != "" {
+		envPort, err := strconv.ParseUint(strings.TrimSpace(strings.Split(envPortStr, "#")[0]), 10, 32)
+		require.NoError(t, err, "Failed to parse PG_PORT")
+		pgPort = uint32(envPort)
+	}
+
 	connector, err := NewPostgresConnector(ctx, nil, &protos.PostgresConfig{
-		Host:     "catalog",
-		Port:     5432,
+		Host:     pgHost,
+		Port:     pgPort,
 		User:     "postgres",
 		Password: "postgres",
 		Database: "postgres",