From 7f880f46e6d84fac0a9bca82d4ddbb2d3ac1379d Mon Sep 17 00:00:00 2001 From: Warren Lee <5959690+wrn14897@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:35:18 -0700 Subject: [PATCH 1/3] feat(otel-collector): support Replicated database + tables in schema seed (HDX-4664) Adds ClickStack OTel collector support for the ClickHouse Replicated (DatabaseReplicated) database engine, so the collector's schema seed and clickhouse-operator v0.0.6 (enableDatabaseSync: true) agree on the default database engine and the ClickStack Helm chart can move to the Replicated engine. - New opt-in env var HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE=Replicated. Before goose runs, the seed ensures the target database uses the Replicated engine: missing -> created with the operator's Keeper path convention; already Replicated -> no-op; non-Replicated + empty -> DROP DATABASE ... SYNC and recreate as Replicated (mirrors the operator's conversion, resolving the startup race from either side); non-Replicated + has tables -> never dropped, loud warning, seed continues (no data loss). - Auto-detected table-engine rewrite: whenever the target database uses the Replicated engine - whether created by the seed or by the operator - the processed schema is rewritten MergeTree -> ReplicatedMergeTree and SummingMergeTree -> ReplicatedSummingMergeTree so table data replicates across replicas. - Smoke tests: new ch-server-replicated (single-node ClickHouse with embedded Keeper + shard/replica macros) and otel-collector-replicated services, plus replicated-schema.bats asserting the Atomic->Replicated conversion, Replicated engines on all tables, and an ingest/query round-trip. - README documentation for the new env var and behavior. --- .../otel-collector-replicated-database.md | 14 ++ packages/otel-collector/README.md | 48 ++++ packages/otel-collector/cmd/migrate/main.go | 219 ++++++++++++++++ .../otel-collector/cmd/migrate/main_test.go | 233 ++++++++++++++++++ .../otel-collector/clickhouse-replicated.xml | 45 ++++ .../basic-insert/assert_query.sql | 1 + .../basic-insert/expected.snap | 2 + .../replicated-schema/basic-insert/input.json | 51 ++++ .../engines/assert_query.sql | 12 + .../replicated-schema/engines/expected.snap | 5 + .../otel-collector/docker-compose.yaml | 50 ++++ .../otel-collector/replicated-schema.bats | 20 ++ smoke-tests/otel-collector/setup_suite.bash | 1 + .../test_helpers/assertions.bash | 4 + 14 files changed, 705 insertions(+) create mode 100644 .changeset/otel-collector-replicated-database.md create mode 100644 smoke-tests/otel-collector/clickhouse-replicated.xml create mode 100644 smoke-tests/otel-collector/data/replicated-schema/basic-insert/assert_query.sql create mode 100644 smoke-tests/otel-collector/data/replicated-schema/basic-insert/expected.snap create mode 100644 smoke-tests/otel-collector/data/replicated-schema/basic-insert/input.json create mode 100644 smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql create mode 100644 smoke-tests/otel-collector/data/replicated-schema/engines/expected.snap create mode 100644 smoke-tests/otel-collector/replicated-schema.bats diff --git a/.changeset/otel-collector-replicated-database.md b/.changeset/otel-collector-replicated-database.md new file mode 100644 index 0000000000..15d2b5fac9 --- /dev/null +++ b/.changeset/otel-collector-replicated-database.md @@ -0,0 +1,14 @@ +--- +'@hyperdx/otel-collector': minor +--- + +Add support for the ClickHouse Replicated (DatabaseReplicated) database engine +to the collector's schema seed. Setting +`HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE=Replicated` makes the seed +ensure the target database uses the Replicated engine (creating it, or +converting an empty non-Replicated database, mirroring clickhouse-operator's +`enableDatabaseSync` behavior; a non-empty database is never dropped). Whenever +the target database uses the Replicated engine — regardless of the env var — +table engines are rewritten to their replicated variants (`MergeTree` → +`ReplicatedMergeTree`, `SummingMergeTree` → `ReplicatedSummingMergeTree`) so +table data replicates across replicas. diff --git a/packages/otel-collector/README.md b/packages/otel-collector/README.md index 95cf13a557..bded457715 100644 --- a/packages/otel-collector/README.md +++ b/packages/otel-collector/README.md @@ -129,6 +129,54 @@ custom OTel configurations without rebuilding the collector. | `https` | core | | `yaml` | core | +## ClickHouse schema seed + +At container start, the entrypoint runs the Go-based seed tool +(`cmd/migrate/main.go`) which applies the idempotent schema SQL from +`docker/otel-collector/schema/seed/` (goose with `WithNoVersioning`, so no +goose tracking table is created). The clickhouse exporter itself runs with +`create_schema: false`. + +### Replicated database engine (HDX-4664) + +By default the seed creates the target database with ClickHouse's default +engine (Atomic). Setting +`HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE=Replicated` makes the seed +ensure the database uses the **Replicated (DatabaseReplicated)** engine +instead, matching clickhouse-operator's `enableDatabaseSync` behavior (table +metadata stored in Keeper): + +- **Missing database** — created with + `ENGINE = Replicated('/clickhouse/databases/', '{shard}', '{replica}')` + (the operator's path convention). +- **Already Replicated** — no-op. +- **Non-Replicated and empty** — dropped and recreated as Replicated. This + mirrors the operator's conversion of the empty Atomic `default` database, so + the collector and operator agree on the engine no matter which side runs + first. +- **Non-Replicated with tables** — never dropped (that would lose data); the + seed logs a warning and continues against the existing database. + +Independently of this env var, whenever the target database uses the +Replicated engine — whether created by the seed or by clickhouse-operator — +the seed rewrites the table engines in the schema to their replicated +variants (`MergeTree` → `ReplicatedMergeTree`, `SummingMergeTree` → +`ReplicatedSummingMergeTree`) so table **data** replicates across replicas +(plain MergeTree tables in a Replicated database only replicate metadata). + +Notes: + +- Requires ClickHouse Keeper (or ZooKeeper) plus `{shard}`/`{replica}` macros + on the server, as in operator-managed deployments. See + `smoke-tests/otel-collector/clickhouse-replicated.xml` for a single-node + example. +- The experimental PromQL `TimeSeries` schema (`ENABLE_PROMQL=true`) is left + untouched and is not replication-aware. +- The legacy exporter-managed schema path + (`HYPERDX_OTEL_EXPORTER_CREATE_LEGACY_SCHEMA=true`, also implied by + `HYPERDX_OTEL_EXPORTER_CLICKHOUSE_JSON_ENABLE=true`) skips the seed tool + entirely, so this env var has no effect there. + ## Ingesting Datadog traces, metrics, and logs The `datadogreceiver` contrib component is compiled into the binary so a diff --git a/packages/otel-collector/cmd/migrate/main.go b/packages/otel-collector/cmd/migrate/main.go index f58608657b..0992247aaa 100644 --- a/packages/otel-collector/cmd/migrate/main.go +++ b/packages/otel-collector/cmd/migrate/main.go @@ -9,11 +9,13 @@ import ( "crypto/tls" "crypto/x509" "database/sql" + "errors" "fmt" "log" "net/url" "os" "path/filepath" + "regexp" "sort" "strconv" "strings" @@ -31,6 +33,14 @@ type Config struct { Password string Database string + // DatabaseEngine selects the engine used when creating the target + // database. Empty (default) keeps the current behavior where the seed SQL + // creates the database with the server default (Atomic). "Replicated" + // makes the seed ensure the database uses the Replicated + // (DatabaseReplicated) engine, matching clickhouse-operator's + // enableDatabaseSync behavior. + DatabaseEngine string + // Table TTL (Go duration string, e.g. "720h") TablesTTL string @@ -53,6 +63,7 @@ func loadConfig() (*Config, error) { User: getEnv("CLICKHOUSE_USER", "default"), Password: getEnv("CLICKHOUSE_PASSWORD", ""), Database: getEnv("HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE", "default"), + DatabaseEngine: getEnv("HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE", ""), TablesTTL: getEnv("HYPERDX_OTEL_EXPORTER_TABLES_TTL", "720h"), TLSCAFile: getEnv("CLICKHOUSE_TLS_CA_FILE", ""), TLSCertFile: getEnv("CLICKHOUSE_TLS_CERT_FILE", ""), @@ -62,6 +73,15 @@ func loadConfig() (*Config, error) { MaxRetries: 5, } + // Validate the database engine. Only the default (Atomic) and Replicated + // engines are supported; anything else is a deterministic + // misconfiguration. + if !isDefaultEngine(cfg.DatabaseEngine) && !isReplicatedEngine(cfg.DatabaseEngine) { + return nil, fmt.Errorf( + "invalid HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE %q (supported: %q, %q)", + cfg.DatabaseEngine, "Atomic", "Replicated") + } + // Get schema directory from CLI argument if len(os.Args) < 2 { return nil, fmt.Errorf("usage: %s ", os.Args[0]) @@ -379,6 +399,175 @@ func supportsFullTextSearch(major, minor int) bool { return major > 26 || (major == 26 && minor >= 2) } +// replicatedEngineName is the engine name reported by system.databases for +// databases using the Replicated (DatabaseReplicated) engine. +const replicatedEngineName = "Replicated" + +// isDefaultEngine returns true when the configured database engine keeps the +// default behavior (the seed SQL creates the database with the server default, +// Atomic). +func isDefaultEngine(engine string) bool { + return engine == "" || strings.EqualFold(engine, "atomic") +} + +// isReplicatedEngine returns true when the configured database engine requests +// the Replicated (DatabaseReplicated) engine. +func isReplicatedEngine(engine string) bool { + return strings.EqualFold(engine, replicatedEngineName) +} + +// databaseAction describes what ensureReplicatedDatabase should do with the +// target database. +type databaseAction int + +const ( + // dbActionCreate: the database does not exist; create it as Replicated. + dbActionCreate databaseAction = iota + // dbActionNone: the database already uses the Replicated engine. + dbActionNone + // dbActionConvert: the database exists with a non-Replicated engine and is + // empty; drop it and recreate it as Replicated. This mirrors + // clickhouse-operator's enableDatabaseSync conversion so the collector and + // operator agree on the engine no matter which side runs first. + dbActionConvert + // dbActionKeep: the database exists with a non-Replicated engine and + // already has tables. Never drop it (that would lose data); keep it as-is. + dbActionKeep +) + +// decideDatabaseAction is the pure decision function behind +// ensureReplicatedDatabase, factored out for testability. +func decideDatabaseAction(exists bool, engine string, tableCount uint64) databaseAction { + if !exists { + return dbActionCreate + } + if engine == replicatedEngineName { + return dbActionNone + } + if tableCount == 0 { + return dbActionConvert + } + return dbActionKeep +} + +// replicatedDatabaseDDL returns the DDL for creating the target database with +// the Replicated engine. The Keeper path follows clickhouse-operator's +// convention (/clickhouse/databases/) and the shard/replica names come +// from the server-side {shard}/{replica} macros. +func replicatedDatabaseDDL(database string) string { + return fmt.Sprintf( + "CREATE DATABASE IF NOT EXISTS `%s` ENGINE = Replicated('/clickhouse/databases/%s', '{shard}', '{replica}')", + database, database) +} + +// getDatabaseEngine queries system.databases for the target database's engine. +// exists is false when the database does not exist. +func getDatabaseEngine(ctx context.Context, db *sql.DB, database string) (engine string, exists bool, err error) { + err = db.QueryRowContext(ctx, + "SELECT engine FROM system.databases WHERE name = ?", database).Scan(&engine) + if errors.Is(err, sql.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("failed to query engine of database %q: %w", database, err) + } + return engine, true, nil +} + +// countDatabaseTables returns the number of tables in the target database. +func countDatabaseTables(ctx context.Context, db *sql.DB, database string) (uint64, error) { + var count uint64 + err := db.QueryRowContext(ctx, + "SELECT count() FROM system.tables WHERE database = ?", database).Scan(&count) + if err != nil { + return 0, fmt.Errorf("failed to count tables in database %q: %w", database, err) + } + return count, nil +} + +// ensureReplicatedDatabase makes sure the target database uses the Replicated +// engine before the schema seed runs: +// - missing -> create it as Replicated +// - already Replicated -> nothing to do +// - other engine, empty -> drop + recreate as Replicated (mirrors +// clickhouse-operator's enableDatabaseSync conversion; resolves the +// startup race between the collector seed and the operator) +// - other engine, has tables -> keep as-is and warn; dropping it would lose +// data +func ensureReplicatedDatabase(ctx context.Context, db *sql.DB, database string) error { + engine, exists, err := getDatabaseEngine(ctx, db, database) + if err != nil { + return err + } + + var tableCount uint64 + if exists { + tableCount, err = countDatabaseTables(ctx, db, database) + if err != nil { + return err + } + } + + switch decideDatabaseAction(exists, engine, tableCount) { + case dbActionNone: + log.Printf("Database %q already uses the Replicated engine", database) + return nil + case dbActionKeep: + log.Printf("WARNING: Database %q uses the %s engine and already has %d table(s); refusing to drop it to avoid data loss. It will NOT be converted to the Replicated engine.", + database, engine, tableCount) + return nil + case dbActionConvert: + log.Printf("Database %q uses the %s engine and is empty; dropping and recreating it with the Replicated engine", database, engine) + if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s` SYNC", database)); err != nil { + return fmt.Errorf("failed to drop database %q: %w", database, err) + } + case dbActionCreate: + log.Printf("Database %q does not exist; creating it with the Replicated engine", database) + } + + if _, err := db.ExecContext(ctx, replicatedDatabaseDDL(database)); err != nil { + return fmt.Errorf("failed to create Replicated database %q: %w", database, err) + } + return nil +} + +// mergeTreeEngineRe matches the plain (Summing)MergeTree engine lines emitted +// by the seed schema files. It is anchored to whole lines so other engines +// (e.g. the experimental TimeSeries engine) are never touched. +var mergeTreeEngineRe = regexp.MustCompile(`(?m)^ENGINE = (MergeTree|SummingMergeTree)\b`) + +// rewriteEnginesForReplicated rewrites the table engines in the processed +// schema directory to their Replicated variants (MergeTree -> +// ReplicatedMergeTree, SummingMergeTree -> ReplicatedSummingMergeTree). In a +// Replicated database, plain MergeTree tables would have replicated metadata +// but local (non-replicated) data, so replicated table engines are required +// for correctness on multi-replica clusters. +func rewriteEnginesForReplicated(tempDir string) error { + return filepath.Walk(tempDir, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + if info.IsDir() || !strings.HasSuffix(path, ".sql") { + return nil + } + + content, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("failed to read schema file %s: %w", path, err) + } + + rewritten := mergeTreeEngineRe.ReplaceAll(content, []byte("ENGINE = Replicated$1")) + if string(rewritten) == string(content) { + return nil + } + + if err := os.WriteFile(path, rewritten, 0644); err != nil { + return fmt.Errorf("failed to write schema file %s: %w", path, err) + } + return nil + }) +} + // swapLogsSchemaForCompat replaces the full-text-search logs schema with the // compatibility variant (bloom_filter indexes) in the processed temp directory. // It removes 00002_otel_logs.sql and renames 00002_otel_logs_compat.sql to @@ -506,6 +695,25 @@ func main() { log.Fatalf("Failed to determine ClickHouse version: %v", err) } + // When the Replicated database engine is requested, make sure the target + // database uses it before seeding tables. + if isReplicatedEngine(cfg.DatabaseEngine) { + log.Printf("Requested database engine: %s", replicatedEngineName) + if err := ensureReplicatedDatabase(ctx, db, cfg.Database); err != nil { + log.Fatalf("Failed to ensure Replicated database %q: %v", cfg.Database, err) + } + } + + // Detect the actual engine of the target database. If it uses the + // Replicated engine — whether created above or e.g. by + // clickhouse-operator's enableDatabaseSync — the table engines in the + // schema are rewritten to their Replicated variants below. + dbEngine, dbExists, err := getDatabaseEngine(ctx, db, cfg.Database) + if err != nil { + log.Fatalf("Failed to determine engine of database %q: %v", cfg.Database, err) + } + targetIsReplicated := dbExists && dbEngine == replicatedEngineName + // Parse tables TTL tablesTTLExpr, err := ttlToClickHouseInterval(cfg.TablesTTL) if err != nil { @@ -546,6 +754,17 @@ func main() { } } + // Rewrite table engines to their Replicated variants when the target + // database uses the Replicated engine, so table data is replicated across + // replicas (plain MergeTree tables in a Replicated database only replicate + // metadata). + if targetIsReplicated { + log.Printf("Database %q uses the Replicated engine; rewriting table engines to Replicated variants", cfg.Database) + if err := rewriteEnginesForReplicated(tempDir); err != nil { + log.Fatalf("Failed to rewrite table engines for Replicated database: %v", err) + } + } + // List SQL files sqlFiles, err := listSQLFiles(tempDir) if err != nil { diff --git a/packages/otel-collector/cmd/migrate/main_test.go b/packages/otel-collector/cmd/migrate/main_test.go index 023bf77f85..69686473a2 100644 --- a/packages/otel-collector/cmd/migrate/main_test.go +++ b/packages/otel-collector/cmd/migrate/main_test.go @@ -966,3 +966,236 @@ SETTINGS ttl_only_drop_parts = 1;` t.Errorf("processSchemaDir output mismatch\ngot:\n%s\nwant:\n%s", string(got), expected) } } + +// --------------------------------------------------------------------------- +// Replicated database engine support (HDX-4664) +// --------------------------------------------------------------------------- + +func TestDatabaseEngineHelpers(t *testing.T) { + defaultCases := map[string]bool{ + "": true, + "Atomic": true, + "atomic": true, + "ATOMIC": true, + "Replicated": false, + "bogus": false, + } + for engine, want := range defaultCases { + if got := isDefaultEngine(engine); got != want { + t.Errorf("isDefaultEngine(%q): got %v, want %v", engine, got, want) + } + } + + replicatedCases := map[string]bool{ + "Replicated": true, + "replicated": true, + "REPLICATED": true, + "": false, + "Atomic": false, + "bogus": false, + } + for engine, want := range replicatedCases { + if got := isReplicatedEngine(engine); got != want { + t.Errorf("isReplicatedEngine(%q): got %v, want %v", engine, got, want) + } + } +} + +func TestLoadConfigDatabaseEngine(t *testing.T) { + origArgs := os.Args + t.Cleanup(func() { os.Args = origArgs }) + + t.Run("accepts supported engines", func(t *testing.T) { + for _, engine := range []string{"", "Atomic", "atomic", "Replicated", "replicated"} { + dir := t.TempDir() + os.Args = []string{"migrate", dir} + t.Setenv("HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE", engine) + + cfg, err := loadConfig() + if err != nil { + t.Fatalf("engine %q: unexpected error: %v", engine, err) + } + if cfg.DatabaseEngine != engine { + t.Errorf("engine %q: DatabaseEngine got %q", engine, cfg.DatabaseEngine) + } + } + }) + + t.Run("rejects unsupported engine", func(t *testing.T) { + dir := t.TempDir() + os.Args = []string{"migrate", dir} + t.Setenv("HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE", "Ordinary") + + _, err := loadConfig() + if err == nil { + t.Fatal("expected error for unsupported database engine") + } + if !strings.Contains(err.Error(), "HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE") { + t.Errorf("expected engine env var in error, got: %v", err) + } + }) +} + +func TestDecideDatabaseAction(t *testing.T) { + tests := []struct { + name string + exists bool + engine string + tableCount uint64 + want databaseAction + }{ + {"missing database is created", false, "", 0, dbActionCreate}, + {"already Replicated is a no-op", true, "Replicated", 0, dbActionNone}, + {"already Replicated with tables is a no-op", true, "Replicated", 12, dbActionNone}, + {"empty Atomic is converted", true, "Atomic", 0, dbActionConvert}, + {"non-empty Atomic is kept", true, "Atomic", 3, dbActionKeep}, + {"non-empty other engine is kept", true, "Ordinary", 1, dbActionKeep}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := decideDatabaseAction(tt.exists, tt.engine, tt.tableCount); got != tt.want { + t.Errorf("decideDatabaseAction(%v, %q, %d): got %v, want %v", + tt.exists, tt.engine, tt.tableCount, got, tt.want) + } + }) + } +} + +func TestReplicatedDatabaseDDL(t *testing.T) { + got := replicatedDatabaseDDL("default") + want := "CREATE DATABASE IF NOT EXISTS `default` ENGINE = Replicated('/clickhouse/databases/default', '{shard}', '{replica}')" + if got != want { + t.Errorf("replicatedDatabaseDDL:\ngot: %s\nwant: %s", got, want) + } +} + +func TestRewriteEnginesForReplicated(t *testing.T) { + t.Run("rewrites MergeTree and SummingMergeTree engines", func(t *testing.T) { + dir := t.TempDir() + sqlContent := `CREATE TABLE IF NOT EXISTS default.otel_logs +(col1 String) +ENGINE = MergeTree +PARTITION BY toDate(Timestamp) +ORDER BY col1 +SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1; + +CREATE TABLE IF NOT EXISTS default.otel_logs_kv_rollup_15m +(col1 String) +ENGINE = SummingMergeTree +ORDER BY col1; +` + if err := os.WriteFile(filepath.Join(dir, "001_test.sql"), []byte(sqlContent), 0644); err != nil { + t.Fatal(err) + } + + if err := rewriteEnginesForReplicated(dir); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + got, err := os.ReadFile(filepath.Join(dir, "001_test.sql")) + if err != nil { + t.Fatal(err) + } + + want := `CREATE TABLE IF NOT EXISTS default.otel_logs +(col1 String) +ENGINE = ReplicatedMergeTree +PARTITION BY toDate(Timestamp) +ORDER BY col1 +SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1; + +CREATE TABLE IF NOT EXISTS default.otel_logs_kv_rollup_15m +(col1 String) +ENGINE = ReplicatedSummingMergeTree +ORDER BY col1; +` + if string(got) != want { + t.Errorf("rewriteEnginesForReplicated output mismatch\ngot:\n%s\nwant:\n%s", string(got), want) + } + }) + + t.Run("leaves other engines untouched", func(t *testing.T) { + dir := t.TempDir() + sqlContent := `CREATE TABLE IF NOT EXISTS default.metrics_ts +ENGINE = TimeSeries +SETTINGS allow_experimental_time_series_table = 1; +` + path := filepath.Join(dir, "001_timeseries.sql") + if err := os.WriteFile(path, []byte(sqlContent), 0644); err != nil { + t.Fatal(err) + } + + if err := rewriteEnginesForReplicated(dir); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(got) != sqlContent { + t.Errorf("TimeSeries schema should be untouched\ngot:\n%s\nwant:\n%s", string(got), sqlContent) + } + }) + + t.Run("ignores non-SQL files", func(t *testing.T) { + dir := t.TempDir() + content := "ENGINE = MergeTree\n" + path := filepath.Join(dir, "README.md") + if err := os.WriteFile(path, []byte(content), 0644); err != nil { + t.Fatal(err) + } + + if err := rewriteEnginesForReplicated(dir); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(got) != content { + t.Errorf("non-SQL file should be untouched, got %q", string(got)) + } + }) + + t.Run("rewrites the real seed schema files", func(t *testing.T) { + // Run against the actual seed directory (after macro processing, like + // main() does) to guarantee every shipped schema is rewritten to a + // Replicated-compatible engine. + seedDir := filepath.Join("..", "..", "..", "..", "docker", "otel-collector", "schema", "seed") + if _, err := os.Stat(seedDir); err != nil { + t.Skipf("seed directory not available: %v", err) + } + + tempDir, err := processSchemaDir(seedDir, "default", "toIntervalDay(30)") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(tempDir) + + if err := rewriteEnginesForReplicated(tempDir); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + files, err := listSQLFiles(tempDir) + if err != nil { + t.Fatal(err) + } + for _, f := range files { + content, err := os.ReadFile(filepath.Join(tempDir, f)) + if err != nil { + t.Fatal(err) + } + if mergeTreeEngineRe.Match(content) { + t.Errorf("%s: still contains a non-replicated MergeTree engine after rewrite", f) + } + if f == "00008_otel_metrics_timeseries.sql" { + if !strings.Contains(string(content), "ENGINE = TimeSeries") { + t.Errorf("%s: TimeSeries engine should be untouched", f) + } + } + } + }) +} diff --git a/smoke-tests/otel-collector/clickhouse-replicated.xml b/smoke-tests/otel-collector/clickhouse-replicated.xml new file mode 100644 index 0000000000..f587aa747a --- /dev/null +++ b/smoke-tests/otel-collector/clickhouse-replicated.xml @@ -0,0 +1,45 @@ + + + + + 9181 + 1 + /var/lib/clickhouse/coordination/log + /var/lib/clickhouse/coordination/snapshots + + 10000 + 30000 + + + + 1 + localhost + 9234 + + + + + + + localhost + 9181 + + + + + s1 + r1 + + + + ch-server-replicated + diff --git a/smoke-tests/otel-collector/data/replicated-schema/basic-insert/assert_query.sql b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/assert_query.sql new file mode 100644 index 0000000000..9450365303 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/assert_query.sql @@ -0,0 +1 @@ +SELECT SeverityText, SeverityNumber, Body FROM otel_logs WHERE ResourceAttributes['suite-id'] = 'replicated-schema' AND ResourceAttributes['test-id'] = 'basic-insert' ORDER BY (toStartOfFiveMinutes(Timestamp), Timestamp) FORMAT CSV diff --git a/smoke-tests/otel-collector/data/replicated-schema/basic-insert/expected.snap b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/expected.snap new file mode 100644 index 0000000000..57c83de95b --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/expected.snap @@ -0,0 +1,2 @@ +"info",9,"replicated schema test log entry" +"warn",13,"replicated schema warning message" diff --git a/smoke-tests/otel-collector/data/replicated-schema/basic-insert/input.json b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/input.json new file mode 100644 index 0000000000..ead6af1c41 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/basic-insert/input.json @@ -0,0 +1,51 @@ +{ + "resourceLogs": [ + { + "resource": { + "attributes": [ + { + "key": "suite-id", + "value": { + "stringValue": "replicated-schema" + } + }, + { + "key": "test-id", + "value": { + "stringValue": "basic-insert" + } + }, + { + "key": "service.name", + "value": { + "stringValue": "replicated-test-service" + } + } + ] + }, + "scopeLogs": [ + { + "scope": {}, + "logRecords": [ + { + "timeUnixNano": "1901999580000000000", + "severityNumber": 9, + "severityText": "INFO", + "body": { + "stringValue": "replicated schema test log entry" + } + }, + { + "timeUnixNano": "1901999580000000001", + "severityNumber": 13, + "severityText": "WARN", + "body": { + "stringValue": "replicated schema warning message" + } + } + ] + } + ] + } + ] +} diff --git a/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql b/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql new file mode 100644 index 0000000000..b4702f9974 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql @@ -0,0 +1,12 @@ +SELECT engine FROM system.databases WHERE name = 'default' FORMAT CSV; +SELECT name, engine +FROM system.tables +WHERE database = 'default' + AND engine != 'MaterializedView' + AND engine NOT LIKE 'Replicated%' +ORDER BY name +FORMAT CSV; +SELECT engine FROM system.tables WHERE database = 'default' AND name = 'otel_logs' FORMAT CSV; +SELECT engine FROM system.tables WHERE database = 'default' AND name = 'otel_traces' FORMAT CSV; +SELECT engine FROM system.tables WHERE database = 'default' AND name = 'otel_logs_kv_rollup_15m' FORMAT CSV; +SELECT engine FROM system.tables WHERE database = 'default' AND name = 'otel_traces_kv_rollup_15m' FORMAT CSV; diff --git a/smoke-tests/otel-collector/data/replicated-schema/engines/expected.snap b/smoke-tests/otel-collector/data/replicated-schema/engines/expected.snap new file mode 100644 index 0000000000..fdb73e1e64 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/engines/expected.snap @@ -0,0 +1,5 @@ +"Replicated" +"ReplicatedMergeTree" +"ReplicatedMergeTree" +"ReplicatedSummingMergeTree" +"ReplicatedSummingMergeTree" diff --git a/smoke-tests/otel-collector/docker-compose.yaml b/smoke-tests/otel-collector/docker-compose.yaml index 21d623e22b..6f2ff8517a 100644 --- a/smoke-tests/otel-collector/docker-compose.yaml +++ b/smoke-tests/otel-collector/docker-compose.yaml @@ -156,6 +156,56 @@ services: ch-server-compat: condition: service_healthy + ch-server-replicated: + image: clickhouse/clickhouse-server:26.5-alpine + environment: + CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 + volumes: + - ../../docker/clickhouse/local/config.xml:/etc/clickhouse-server/config.xml + - ../../docker/clickhouse/local/users.xml:/etc/clickhouse-server/users.xml + # Enables the embedded Keeper and {shard}/{replica} macros required by + # the Replicated database engine. + - ./clickhouse-replicated.xml:/etc/clickhouse-server/config.d/clickhouse-replicated.xml + ports: + - 39000:9000 + - 38123:8123 + networks: + - internal + healthcheck: + test: + wget -O /dev/null --no-verbose --tries=1 http://127.0.0.1:8123/ping || + exit 1 + interval: 5s + timeout: 3s + retries: 5 + start_period: 10s + otel-collector-replicated: + build: + context: ../.. + dockerfile: docker/otel-collector/Dockerfile + target: dev + args: + OTEL_COLLECTOR_VERSION: ${OTEL_COLLECTOR_VERSION:-0.155.0} + OTEL_COLLECTOR_CORE_VERSION: ${OTEL_COLLECTOR_CORE_VERSION:-1.61.0} + environment: + - CLICKHOUSE_ENDPOINT=tcp://ch-server-replicated:9000?dial_timeout=10s + - CLICKHOUSE_PROMETHEUS_METRICS_ENDPOINT=ch-server-replicated:9363 + - CLICKHOUSE_USER=default + - CLICKHOUSE_PASSWORD= + - HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE=default + - HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE=Replicated + - HYPERDX_LOG_LEVEL=info + # See the note on the otel-collector service above for rationale. + - HYPERDX_OTEL_BATCH_TIMEOUT=100ms + # OPAMP_SERVER_URL is intentionally not set to run in standalone mode + ports: + - 54318:4318 # OTLP http receiver + networks: + - internal + depends_on: + ch-server-replicated: + condition: service_healthy + networks: internal: name: 'smoke-test-internal-network' diff --git a/smoke-tests/otel-collector/replicated-schema.bats b/smoke-tests/otel-collector/replicated-schema.bats new file mode 100644 index 0000000000..ed3313e915 --- /dev/null +++ b/smoke-tests/otel-collector/replicated-schema.bats @@ -0,0 +1,20 @@ +#!/usr/bin/env bats + +# HDX-4664: the schema seed supports the Replicated (DatabaseReplicated) +# database engine. The ch-server-replicated service boots with an empty Atomic +# `default` database (ClickHouse's own bootstrap), so these tests also cover +# the seed's Atomic -> Replicated conversion path used when the collector +# starts before clickhouse-operator's enableDatabaseSync. + +load 'test_helpers/utilities.bash' +load 'test_helpers/assertions.bash' + +@test "replicated schema should create the database and tables with Replicated engines" { + assert_test_data_replicated "data/replicated-schema/engines" +} + +@test "replicated schema should insert and query log data correctly" { + emit_otel_data "http://localhost:54318" "data/replicated-schema/basic-insert" + wait_for_rows 39000 "SELECT count() FROM otel_logs WHERE ResourceAttributes['suite-id'] = 'replicated-schema' AND ResourceAttributes['test-id'] = 'basic-insert'" 2 + assert_test_data_replicated "data/replicated-schema/basic-insert" +} diff --git a/smoke-tests/otel-collector/setup_suite.bash b/smoke-tests/otel-collector/setup_suite.bash index 24cd934109..7f45987dc5 100644 --- a/smoke-tests/otel-collector/setup_suite.bash +++ b/smoke-tests/otel-collector/setup_suite.bash @@ -9,6 +9,7 @@ setup_suite() { wait_for_ready "otel-collector" wait_for_ready "otel-collector-json" wait_for_ready "otel-collector-compat" + wait_for_ready "otel-collector-replicated" wait_for_ready "otel-collector-custom" } diff --git a/smoke-tests/otel-collector/test_helpers/assertions.bash b/smoke-tests/otel-collector/test_helpers/assertions.bash index a4ed90c0ca..068ee49280 100644 --- a/smoke-tests/otel-collector/test_helpers/assertions.bash +++ b/smoke-tests/otel-collector/test_helpers/assertions.bash @@ -6,6 +6,10 @@ assert_test_data_compat() { _assert_test_data_on_port "29000" "$@" } +assert_test_data_replicated() { + _assert_test_data_on_port "39000" "$@" +} + _assert_test_data_on_port() { local port=$1 local testdir=$2 From 5b28d74ff41585af18d9739a2e05933c46382015 Mon Sep 17 00:00:00 2001 From: Warren Lee <5959690+wrn14897@users.noreply.github.com> Date: Thu, 6 Aug 2026 22:04:03 -0700 Subject: [PATCH 2/3] fix(otel-collector): make Atomic->Replicated conversion race-safe via rename fence The empty-database check and the DROP DATABASE SYNC were not atomic: a table created by another schema manager or collector replica between the count and the drop would be cascade-dropped, defeating the non-empty-database safeguard. Instead of dropping based on the initial check, the conversion now atomically RENAMEs the old database aside (_pre_replicated_), fencing all name-based writers, creates the Replicated database under the original name, and drops the renamed database only after re-verifying it is still empty. A table that raced the check survives in the renamed database with a loud warning instead of being destroyed. Adds a smoke-test assertion that a clean conversion leaves no fence database behind. --- .../otel-collector-replicated-database.md | 5 +- packages/otel-collector/README.md | 13 +++-- packages/otel-collector/cmd/migrate/main.go | 52 ++++++++++++++++--- .../otel-collector/cmd/migrate/main_test.go | 17 ++++++ .../engines/assert_query.sql | 4 ++ 5 files changed, 79 insertions(+), 12 deletions(-) diff --git a/.changeset/otel-collector-replicated-database.md b/.changeset/otel-collector-replicated-database.md index 15d2b5fac9..3d0ac0ffbd 100644 --- a/.changeset/otel-collector-replicated-database.md +++ b/.changeset/otel-collector-replicated-database.md @@ -7,7 +7,10 @@ to the collector's schema seed. Setting `HYPERDX_OTEL_EXPORTER_CLICKHOUSE_DATABASE_ENGINE=Replicated` makes the seed ensure the target database uses the Replicated engine (creating it, or converting an empty non-Replicated database, mirroring clickhouse-operator's -`enableDatabaseSync` behavior; a non-empty database is never dropped). Whenever +`enableDatabaseSync` behavior; a non-empty database is never dropped, and the +conversion renames the old database aside and re-verifies emptiness before +dropping it, so tables created concurrently with the check are preserved +rather than cascade-dropped). Whenever the target database uses the Replicated engine — regardless of the env var — table engines are rewritten to their replicated variants (`MergeTree` → `ReplicatedMergeTree`, `SummingMergeTree` → `ReplicatedSummingMergeTree`) so diff --git a/packages/otel-collector/README.md b/packages/otel-collector/README.md index bded457715..051fc07010 100644 --- a/packages/otel-collector/README.md +++ b/packages/otel-collector/README.md @@ -150,10 +150,15 @@ metadata stored in Keeper): `ENGINE = Replicated('/clickhouse/databases/', '{shard}', '{replica}')` (the operator's path convention). - **Already Replicated** — no-op. -- **Non-Replicated and empty** — dropped and recreated as Replicated. This - mirrors the operator's conversion of the empty Atomic `default` database, so - the collector and operator agree on the engine no matter which side runs - first. +- **Non-Replicated and empty** — converted to Replicated. This mirrors the + operator's conversion of the empty Atomic `default` database, so the + collector and operator agree on the engine no matter which side runs first. + The conversion is race-safe: the old database is atomically renamed aside + (`_pre_replicated_`), the Replicated database is created under + the original name, and the renamed database is dropped only after + re-verifying it is still empty. A table created concurrently with the + emptiness check is preserved in the renamed database (with a loud warning) + instead of being cascade-dropped. - **Non-Replicated with tables** — never dropped (that would lose data); the seed logs a warning and continues against the existing database. diff --git a/packages/otel-collector/cmd/migrate/main.go b/packages/otel-collector/cmd/migrate/main.go index 0992247aaa..9375c8d801 100644 --- a/packages/otel-collector/cmd/migrate/main.go +++ b/packages/otel-collector/cmd/migrate/main.go @@ -426,9 +426,12 @@ const ( // dbActionNone: the database already uses the Replicated engine. dbActionNone // dbActionConvert: the database exists with a non-Replicated engine and is - // empty; drop it and recreate it as Replicated. This mirrors - // clickhouse-operator's enableDatabaseSync conversion so the collector and - // operator agree on the engine no matter which side runs first. + // empty; convert it to Replicated. This mirrors clickhouse-operator's + // enableDatabaseSync conversion so the collector and operator agree on the + // engine no matter which side runs first. The conversion renames the old + // database aside first (atomic fence) and only drops it after re-verifying + // it is still empty, so a table created concurrently with the emptiness + // check is preserved instead of being cascade-dropped. dbActionConvert // dbActionKeep: the database exists with a non-Replicated engine and // already has tables. Never drop it (that would lose data); keep it as-is. @@ -460,6 +463,13 @@ func replicatedDatabaseDDL(database string) string { database, database) } +// renamedDatabaseName returns the fence name the target database is renamed +// to during the Replicated conversion. The timestamp suffix avoids collisions +// with leftovers from a previously crashed conversion. +func renamedDatabaseName(database string, now time.Time) string { + return fmt.Sprintf("%s_pre_replicated_%d", database, now.Unix()) +} + // getDatabaseEngine queries system.databases for the target database's engine. // exists is false when the database does not exist. func getDatabaseEngine(ctx context.Context, db *sql.DB, database string) (engine string, exists bool, err error) { @@ -489,11 +499,20 @@ func countDatabaseTables(ctx context.Context, db *sql.DB, database string) (uint // engine before the schema seed runs: // - missing -> create it as Replicated // - already Replicated -> nothing to do -// - other engine, empty -> drop + recreate as Replicated (mirrors +// - other engine, empty -> convert to Replicated (mirrors // clickhouse-operator's enableDatabaseSync conversion; resolves the // startup race between the collector seed and the operator) // - other engine, has tables -> keep as-is and warn; dropping it would lose // data +// +// The conversion never drops a database based on the initial emptiness check +// alone: the emptiness check and the drop are not atomic, so a table created +// in between by another schema manager or collector replica would be +// cascade-dropped. Instead the old database is atomically RENAMEd aside +// (fencing all name-based writers), the Replicated database is created under +// the original name, and the renamed database is dropped only after +// re-verifying it is still empty. A table that raced the check therefore +// survives in the renamed database instead of being destroyed. func ensureReplicatedDatabase(ctx context.Context, db *sql.DB, database string) error { engine, exists, err := getDatabaseEngine(ctx, db, database) if err != nil { @@ -517,10 +536,29 @@ func ensureReplicatedDatabase(ctx context.Context, db *sql.DB, database string) database, engine, tableCount) return nil case dbActionConvert: - log.Printf("Database %q uses the %s engine and is empty; dropping and recreating it with the Replicated engine", database, engine) - if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s` SYNC", database)); err != nil { - return fmt.Errorf("failed to drop database %q: %w", database, err) + renamed := renamedDatabaseName(database, time.Now()) + log.Printf("Database %q uses the %s engine and is empty; renaming it to %q and recreating it with the Replicated engine", database, engine, renamed) + if _, err := db.ExecContext(ctx, fmt.Sprintf("RENAME DATABASE `%s` TO `%s`", database, renamed)); err != nil { + return fmt.Errorf("failed to rename database %q to %q: %w", database, renamed, err) + } + if _, err := db.ExecContext(ctx, replicatedDatabaseDDL(database)); err != nil { + return fmt.Errorf("failed to create Replicated database %q: %w", database, err) + } + // The renamed database is fenced off from name-based writers, so this + // re-check is stable: drop it only if it is provably still empty. + renamedCount, err := countDatabaseTables(ctx, db, renamed) + if err != nil { + return err + } + if renamedCount > 0 { + log.Printf("WARNING: Database %q gained %d table(s) concurrently with the Replicated conversion; it has been preserved as %q instead of being dropped. Move any needed tables into %q and drop %q manually.", + database, renamedCount, renamed, database, renamed) + return nil } + if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s` SYNC", renamed)); err != nil { + return fmt.Errorf("failed to drop empty renamed database %q: %w", renamed, err) + } + return nil case dbActionCreate: log.Printf("Database %q does not exist; creating it with the Replicated engine", database) } diff --git a/packages/otel-collector/cmd/migrate/main_test.go b/packages/otel-collector/cmd/migrate/main_test.go index 69686473a2..0f2f953e59 100644 --- a/packages/otel-collector/cmd/migrate/main_test.go +++ b/packages/otel-collector/cmd/migrate/main_test.go @@ -1070,6 +1070,23 @@ func TestReplicatedDatabaseDDL(t *testing.T) { } } +func TestRenamedDatabaseName(t *testing.T) { + now := time.Unix(1753000000, 0) + got := renamedDatabaseName("default", now) + want := "default_pre_replicated_1753000000" + if got != want { + t.Errorf("renamedDatabaseName: got %q, want %q", got, want) + } + + // The timestamp suffix must make names from different conversion attempts + // distinct, so a leftover from a crashed run never collides with the + // rename target of a later run. + other := renamedDatabaseName("default", now.Add(time.Second)) + if other == got { + t.Errorf("renamedDatabaseName: expected distinct names across attempts, got %q twice", got) + } +} + func TestRewriteEnginesForReplicated(t *testing.T) { t.Run("rewrites MergeTree and SummingMergeTree engines", func(t *testing.T) { dir := t.TempDir() diff --git a/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql b/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql index b4702f9974..247246966b 100644 --- a/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql +++ b/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql @@ -1,4 +1,8 @@ SELECT engine FROM system.databases WHERE name = 'default' FORMAT CSV; +-- The Atomic -> Replicated conversion renames the old database aside and +-- drops it after re-verifying emptiness; a clean conversion must not leave +-- the fence database behind. +SELECT name FROM system.databases WHERE name LIKE 'default_pre_replicated_%' FORMAT CSV; SELECT name, engine FROM system.tables WHERE database = 'default' From 160369dede93d945637c85f6be81bb40828405e5 Mon Sep 17 00:00:00 2001 From: Warren Lee <5959690+wrn14897@users.noreply.github.com> Date: Thu, 6 Aug 2026 22:40:05 -0700 Subject: [PATCH 3/3] fix(otel-collector): recover interrupted Replicated conversions instead of stranding data If creating the Replicated database failed after the old database was renamed aside (e.g. Keeper not ready at bootstrap), the seed exited fatally with the target database absent, and a restart created a fresh database while any table that raced into the fence stayed stranded there permanently and silently. Two complementary fixes: - Best-effort rollback: when the CREATE fails after the rename and the target database was not recreated concurrently, rename the fence back before returning the error, so the restart retries against the intact original database. The rename is metadata-local, so it succeeds precisely in the Keeper-down case that made the CREATE fail. - Startup fence recovery: before the conversion decision, converge fence databases left behind by any interrupted conversion (including hard kills between rename and create): empty fences are dropped, a non-empty fence is renamed back when the target database is missing, and a non-empty fence whose target already exists is kept and warned about on every startup so stranded tables are surfaced persistently, never silently. Adds unit tests for the recovery decision and the LIKE-escaped fence pattern, and a smoke test that simulates crash leftovers and restarts the collector to assert the empty fence is dropped and the non-empty fence survives intact. --- .../otel-collector-replicated-database.md | 4 +- packages/otel-collector/README.md | 8 + packages/otel-collector/cmd/migrate/main.go | 142 +++++++++++++++++- .../otel-collector/cmd/migrate/main_test.go | 50 ++++++ .../fence-recovery/assert_query.sql | 6 + .../fence-recovery/expected.snap | 3 + .../otel-collector/replicated-schema.bats | 25 +++ 7 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 smoke-tests/otel-collector/data/replicated-schema/fence-recovery/assert_query.sql create mode 100644 smoke-tests/otel-collector/data/replicated-schema/fence-recovery/expected.snap diff --git a/.changeset/otel-collector-replicated-database.md b/.changeset/otel-collector-replicated-database.md index 3d0ac0ffbd..06c0f39dc9 100644 --- a/.changeset/otel-collector-replicated-database.md +++ b/.changeset/otel-collector-replicated-database.md @@ -10,7 +10,9 @@ converting an empty non-Replicated database, mirroring clickhouse-operator's `enableDatabaseSync` behavior; a non-empty database is never dropped, and the conversion renames the old database aside and re-verifies emptiness before dropping it, so tables created concurrently with the check are preserved -rather than cascade-dropped). Whenever +rather than cascade-dropped; interrupted conversions roll back or are +recovered on the next startup so no database or table is ever stranded). +Whenever the target database uses the Replicated engine — regardless of the env var — table engines are rewritten to their replicated variants (`MergeTree` → `ReplicatedMergeTree`, `SummingMergeTree` → `ReplicatedSummingMergeTree`) so diff --git a/packages/otel-collector/README.md b/packages/otel-collector/README.md index 051fc07010..fb3d460f5d 100644 --- a/packages/otel-collector/README.md +++ b/packages/otel-collector/README.md @@ -159,6 +159,14 @@ metadata stored in Keeper): re-verifying it is still empty. A table created concurrently with the emptiness check is preserved in the renamed database (with a loud warning) instead of being cascade-dropped. + + The conversion is also crash-safe. If creating the Replicated database + fails after the rename (e.g. Keeper not ready yet), the rename is rolled + back so the target database is never left absent. Fence databases left + behind by an interrupted conversion are recovered on the next startup: + empty fences are dropped, a non-empty fence is renamed back when the target + database is missing, and a non-empty fence whose target already exists is + kept and warned about on every startup (never dropped). - **Non-Replicated with tables** — never dropped (that would lose data); the seed logs a warning and continues against the existing database. diff --git a/packages/otel-collector/cmd/migrate/main.go b/packages/otel-collector/cmd/migrate/main.go index 9375c8d801..2ffb33524a 100644 --- a/packages/otel-collector/cmd/migrate/main.go +++ b/packages/otel-collector/cmd/migrate/main.go @@ -463,11 +463,120 @@ func replicatedDatabaseDDL(database string) string { database, database) } +// fenceSuffix is the marker embedded in the fence name the target database is +// renamed to during the Replicated conversion. +const fenceSuffix = "_pre_replicated_" + // renamedDatabaseName returns the fence name the target database is renamed // to during the Replicated conversion. The timestamp suffix avoids collisions // with leftovers from a previously crashed conversion. func renamedDatabaseName(database string, now time.Time) string { - return fmt.Sprintf("%s_pre_replicated_%d", database, now.Unix()) + return fmt.Sprintf("%s%s%d", database, fenceSuffix, now.Unix()) +} + +// likeEscaper escapes the ClickHouse LIKE metacharacters (backslash first). +var likeEscaper = strings.NewReplacer(`\`, `\\`, `_`, `\_`, `%`, `\%`) + +// fenceDatabasePattern returns the LIKE pattern matching fence databases left +// behind by a previous (possibly crashed) Replicated conversion of database. +// The literal prefix is escaped so `_` in it does not act as a wildcard. +func fenceDatabasePattern(database string) string { + return likeEscaper.Replace(database+fenceSuffix) + "%" +} + +// listFenceDatabases returns the names of fence databases left behind by +// previous Replicated conversions of database, oldest first (the timestamp +// suffix makes lexicographic order chronological). +func listFenceDatabases(ctx context.Context, db *sql.DB, database string) ([]string, error) { + rows, err := db.QueryContext(ctx, + "SELECT name FROM system.databases WHERE name LIKE ? ORDER BY name", fenceDatabasePattern(database)) + if err != nil { + return nil, fmt.Errorf("failed to list fence databases for %q: %w", database, err) + } + defer rows.Close() + + var names []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, fmt.Errorf("failed to scan fence database name: %w", err) + } + names = append(names, name) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to list fence databases for %q: %w", database, err) + } + return names, nil +} + +// fenceRecovery describes what recoverFenceDatabases should do with a fence +// database left behind by a previous (crashed) Replicated conversion. +type fenceRecovery int + +const ( + // fenceDrop: the fence is empty; drop it (junk from a crashed conversion). + fenceDrop fenceRecovery = iota + // fenceRestore: the fence has tables and the target database is missing; + // rename the fence back so its data becomes reachable again. + fenceRestore + // fenceWarn: the fence has tables but the target database already exists; + // never drop it, warn loudly so the stranded tables stay visible. + fenceWarn +) + +// decideFenceRecovery is the pure decision function behind +// recoverFenceDatabases, factored out for testability. +func decideFenceRecovery(targetExists bool, fenceTableCount uint64) fenceRecovery { + if fenceTableCount == 0 { + return fenceDrop + } + if !targetExists { + return fenceRestore + } + return fenceWarn +} + +// recoverFenceDatabases converges fence databases left behind when a previous +// Replicated conversion crashed between the RENAME and a successful cleanup: +// - empty fence -> drop it (provably junk) +// - non-empty fence, target missing -> rename it back to the target name so +// the stranded tables become reachable again (the normal decision flow +// then keeps the non-empty database as-is) +// - non-empty fence, target exists -> keep it and warn loudly on every +// startup so stranded tables are surfaced persistently, never silently +func recoverFenceDatabases(ctx context.Context, db *sql.DB, database string) error { + fences, err := listFenceDatabases(ctx, db, database) + if err != nil { + return err + } + + for _, fence := range fences { + _, targetExists, err := getDatabaseEngine(ctx, db, database) + if err != nil { + return err + } + tableCount, err := countDatabaseTables(ctx, db, fence) + if err != nil { + return err + } + + switch decideFenceRecovery(targetExists, tableCount) { + case fenceDrop: + log.Printf("Dropping empty fence database %q left behind by a previous Replicated conversion", fence) + if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s` SYNC", fence)); err != nil { + return fmt.Errorf("failed to drop empty fence database %q: %w", fence, err) + } + case fenceRestore: + log.Printf("Restoring fence database %q with %d table(s) back to %q after an interrupted Replicated conversion", fence, tableCount, database) + if _, err := db.ExecContext(ctx, fmt.Sprintf("RENAME DATABASE `%s` TO `%s`", fence, database)); err != nil { + return fmt.Errorf("failed to restore fence database %q to %q: %w", fence, database, err) + } + case fenceWarn: + log.Printf("WARNING: Fence database %q from a previous Replicated conversion still has %d table(s) and %q already exists; it will NOT be dropped. Move any needed tables into %q and drop %q manually.", + fence, tableCount, database, database, fence) + } + } + return nil } // getDatabaseEngine queries system.databases for the target database's engine. @@ -513,7 +622,19 @@ func countDatabaseTables(ctx context.Context, db *sql.DB, database string) (uint // the original name, and the renamed database is dropped only after // re-verifying it is still empty. A table that raced the check therefore // survives in the renamed database instead of being destroyed. +// +// The conversion is also crash-safe: fence databases left behind by a +// previous interrupted conversion are recovered first (empty -> dropped, +// non-empty with the target missing -> renamed back, non-empty with the +// target present -> kept with a persistent warning), and a failure to create +// the Replicated database after the rename rolls the rename back so the +// target database is never left absent with data stranded under the fence +// name. func ensureReplicatedDatabase(ctx context.Context, db *sql.DB, database string) error { + if err := recoverFenceDatabases(ctx, db, database); err != nil { + return err + } + engine, exists, err := getDatabaseEngine(ctx, db, database) if err != nil { return err @@ -542,7 +663,24 @@ func ensureReplicatedDatabase(ctx context.Context, db *sql.DB, database string) return fmt.Errorf("failed to rename database %q to %q: %w", database, renamed, err) } if _, err := db.ExecContext(ctx, replicatedDatabaseDDL(database)); err != nil { - return fmt.Errorf("failed to create Replicated database %q: %w", database, err) + createErr := fmt.Errorf("failed to create Replicated database %q: %w", database, err) + // Roll the rename back (best-effort) so the target database is + // not left absent — and no tables stranded under the fence name — + // while the seed fails and the container restarts. The rename is + // metadata-local, so it succeeds even when the CREATE failed + // because Keeper is unavailable. If the target reappeared in the + // meantime (e.g. clickhouse-operator created it), keep the fence; + // recoverFenceDatabases surfaces it on the next startup. + if _, targetExists, checkErr := getDatabaseEngine(ctx, db, database); checkErr != nil { + log.Printf("WARNING: Failed to check database %q while rolling back the Replicated conversion: %v. Database may be preserved as %q.", database, checkErr, renamed) + } else if targetExists { + log.Printf("WARNING: Database %q was recreated concurrently; the original database is preserved as %q.", database, renamed) + } else if _, renameErr := db.ExecContext(ctx, fmt.Sprintf("RENAME DATABASE `%s` TO `%s`", renamed, database)); renameErr != nil { + log.Printf("WARNING: Failed to roll back rename of %q to %q: %v. Database is preserved as %q.", renamed, database, renameErr, renamed) + } else { + log.Printf("Rolled back rename: restored %q from %q after failed Replicated conversion", database, renamed) + } + return createErr } // The renamed database is fenced off from name-based writers, so this // re-check is stable: drop it only if it is provably still empty. diff --git a/packages/otel-collector/cmd/migrate/main_test.go b/packages/otel-collector/cmd/migrate/main_test.go index 0f2f953e59..7da65beb5b 100644 --- a/packages/otel-collector/cmd/migrate/main_test.go +++ b/packages/otel-collector/cmd/migrate/main_test.go @@ -1087,6 +1087,56 @@ func TestRenamedDatabaseName(t *testing.T) { } } +func TestFenceDatabasePattern(t *testing.T) { + tests := []struct { + database string + want string + }{ + // Literal underscores must be escaped so they don't act as LIKE + // single-character wildcards. + {"default", `default\_pre\_replicated\_%`}, + {"otel_json", `otel\_json\_pre\_replicated\_%`}, + {"my%db", `my\%db\_pre\_replicated\_%`}, + {`my\db`, `my\\db\_pre\_replicated\_%`}, + } + for _, tt := range tests { + if got := fenceDatabasePattern(tt.database); got != tt.want { + t.Errorf("fenceDatabasePattern(%q): got %q, want %q", tt.database, got, tt.want) + } + } + + // Every name produced by renamedDatabaseName must fall inside the literal + // prefix the pattern matches. + name := renamedDatabaseName("otel_json", time.Unix(1753000000, 0)) + prefix := "otel_json" + fenceSuffix + if !strings.HasPrefix(name, prefix) { + t.Errorf("renamedDatabaseName %q does not start with fence prefix %q", name, prefix) + } +} + +func TestDecideFenceRecovery(t *testing.T) { + tests := []struct { + name string + targetExists bool + tableCount uint64 + want fenceRecovery + }{ + {"empty fence is dropped even when target exists", true, 0, fenceDrop}, + {"empty fence is dropped when target is missing", false, 0, fenceDrop}, + {"non-empty fence is restored when target is missing", false, 3, fenceRestore}, + {"non-empty fence is kept with a warning when target exists", true, 3, fenceWarn}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := decideFenceRecovery(tt.targetExists, tt.tableCount); got != tt.want { + t.Errorf("decideFenceRecovery(%v, %d): got %v, want %v", + tt.targetExists, tt.tableCount, got, tt.want) + } + }) + } +} + func TestRewriteEnginesForReplicated(t *testing.T) { t.Run("rewrites MergeTree and SummingMergeTree engines", func(t *testing.T) { dir := t.TempDir() diff --git a/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/assert_query.sql b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/assert_query.sql new file mode 100644 index 0000000000..d443f354f5 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/assert_query.sql @@ -0,0 +1,6 @@ +-- Fence recovery after an interrupted conversion: the empty fence is dropped, +-- the non-empty fence is preserved with its table (and warned about, never +-- dropped), and the target database stays Replicated. +SELECT name FROM system.databases WHERE name LIKE 'default_pre_replicated_%' ORDER BY name FORMAT CSV; +SELECT count() FROM default_pre_replicated_1000000001.stranded FORMAT CSV; +SELECT engine FROM system.databases WHERE name = 'default' FORMAT CSV; diff --git a/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/expected.snap b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/expected.snap new file mode 100644 index 0000000000..d0fac29159 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/expected.snap @@ -0,0 +1,3 @@ +"default_pre_replicated_1000000001" +1 +"Replicated" diff --git a/smoke-tests/otel-collector/replicated-schema.bats b/smoke-tests/otel-collector/replicated-schema.bats index ed3313e915..bc5dcc6d27 100644 --- a/smoke-tests/otel-collector/replicated-schema.bats +++ b/smoke-tests/otel-collector/replicated-schema.bats @@ -18,3 +18,28 @@ load 'test_helpers/assertions.bash' wait_for_rows 39000 "SELECT count() FROM otel_logs WHERE ResourceAttributes['suite-id'] = 'replicated-schema' AND ResourceAttributes['test-id'] = 'basic-insert'" 2 assert_test_data_replicated "data/replicated-schema/basic-insert" } + +@test "replicated schema should recover fence databases left by an interrupted conversion" { + # Simulate leftovers from a conversion that crashed between the RENAME and + # a successful cleanup: an empty fence (junk) and a non-empty fence + # (stranded table that raced into the database before the crash). + clickhouse-client --port=39000 --query="CREATE DATABASE IF NOT EXISTS default_pre_replicated_1000000000" + clickhouse-client --port=39000 --query="CREATE DATABASE IF NOT EXISTS default_pre_replicated_1000000001" + clickhouse-client --port=39000 --query="CREATE TABLE IF NOT EXISTS default_pre_replicated_1000000001.stranded (x UInt8) ENGINE = MergeTree ORDER BY x" + clickhouse-client --port=39000 --query="INSERT INTO default_pre_replicated_1000000001.stranded VALUES (1)" + + # Restart the collector so the schema seed's fence recovery runs again. + docker compose restart otel-collector-replicated + + # The seed drops the empty fence during startup; poll until it is gone + # (the inverted count flips to 1 once the database no longer exists). + wait_for_rows 39000 "SELECT count() == 0 FROM system.databases WHERE name = 'default_pre_replicated_1000000000'" 1 + + # The non-empty fence must survive with its table intact, and the target + # database must still be the Replicated one. + assert_test_data_replicated "data/replicated-schema/fence-recovery" + + # Clean up the preserved fence so the suite is idempotent under + # SKIP_CLEANUP re-runs. + clickhouse-client --port=39000 --query="DROP DATABASE default_pre_replicated_1000000001 SYNC" +}