diff --git a/.changeset/otel-collector-replicated-database.md b/.changeset/otel-collector-replicated-database.md new file mode 100644 index 0000000000..74a4770fec --- /dev/null +++ b/.changeset/otel-collector-replicated-database.md @@ -0,0 +1,23 @@ +--- +'@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, and the +conversion renames the old database aside and re-verifies emptiness — +including detached tables — before dropping it, so tables created +concurrently with the check are preserved rather than cascade-dropped; +interrupted conversions roll back or are recovered on the next startup so no +database or table is ever stranded; and stale conversion decisions on +concurrent replicas are detected instead of renaming away a freshly converted +database). 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. After seeding, pre-existing +MergeTree-family tables that still use a non-replicated engine are surfaced +with a loud per-table warning and remediation on every startup. diff --git a/packages/otel-collector/README.md b/packages/otel-collector/README.md index 95cf13a557..8bf8993c99 100644 --- a/packages/otel-collector/README.md +++ b/packages/otel-collector/README.md @@ -129,6 +129,86 @@ 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** — 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. + + Emptiness counts both attached and detached tables (`system.tables` + + `system.detached_tables`; dictionaries already appear in `system.tables`), + so nothing `DROP DATABASE` would destroy escapes the check. Because + `system.tables` visibility is grant-scoped, the seed must run as a user + with full visibility on the target database (the default in shipped + configurations). + + The conversion is also crash-safe and race-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). The engine is + re-read immediately before the rename and verified afterwards, so a stale + conversion decision on one collector replica never renames away a database + another replica just converted. +- **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). + +After seeding, the seed audits the Replicated database for MergeTree-family +tables that still use a non-replicated engine (e.g. tables created before the +conversion, which no-op through `CREATE TABLE IF NOT EXISTS`). It cannot +convert them itself — that requires the server-side `convert_to_replicated` +marker file and a restart — so it logs a loud per-table warning with the +remediation on every startup instead of claiming unqualified success. The +seed still exits 0: the deployment keeps working single-replica, and failing +would crash-loop ingestion on a condition the seed cannot repair. + +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..49819b4ad8 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,492 @@ 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; 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. + 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) +} + +// 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%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. +// 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 attached plus detached tables in +// the target database. This is the emptiness predicate gating the irreversible +// DROP DATABASE during the Replicated conversion, so it must count everything +// a drop would destroy: detached tables are invisible in system.tables but +// their on-disk data is still deleted by DROP DATABASE (system.tables also +// already includes dictionaries, with engine "Dictionary"). It errors when the +// database does not exist, so a missing database can never read as "empty". +func countDatabaseTables(ctx context.Context, db *sql.DB, database string) (uint64, error) { + var exists uint8 + var attached, detached uint64 + err := db.QueryRowContext(ctx, `SELECT + (SELECT count() FROM system.databases WHERE name = ?) > 0, + (SELECT count() FROM system.tables WHERE database = ?), + (SELECT count() FROM system.detached_tables WHERE database = ?)`, + database, database, database).Scan(&exists, &attached, &detached) + if err != nil { + return 0, fmt.Errorf("failed to count tables in database %q: %w", database, err) + } + if exists == 0 { + return 0, fmt.Errorf("cannot count tables: database %q does not exist", database) + } + return attached + detached, 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 -> 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. +// +// The conversion is also crash-safe and race-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); 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; and the conversion re-reads the engine immediately before renaming +// (and verifies the fenced database afterwards) so a stale convert decision +// never renames away a database another replica just converted. See +// convertDatabaseToReplicated. +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 + } + + 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: + return convertDatabaseToReplicated(ctx, db, database) + 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 +} + +// convertDatabaseToReplicated converts an (empty, non-Replicated) target +// database to the Replicated engine via rename-fence-verify-drop: +// +// 1. Re-read the engine immediately before acting: the dbActionConvert +// decision may be stale (another collector replica or the operator may +// have converted the database in the meantime), and renaming away a +// freshly converted Replicated database would swap out the active +// database under its writers. +// 2. Atomically RENAME the old database aside, fencing name-based writers. +// 3. Verify what was fenced really is non-Replicated; if a racer's +// conversion slipped between the re-read and the rename, undo the rename. +// 4. Create the Replicated database under the original name; on failure, +// roll the rename back so the target is never left absent. +// 5. Drop the fence only after re-verifying it is provably empty. +func convertDatabaseToReplicated(ctx context.Context, db *sql.DB, database string) error { + engine, exists, err := getDatabaseEngine(ctx, db, database) + if err != nil { + return err + } + if exists && engine == replicatedEngineName { + log.Printf("Database %q was converted to the Replicated engine concurrently; nothing to do", database) + return nil + } + if !exists { + // Another actor is mid-conversion (renamed the database aside but has + // not recreated it yet). CREATE IF NOT EXISTS is concurrency-safe. + log.Printf("Database %q disappeared concurrently; 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 + } + + 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) + } + + // Verify what was fenced: if a racer converted the database between the + // re-read above and the rename, the fence now holds their Replicated + // database. Undo the rename and keep the converted database. (On most + // ClickHouse versions renaming a Replicated database fails outright, in + // which case the rename error above already aborted the conversion; this + // check does not rely on that behavior.) + fencedEngine, fencedExists, err := getDatabaseEngine(ctx, db, renamed) + if err != nil { + return err + } + if fencedExists && fencedEngine == replicatedEngineName { + log.Printf("Renamed database %q already uses the Replicated engine (converted concurrently); renaming it back to %q", renamed, database) + if _, err := db.ExecContext(ctx, fmt.Sprintf("RENAME DATABASE `%s` TO `%s`", renamed, database)); err != nil { + return fmt.Errorf("failed to rename Replicated database %q back to %q: %w", renamed, database, err) + } + return nil + } + + if _, err := db.ExecContext(ctx, replicatedDatabaseDDL(database)); err != nil { + 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 + // (including detached tables, whose data DROP DATABASE would destroy). + 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 +} + +// 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 + }) +} + +// isUnreplicatedMergeTreeEngine reports whether a table engine is a +// MergeTree-family engine whose data does not replicate across replicas +// (e.g. MergeTree, SummingMergeTree), as opposed to Replicated*/Shared* +// variants and non-MergeTree engines (MaterializedView, Dictionary, +// TimeSeries, Distributed, ...). +func isUnreplicatedMergeTreeEngine(engine string) bool { + return strings.HasSuffix(engine, "MergeTree") && + !strings.HasPrefix(engine, "Replicated") && + !strings.HasPrefix(engine, "Shared") +} + +// tableEngine pairs a table name with its engine for the post-seed audit. +type tableEngine struct { + Name string + Engine string +} + +// listUnreplicatedTables returns the MergeTree-family tables in the target +// database whose engine is not a Replicated variant. In a Replicated database +// such tables no-op through the seed's CREATE TABLE IF NOT EXISTS statements +// (e.g. tables created before the conversion by the legacy exporter path), so +// their data silently does not replicate; the caller surfaces them loudly. +func listUnreplicatedTables(ctx context.Context, db *sql.DB, database string) ([]tableEngine, error) { + rows, err := db.QueryContext(ctx, + "SELECT name, engine FROM system.tables WHERE database = ? ORDER BY name", database) + if err != nil { + return nil, fmt.Errorf("failed to list tables in database %q: %w", database, err) + } + defer rows.Close() + + var unreplicated []tableEngine + for rows.Next() { + var t tableEngine + if err := rows.Scan(&t.Name, &t.Engine); err != nil { + return nil, fmt.Errorf("failed to scan table engine: %w", err) + } + if isUnreplicatedMergeTreeEngine(t.Engine) { + unreplicated = append(unreplicated, t) + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to list tables in database %q: %w", database, err) + } + return unreplicated, nil +} + +// auditReplicatedTables warns, loudly and per table, about MergeTree-family +// tables in the Replicated target database that do not use a Replicated +// engine. The seed cannot convert them itself (MergeTree -> +// ReplicatedMergeTree conversion requires the server-side convert_to_replicated +// marker file and a restart), so it reports them on every startup instead of +// claiming unqualified success. Audit failures only warn: this is an +// observability query and must not fail an otherwise successful seed. +func auditReplicatedTables(ctx context.Context, db *sql.DB, database string) { + unreplicated, err := listUnreplicatedTables(ctx, db, database) + if err != nil { + log.Printf("WARNING: Failed to audit table engines in Replicated database %q: %v", database, err) + return + } + for _, t := range unreplicated { + log.Printf("WARNING: Table %q.%q uses the non-replicated %s engine inside a Replicated database; its data will NOT replicate across replicas. Convert it manually via the convert_to_replicated marker file (https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/replication#converting-from-mergetree-to-replicatedmergetree).", + database, t.Name, t.Engine) + } + if len(unreplicated) > 0 { + log.Printf("WARNING: Schema seed completed with %d non-replicated table(s) in Replicated database %q", len(unreplicated), database) + } +} + // 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 +1012,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 +1071,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 { @@ -563,6 +1099,14 @@ func main() { os.Exit(1) } + // Audit the end state: pre-existing plain MergeTree tables in a + // Replicated database no-op through CREATE TABLE IF NOT EXISTS above and + // would otherwise stay silently unreplicated. The seed cannot fix them, + // so it reports them loudly instead of claiming unqualified success. + if targetIsReplicated { + auditReplicatedTables(ctx, db, cfg.Database) + } + log.Println("========================================") log.Println("Schema seed completed successfully") log.Println("========================================") diff --git a/packages/otel-collector/cmd/migrate/main_test.go b/packages/otel-collector/cmd/migrate/main_test.go index 023bf77f85..173cf47381 100644 --- a/packages/otel-collector/cmd/migrate/main_test.go +++ b/packages/otel-collector/cmd/migrate/main_test.go @@ -966,3 +966,329 @@ 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 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 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 TestIsUnreplicatedMergeTreeEngine(t *testing.T) { + tests := map[string]bool{ + // Plain MergeTree-family engines: data does not replicate. + "MergeTree": true, + "SummingMergeTree": true, + "ReplacingMergeTree": true, + "AggregatingMergeTree": true, + // Replicated variants replicate data. + "ReplicatedMergeTree": false, + "ReplicatedSummingMergeTree": false, + // SharedMergeTree (ClickHouse Cloud) manages replication itself. + "SharedMergeTree": false, + // Non-MergeTree engines are out of scope for the audit. + "MaterializedView": false, + "Dictionary": false, + "TimeSeries": false, + "Distributed": false, + "View": false, + } + for engine, want := range tests { + if got := isUnreplicatedMergeTreeEngine(engine); got != want { + t.Errorf("isUnreplicatedMergeTreeEngine(%q): got %v, want %v", engine, got, want) + } + } +} + +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() + 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..247246966b --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/engines/assert_query.sql @@ -0,0 +1,16 @@ +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' + 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/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..0cb9830113 --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/assert_query.sql @@ -0,0 +1,8 @@ +-- Fence recovery after an interrupted conversion: the empty fence is dropped, +-- the non-empty fences are preserved (attached table with its data, detached +-- table still detached) 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 count() FROM system.detached_tables WHERE database = 'default_pre_replicated_1000000002' AND table = 'stranded_detached' 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..3326625b2f --- /dev/null +++ b/smoke-tests/otel-collector/data/replicated-schema/fence-recovery/expected.snap @@ -0,0 +1,5 @@ +"default_pre_replicated_1000000001" +"default_pre_replicated_1000000002" +1 +1 +"Replicated" 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..14a6a5f7bd --- /dev/null +++ b/smoke-tests/otel-collector/replicated-schema.bats @@ -0,0 +1,82 @@ +#!/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" +} + +@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), a non-empty fence (stranded + # table that raced into the database before the crash), and a fence whose + # only table is detached (invisible in system.tables but still destroyed + # by DROP DATABASE, so it must count against the emptiness check). + 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)" + clickhouse-client --port=39000 --query="CREATE DATABASE IF NOT EXISTS default_pre_replicated_1000000002" + clickhouse-client --port=39000 --query="CREATE TABLE IF NOT EXISTS default_pre_replicated_1000000002.stranded_detached (x UInt8) ENGINE = MergeTree ORDER BY x" + clickhouse-client --port=39000 --query="DETACH TABLE default_pre_replicated_1000000002.stranded_detached" + + # 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 fences must survive (attached table intact, detached + # table still detached), and the target database must still be the + # Replicated one. + assert_test_data_replicated "data/replicated-schema/fence-recovery" + + # Clean up the preserved fences so the suite is idempotent under + # SKIP_CLEANUP re-runs. + clickhouse-client --port=39000 --query="DROP DATABASE default_pre_replicated_1000000001 SYNC" + clickhouse-client --port=39000 --query="DROP DATABASE default_pre_replicated_1000000002 SYNC" +} + +@test "replicated schema should warn about pre-existing non-replicated tables" { + # A plain MergeTree table inside the Replicated database (e.g. created + # before the conversion by another schema manager) no-ops through the + # seed's CREATE TABLE IF NOT EXISTS, so its data would silently not + # replicate. The post-seed audit must call it out on every startup. + clickhouse-client --port=39000 --query="CREATE TABLE IF NOT EXISTS default.legacy_plain (x UInt8) ENGINE = MergeTree ORDER BY x" + clickhouse-client --port=39000 --query="INSERT INTO default.legacy_plain VALUES (1)" + + docker compose restart otel-collector-replicated + + # Wait for the restarted seed to finish and emit the audit warning. + local attempt=0 + until docker compose logs otel-collector-replicated | grep -q 'uses the non-replicated MergeTree engine inside a Replicated database'; do + attempt=$((attempt + 1)) + if [ "$attempt" -gt 30 ]; then + echo "❌ Error: audit warning for non-replicated table not found in collector logs" >&3 + return 1 + fi + sleep 1 + done + + # The audited table must be untouched (never dropped or altered). + run clickhouse-client --port=39000 --query="SELECT engine, (SELECT count() FROM default.legacy_plain) FROM system.tables WHERE database = 'default' AND name = 'legacy_plain' FORMAT CSV" + [ "$status" -eq 0 ] + [ "$output" = '"MergeTree",1' ] + + # Clean up so the suite is idempotent under SKIP_CLEANUP re-runs. + clickhouse-client --port=39000 --query="DROP TABLE default.legacy_plain SYNC" +} 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