diff --git a/scripts/clickhouse-upgrade-test/Makefile b/scripts/clickhouse-upgrade-test/Makefile new file mode 100644 index 00000000..67e4927d --- /dev/null +++ b/scripts/clickhouse-upgrade-test/Makefile @@ -0,0 +1,21 @@ +.PHONY: test test-staged test-direct config down clean + +# Full test: staged (recommended) upgrade path, then the naive direct-jump path +test: + python3 run_test.py --scenario both + +test-staged: + python3 run_test.py --scenario staged + +test-direct: + python3 run_test.py --scenario direct + +# Validate docker-compose.yml without needing registry access +config: + docker compose config + +down: + docker compose down -v + +clean: down + rm -rf results/report.md results/report.json diff --git a/scripts/clickhouse-upgrade-test/README.md b/scripts/clickhouse-upgrade-test/README.md new file mode 100644 index 00000000..6374b9ba --- /dev/null +++ b/scripts/clickhouse-upgrade-test/README.md @@ -0,0 +1,180 @@ +# ClickHouse cluster upgrade test (ooni/devops#437) + +Answers the question behind [ooni/devops#437](https://github.com/ooni/devops/issues/437): +**can OONI's production ClickHouse cluster be upgraded from its current +version to the latest stable release one node at a time, or does it need a +scheduled-downtime, all-nodes-at-once upgrade?** + +## TL;DR + +- Production is on **24.8.6.70** (LTS, Aug 2024) — confirmed from + `ooni/devops` `ansible/group_vars/clickhouse/vars.yml` (`clickhouse_version: 24.8.6.70`), + matching what issue #437 reports. +- Latest stable as of 2026-08-10 is **26.7.3.19** (released 2026-07-22). +- That's about **23 months apart**. ClickHouse's own docs + ([clickhouse.com/docs/operations/update](https://clickhouse.com/docs/operations/update)) + say replicas of the same shard should not run versions more than + **~1 year apart** — beyond that window the docs warn the cluster "may not + work", queries can fail with arbitrary errors, and downgrading stops being + an option. +- **Recommendation: do a rolling, node-by-node upgrade, but stage it through + each intermediate LTS release rather than jumping straight to latest.** + No full-cluster downtime is needed either way — the risk isn't downtime, + it's version skew during the upgrade window. + + ``` + 24.8.6.70 → 25.3.14.14 → 25.8.29.51 → 26.3.17.110 → 26.7.3.19 + (current) LTS LTS LTS (latest stable) + ``` + + Each hop is 4–7 months of releases apart, comfortably inside the + compatibility window. Do all 3 replicas one at a time for a given hop + before starting the next hop (never skip ahead on one node while another + is still 2+ hops behind). + +This repo contains a dockerized test that *exercises* this rather than just +asserting it: it spins up a 3-node cluster shaped exactly like OONI's +`oonidata_cluster` (1 shard, 3 replicas, embedded ClickHouse Keeper, same +table schemas), loads it with data, and mechanically upgrades one node at a +time — first via the direct jump (to show what breaks), then via the staged +LTS path (to confirm it doesn't). + +## Where the numbers come from + +| Fact | Source | +|---|---| +| Current version `24.8.6.70` | `ooni/devops` `ansible/group_vars/clickhouse/vars.yml` → `clickhouse_version:` | +| Cluster topology: 1 shard, 3 replicas, embedded Keeper on `data1/2/3.htz-fsn.prod.ooni.nu` | `ooni/devops` `ansible/group_vars/clickhouse/vars.yml` (`clickhouse_remote_servers`, `clickhouse_keeper`, `clickhouse_macros`), `ansible/roles/oonidata_clickhouse/tasks/main.yml`, `ansible/inventory` | +| Production table schemas (`fastpath`, `citizenlab`, `jsonl`, `analysis_web_measurement`, `event_detector_changepoints`, `faulty_measurements`) | `ooni/devops` `scripts/cluster-migration/schema.sql` | +| `obs_web` column list | `ooni/backend` `ooniapi/services/oonimeasurements/tests/fixtures/initdb/clickhouse.sql` | +| Other table column lists (test/CI copies) | `ooni/backend` `ooniapi/services/oonimeasurements/tests/migrations/0_clickhouse_init_tables.sql` | +| Latest stable / LTS version history | [clickhouse.com/docs/whats-new/changelog](https://clickhouse.com/docs/whats-new/changelog), [endoflife.date/clickhouse](https://endoflife.date/clickhouse) | +| Mixed-version / rolling-upgrade guidance | [clickhouse.com/docs/operations/update](https://clickhouse.com/docs/operations/update) | + +## What the test actually does + +`docker-compose.yml` brings up 3 ClickHouse nodes (`ch1`, `ch2`, `ch3`) on a +private docker network, each running **both** `clickhouse-server` and an +embedded **ClickHouse Keeper** instance (ports 9181/9234) — the same +topology as `data1/data2/data3` in production, just condensed onto one +Docker host. `sql/001_schema.sql` creates the real table schemas +(`ReplicatedReplacingMergeTree`, `ON CLUSTER oonidata_cluster`) and +`harness/seed_data.py` loads synthetic-but-schema-accurate rows into them. + +`run_test.py` then runs one or both scenarios: + +- **`staged`** — walks the version ladder above, upgrading `ch1`, then + `ch2`, then `ch3` at each hop (never more than one node down at a time, + never all 3 nodes on different versions at once), validating after every + single node swap that: + - the node comes back up, + - a write issued anywhere is readable from every replica within the + timeout (`harness/validate.py:probe_write_then_read`), + - row counts converge across all 3 nodes, + - `system.errors` hasn't accumulated any replication/checksum/protocol + errors, + - `system.replication_queue` has no stuck tasks, + - and, once a hop is fully rolled out, an `ALTER TABLE ... ON CLUSTER` + still propagates cluster-wide. +- **`direct`** — does the same node-by-node mechanics but jumps straight + from `24.8.6.70` to `26.7.3.19`, to surface (not just cite) whatever + breaks when replicas are held ~2 years apart in version for the whole + rollout. + +Results land in `results/report.md` (human-readable) and +`results/report.json` (full structured data, including every row-count +snapshot and every error ClickHouse logged). + +## Running it + +Requires Docker + Compose v2, and — this matters — **network access to pull +`clickhouse/clickhouse-server` images from Docker Hub**. (This harness was +built inside a sandboxed environment whose egress is restricted to a small +allowlist that does not include Docker Hub or S3, so it could not be +executed end-to-end there; everything here was validated as far as that +constraint allows — see "What was and wasn't verified" below.) + +```bash +# from this directory +make test # both scenarios (staged, then direct), ~20-40 min depending on image pull speed +make test-staged # just the recommended path +make test-direct # just the naive direct-jump path +make config # sanity-check docker-compose.yml without pulling anything +``` + +Or directly: + +```bash +python3 run_test.py --scenario both +``` + +Add `--keep-up` to leave the cluster running after the test so you can poke +at it manually (`docker compose exec ch1 clickhouse-client`). + +## About the seed data + +The task pointed at `ooni/backend`'s initdb sample data. That repo doesn't +actually vendor the sample rows in git — its test fixtures +(`ooniapi/services/oonimeasurements/tests/conftest.py`) download +`obs_web-sample.sql.gz` and `analysis_web_measurement-sample.sql.gz` at test +time from a public S3 bucket +(`ooni-data-eu-fra.s3.eu-central-1.amazonaws.com`). This sandbox's network +egress couldn't reach S3 either, so `harness/seed_data.py` generates +synthetic rows that conform exactly to the real schemas instead (same +columns, types, nullability, realistic cardinality for things like +`probe_cc`/ASN/test names). That's sufficient for what this test is +checking — replication and on-disk part-format compatibility across +ClickHouse versions — since that behavior depends on schema and volume, not +on the specific measurement content. + +If you have S3 access and want to use the real dump instead: + +```bash +curl -sL https://ooni-data-eu-fra.s3.eu-central-1.amazonaws.com/samples/obs_web-sample.sql.gz \ + | gunzip -c | docker compose exec -T ch1 clickhouse-client --database ooni +curl -sL https://ooni-data-eu-fra.s3.eu-central-1.amazonaws.com/samples/analysis_web_measurement-sample.sql.gz \ + | gunzip -c | docker compose exec -T ch1 clickhouse-client --database ooni +``` + +(after `sql/001_schema.sql` has been applied, and before running an upgrade +scenario — or just skip the seed step in `harness/scenarios.py:load_schema_and_seed` +and load these instead). + +## What was and wasn't verified in this sandbox + +Verified here: +- `docker-compose.yml` parses and interpolates correctly (`docker compose config`). +- All ClickHouse XML config files (`config/**/*.xml`) are well-formed. +- All Python modules compile and the seed-data generator runs and produces + well-formed `INSERT` statements against the real column lists. +- The Docker daemon itself works in this sandbox (`docker run` succeeds for + locally available images). + +Not verified here (blocked by sandbox network policy — Docker Hub and S3 +are both unreachable; `docker pull` fails with `403 Forbidden` regardless of +image): +- Actually pulling the `clickhouse/clickhouse-server` images. +- Running the containers and confirming the Keeper ensemble forms, the + replicated tables replicate, and the upgrade steps behave as designed. + +**You'll need to run `make test` yourself in an environment with normal +internet access** (a dev laptop, a CI runner, an EC2 box) to get the actual +report. Budget ~10-20 minutes to pull 5 different `clickhouse-server` image +tags the first time; subsequent runs reuse the Docker image cache. + +## Files + +``` +docker-compose.yml 3-node cluster definition, per-node image tag override via env +config/common/ Settings shared by all nodes (remote_servers, zookeeper client, distributed_ddl) +config/ch{1,2,3}/node.xml Per-node macros (shard/replica) + embedded Keeper raft config +sql/001_schema.sql Production table DDL (ReplicatedReplacingMergeTree, ON CLUSTER) +harness/seed_data.py Synthetic data generator (see note above on why it's synthetic) +harness/ch_http.py Minimal stdlib-only ClickHouse HTTP client +harness/compose.py docker-compose wrapper (bring up/tear down/recreate one node at a time) +harness/validate.py Cluster health checks (replication convergence, error scraping, write/read probes) +harness/scenarios.py The two upgrade scenarios +harness/report.py Results -> Markdown report renderer +run_test.py CLI entry point +results/ report.md / report.json land here after a run +``` diff --git a/scripts/clickhouse-upgrade-test/config/ch1/node.xml b/scripts/clickhouse-upgrade-test/config/ch1/node.xml new file mode 100644 index 00000000..403c07cb --- /dev/null +++ b/scripts/clickhouse-upgrade-test/config/ch1/node.xml @@ -0,0 +1,40 @@ + + + + 01 + 01 + oonidata_cluster + + + + + 9181 + 1 + /var/lib/clickhouse/coordination/log + /var/lib/clickhouse/coordination/snapshots + + + 10000 + 30000 + information + + + + + 1 + ch1 + 9234 + + + 2 + ch2 + 9234 + + + 3 + ch3 + 9234 + + + + diff --git a/scripts/clickhouse-upgrade-test/config/ch2/node.xml b/scripts/clickhouse-upgrade-test/config/ch2/node.xml new file mode 100644 index 00000000..5a7379f6 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/config/ch2/node.xml @@ -0,0 +1,40 @@ + + + + 01 + 02 + oonidata_cluster + + + + + 9181 + 2 + /var/lib/clickhouse/coordination/log + /var/lib/clickhouse/coordination/snapshots + + + 10000 + 30000 + information + + + + + 1 + ch1 + 9234 + + + 2 + ch2 + 9234 + + + 3 + ch3 + 9234 + + + + diff --git a/scripts/clickhouse-upgrade-test/config/ch3/node.xml b/scripts/clickhouse-upgrade-test/config/ch3/node.xml new file mode 100644 index 00000000..8db521b6 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/config/ch3/node.xml @@ -0,0 +1,40 @@ + + + + 01 + 03 + oonidata_cluster + + + + + 9181 + 3 + /var/lib/clickhouse/coordination/log + /var/lib/clickhouse/coordination/snapshots + + + 10000 + 30000 + information + + + + + 1 + ch1 + 9234 + + + 2 + ch2 + 9234 + + + 3 + ch3 + 9234 + + + + diff --git a/scripts/clickhouse-upgrade-test/config/common/common.xml b/scripts/clickhouse-upgrade-test/config/common/common.xml new file mode 100644 index 00000000..68ac1210 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/config/common/common.xml @@ -0,0 +1,64 @@ + + + + 0.0.0.0 + :: + 1 + + + information + 1 + + + + + + ch1 + 9181 + + + ch2 + 9181 + + + ch3 + 9181 + + + + + + /clickhouse/task_queue/ddl + default + 1 + 604800 + 60 + 1000 + + + + + + true + + ch1 + 9000 + + + ch2 + 9000 + + + ch3 + 9000 + + + + + diff --git a/scripts/clickhouse-upgrade-test/config/common/users.xml b/scripts/clickhouse-upgrade-test/config/common/users.xml new file mode 100644 index 00000000..30167c3d --- /dev/null +++ b/scripts/clickhouse-upgrade-test/config/common/users.xml @@ -0,0 +1,37 @@ + + + + + + 2000000000 + 0 + random + + + + + + + + ::/0 + + default + default + 1 + + + + + + + 3600 + 0 + 0 + 0 + 0 + 0 + + + + diff --git a/scripts/clickhouse-upgrade-test/docker-compose.yml b/scripts/clickhouse-upgrade-test/docker-compose.yml new file mode 100644 index 00000000..d6c26c60 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/docker-compose.yml @@ -0,0 +1,82 @@ + +# 3-node ClickHouse cluster mirroring OONI's production "oonidata_cluster" +# (1 shard / 3 replicas, embedded ClickHouse Keeper on each node -- see +# ooni/devops ansible/group_vars/clickhouse/vars.yml). +# +# Each node's image tag is controlled independently via CH1_IMAGE / CH2_IMAGE +# / CH3_IMAGE so the test harness can upgrade one node at a time while the +# other two keep running on their previous version (a real rolling upgrade), +# instead of all nodes always tracking one shared version. +# +# Data directories are named volumes that persist across `docker compose up +# --force-recreate`, so swapping a node's image tag is a faithful simulation +# of an in-place binary upgrade on a real host (same on-disk data, new server +# binary). + +x-clickhouse-common: &clickhouse-common + ulimits: + nofile: + soft: 262144 + hard: 262144 + networks: + - chnet + healthcheck: + # clickhouse-client is guaranteed to be in the image regardless of version; + # wget/curl are not always present across the version range this test spans. + test: ["CMD", "clickhouse-client", "--query", "SELECT 1"] + interval: 5s + timeout: 5s + retries: 20 + start_period: 20s + +services: + ch1: + <<: *clickhouse-common + image: "clickhouse/clickhouse-server:${CH1_IMAGE:-24.8.6.70}" + container_name: ooni-ch1 + hostname: ch1 + ports: + - "${CH1_HTTP_PORT:-8123}:8123" + - "${CH1_NATIVE_PORT:-9000}:9000" + volumes: + - ch1_data:/var/lib/clickhouse + - ./config/common/common.xml:/etc/clickhouse-server/config.d/common.xml:ro + - ./config/common/users.xml:/etc/clickhouse-server/users.d/users.xml:ro + - ./config/ch1/node.xml:/etc/clickhouse-server/config.d/node.xml:ro + + ch2: + <<: *clickhouse-common + image: "clickhouse/clickhouse-server:${CH2_IMAGE:-24.8.6.70}" + container_name: ooni-ch2 + hostname: ch2 + ports: + - "${CH2_HTTP_PORT:-8124}:8123" + - "${CH2_NATIVE_PORT:-9001}:9000" + volumes: + - ch2_data:/var/lib/clickhouse + - ./config/common/common.xml:/etc/clickhouse-server/config.d/common.xml:ro + - ./config/common/users.xml:/etc/clickhouse-server/users.d/users.xml:ro + - ./config/ch2/node.xml:/etc/clickhouse-server/config.d/node.xml:ro + + ch3: + <<: *clickhouse-common + image: "clickhouse/clickhouse-server:${CH3_IMAGE:-24.8.6.70}" + container_name: ooni-ch3 + hostname: ch3 + ports: + - "${CH3_HTTP_PORT:-8125}:8123" + - "${CH3_NATIVE_PORT:-9002}:9000" + volumes: + - ch3_data:/var/lib/clickhouse + - ./config/common/common.xml:/etc/clickhouse-server/config.d/common.xml:ro + - ./config/common/users.xml:/etc/clickhouse-server/users.d/users.xml:ro + - ./config/ch3/node.xml:/etc/clickhouse-server/config.d/node.xml:ro + +networks: + chnet: + name: ooni-chnet + +volumes: + ch1_data: + ch2_data: + ch3_data: diff --git a/scripts/clickhouse-upgrade-test/harness/__init__.py b/scripts/clickhouse-upgrade-test/harness/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scripts/clickhouse-upgrade-test/harness/ch_http.py b/scripts/clickhouse-upgrade-test/harness/ch_http.py new file mode 100644 index 00000000..2e8beb60 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/ch_http.py @@ -0,0 +1,77 @@ +""" +Minimal ClickHouse HTTP-interface client. + +Deliberately dependency-free (stdlib only: urllib) so the test harness needs +nothing beyond a Python 3 interpreter and Docker to run. +""" +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from dataclasses import dataclass + + +class ClickHouseError(RuntimeError): + def __init__(self, query: str, status: int, body: str): + self.query = query + self.status = status + self.body = body + super().__init__(f"ClickHouse query failed (HTTP {status}): {body.strip()}\n--- query ---\n{query}") + + +@dataclass +class ChNode: + name: str # e.g. "ch1" -- must match docker-compose service/hostname + http_port: int # host-mapped port for the HTTP interface (8123 default) + host: str = "127.0.0.1" + + @property + def base_url(self) -> str: + return f"http://{self.host}:{self.http_port}/" + + def query(self, sql: str, timeout: float = 30.0, fmt: str | None = "JSONEachRow") -> str: + """Execute a query and return the raw response body (text).""" + q = sql + if fmt and sql.strip().lower().startswith("select"): + q = f"{sql}\nFORMAT {fmt}" + data = q.encode("utf-8") + req = urllib.request.Request(self.base_url, data=data, method="POST") + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise ClickHouseError(sql, e.code, body) from None + except urllib.error.URLError as e: + raise ClickHouseError(sql, -1, str(e.reason)) from None + + def query_rows(self, sql: str, timeout: float = 30.0) -> list[dict]: + body = self.query(sql, timeout=timeout, fmt="JSONEachRow") + rows = [] + for line in body.splitlines(): + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + def query_scalar(self, sql: str, timeout: float = 30.0): + rows = self.query_rows(sql, timeout=timeout) + if not rows: + return None + return next(iter(rows[0].values())) + + def execute(self, sql: str, timeout: float = 60.0) -> None: + """Run a statement where we don't care about the result body (DDL, INSERT).""" + self.query(sql, timeout=timeout, fmt=None) + + def ping(self, timeout: float = 3.0) -> bool: + try: + req = urllib.request.Request(self.base_url + "ping", method="GET") + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status == 200 + except Exception: + return False + + def version(self) -> str: + return self.query_scalar("SELECT version()") diff --git a/scripts/clickhouse-upgrade-test/harness/compose.py b/scripts/clickhouse-upgrade-test/harness/compose.py new file mode 100644 index 00000000..8e7a564b --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/compose.py @@ -0,0 +1,79 @@ +""" +Thin wrapper around `docker compose` so the rest of the harness doesn't +shell out directly. All commands run relative to PROJECT_DIR. +""" +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + +PROJECT_DIR = Path(__file__).resolve().parent.parent + + +def _run(args: list[str], env: dict | None = None, check: bool = True) -> subprocess.CompletedProcess: + full_env = os.environ.copy() + if env: + full_env.update(env) + proc = subprocess.run( + ["docker", "compose", *args], + cwd=PROJECT_DIR, + env=full_env, + capture_output=True, + text=True, + ) + if check and proc.returncode != 0: + raise RuntimeError( + f"docker compose {' '.join(args)} failed (rc={proc.returncode})\n" + f"stdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" + ) + return proc + + +def up(services: list[str] | None = None, env: dict | None = None, force_recreate: bool = False) -> None: + args = ["up", "-d"] + if force_recreate: + args.append("--force-recreate") + if services: + args += ["--no-deps", *services] + _run(args, env=env) + + +def down(volumes: bool = True) -> None: + args = ["down"] + if volumes: + args.append("-v") + _run(args, check=False) + + +def logs(service: str, tail: int = 200) -> str: + proc = _run(["logs", f"--tail={tail}", service], check=False) + return proc.stdout + proc.stderr + + +def ps() -> str: + proc = _run(["ps"], check=False) + return proc.stdout + + +def upgrade_node(service: str, new_image_tag: str, current_env: dict) -> dict: + """ + Recreate a single node with a new image tag while leaving the other + nodes running untouched. `current_env` carries the *other* nodes' image + pins forward (compose interpolates ${CH1_IMAGE} etc. from the process + env / .env file at `up` time, so we must always pass the full set). + + Returns the updated env dict (with this service's image tag changed) so + callers can thread it through subsequent calls. + """ + var_name = f"{service.upper()}_IMAGE" + new_env = dict(current_env) + new_env[var_name] = new_image_tag + up(services=[service], env=new_env, force_recreate=True) + return new_env + + +def config_check() -> str: + """Validate compose file syntax/interpolation without contacting a registry.""" + proc = _run(["config"], check=True) + return proc.stdout diff --git a/scripts/clickhouse-upgrade-test/harness/report.py b/scripts/clickhouse-upgrade-test/harness/report.py new file mode 100644 index 00000000..2a67087c --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/report.py @@ -0,0 +1,77 @@ +"""Turns the structured results dict produced by scenarios.py into a +human-readable Markdown report.""" +from __future__ import annotations + +import json +from pathlib import Path + + +def _fmt_bool(b) -> str: + return "PASS" if b else "FAIL" + + +def render_step(step: dict) -> str: + lines = [] + lines.append(f"### Step: {step['label']}") + lines.append("") + lines.append(f"- Node upgraded: `{step.get('node', '-')}`") + lines.append(f"- New version: `{step.get('new_version', '-')}`") + lines.append(f"- Versions across cluster after this step: `{step.get('versions')}`") + lines.append(f"- Rolling (never all-3-down): **{_fmt_bool(step.get('all_up_throughout'))}**") + lines.append(f"- Row counts converged across nodes: **{_fmt_bool(step.get('converged'))}**") + probe = step.get("probe", {}) + lines.append( + f"- Write-then-read-back probe fully replicated: **{_fmt_bool(probe.get('fully_replicated'))}** " + f"(probe id `{probe.get('probe_id')}`)" + ) + errs = step.get("errors_found", {}) + any_errs = any(v for v in errs.values()) + lines.append(f"- Replication-related errors logged by ClickHouse: **{'YES' if any_errs else 'none'}**") + if any_errs: + for node, errlist in errs.items(): + if errlist: + lines.append(f" - `{node}`:") + for e in errlist[:5]: + lines.append(f" - `{e.get('name')}`: {e.get('last_error_message', '')[:200]}") + qprob = step.get("queue_problems", {}) + any_q = any(v for v in qprob.values()) + if any_q: + lines.append("- Replication queue entries stuck retrying:") + for node, items in qprob.items(): + if items: + lines.append(f" - `{node}`: {len(items)} stuck task(s)") + lines.append("") + return "\n".join(lines) + + +def render_scenario(scenario: dict) -> str: + lines = [] + lines.append(f"## Scenario: {scenario['name']}") + lines.append("") + lines.append(scenario.get("description", "")) + lines.append("") + lines.append(f"**Overall result: {scenario.get('verdict', 'UNKNOWN')}**") + lines.append("") + for step in scenario.get("steps", []): + lines.append(render_step(step)) + return "\n".join(lines) + + +def render_full_report(results: dict) -> str: + lines = ["# ClickHouse Upgrade Test Report", ""] + lines.append(f"Base (production) version: `{results.get('base_version')}`") + lines.append(f"Target (latest stable) version: `{results.get('latest_version')}`") + lines.append("") + for scenario in results.get("scenarios", []): + lines.append(render_scenario(scenario)) + lines.append("") + lines.append("## Raw results (JSON)") + lines.append("") + lines.append("```json") + lines.append(json.dumps(results, indent=2, default=str)) + lines.append("```") + return "\n".join(lines) + + +def write_report(results: dict, path: Path) -> None: + path.write_text(render_full_report(results)) diff --git a/scripts/clickhouse-upgrade-test/harness/scenarios.py b/scripts/clickhouse-upgrade-test/harness/scenarios.py new file mode 100644 index 00000000..129ebb4e --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/scenarios.py @@ -0,0 +1,193 @@ +""" +The two upgrade scenarios. + +Both scenarios stand up the same 3-node "oonidata_cluster" clone from +scratch (fresh volumes), load the schema + synthetic seed data, then walk +through a version ladder upgrading exactly one node at a time -- i.e. a real +rolling upgrade, never taking the whole shard down. + +* scenario_staged_lts(): walks 24.8.6.70 -> 25.3.14.14 -> 25.8.29.51 -> + 26.3.17.110 -> 26.7.3.19, one LTS hop at a time. Each hop stays within + ClickHouse's documented ~1 year mixed-version compatibility window. + +* scenario_direct_jump(): goes straight from 24.8.6.70 to 26.7.3.19, + node-by-node. This intentionally puts the cluster in a state ClickHouse's + own docs say not to run (>1 year version skew between replicas of the same + shard) so we can observe -- rather than assume -- what actually breaks. +""" +from __future__ import annotations + +import time +from pathlib import Path + +from . import compose, validate +from .ch_http import ChNode +from .seed_data import build_all_seed_statements +from .versions import BASE_VERSION, DIRECT_JUMP, LATEST_VERSION, LTS_HOPS + +SQL_DIR = Path(__file__).resolve().parent.parent / "sql" +NODE_ORDER = ["ch1", "ch2", "ch3"] + + +def make_nodes() -> list[ChNode]: + return [ + ChNode("ch1", http_port=8123), + ChNode("ch2", http_port=8124), + ChNode("ch3", http_port=8125), + ] + + +def fresh_cluster(base_version: str, log=print) -> dict: + """Tear down any previous state and bring up all 3 nodes pinned to base_version.""" + log(f"[setup] tearing down any previous cluster state...") + compose.down(volumes=True) + env = {"CH1_IMAGE": base_version, "CH2_IMAGE": base_version, "CH3_IMAGE": base_version} + log(f"[setup] starting fresh 3-node cluster at {base_version}...") + compose.up(env=env, force_recreate=True) + nodes = make_nodes() + up = validate.wait_all_up(nodes, timeout=180) + if not all(up.values()): + raise RuntimeError(f"cluster did not come up cleanly: {up}\nlogs:\n" + "\n".join(compose.logs(n) for n in NODE_ORDER)) + return env + + +def load_schema_and_seed(log=print) -> None: + nodes = make_nodes() + entry = nodes[0] # ch1 -- schema/seed load happens through one node, ON CLUSTER fans it out + log("[setup] applying schema (ON CLUSTER oonidata_cluster)...") + schema_sql = (SQL_DIR / "001_schema.sql").read_text() + for stmt in [s.strip() for s in schema_sql.split(";") if s.strip() and not s.strip().startswith("--")]: + entry.execute(stmt) + + log("[setup] generating + loading synthetic seed data (see harness/seed_data.py for why it's synthetic)...") + seed = build_all_seed_statements() + for table, stmts in seed.items(): + log(f"[setup] loading {len(stmts)} batch(es) into ooni.{table}...") + for stmt in stmts: + entry.execute(stmt, timeout=120) + + log("[setup] waiting for initial replication to converge across all 3 nodes...") + ok, counts = validate.wait_for_convergence(nodes, timeout=180) + if not ok: + raise RuntimeError(f"seed data did not converge across replicas: {counts}") + log(f"[setup] converged. row counts: {counts}") + + +def _run_upgrade_step(env: dict, node_name: str, new_version: str, log=print) -> dict: + nodes = make_nodes() + other_nodes = [n for n in nodes if n.name != node_name] + + log(f"[upgrade] recreating {node_name} on image {new_version} (others stay up)...") + env = compose.upgrade_node(node_name, new_version, env) + + node_up = validate.wait_until_up(next(n for n in nodes if n.name == node_name), timeout=180) + + # Feed a write while the cluster is in this (possibly mixed-version) state and + # confirm it replicates to every other node -- the sharpest signal of whether + # replication is actually functioning right now. + write_from = other_nodes[0] if other_nodes else nodes[0] + probe = validate.probe_write_then_read(write_from, nodes, timeout=90) + + converged, counts = validate.wait_for_convergence(nodes, timeout=90) + versions = validate.get_versions(nodes) + errors = {n.name: validate.recent_replication_errors(n) for n in nodes} + queue_problems = {n.name: validate.replication_queue_problems(n) for n in nodes} + + step = { + "label": f"upgrade {node_name} -> {new_version}", + "node": node_name, + "new_version": new_version, + "node_came_back_up": node_up, + "versions": versions, + "all_up_throughout": all(validate.wait_all_up(nodes, timeout=5).values()), + "converged": converged, + "row_counts": counts, + "probe": probe, + "errors_found": errors, + "queue_problems": queue_problems, + } + return step, env + + +def _hop_ok(step: dict) -> bool: + return bool( + step.get("node_came_back_up") + and step.get("converged") + and step.get("probe", {}).get("fully_replicated") + and not any(v for v in step.get("errors_found", {}).values()) + ) + + +def scenario_staged_lts(log=print) -> dict: + scenario = { + "name": "Staged rolling upgrade via LTS hops", + "description": ( + f"Rolling (one node at a time) upgrade from {BASE_VERSION} to {LATEST_VERSION}, " + "stepping through each intermediate LTS release so no two replicas are ever " + "more than ~1 year of ClickHouse releases apart (per ClickHouse's documented " + "mixed-version compatibility window)." + ), + "steps": [], + } + env = fresh_cluster(BASE_VERSION, log=log) + load_schema_and_seed(log=log) + + hop_versions = [v for v, _months in LTS_HOPS[1:]] # skip the starting version + all_ok = True + for hop_version in hop_versions: + for node_name in NODE_ORDER: + step, env = _run_upgrade_step(env, node_name, hop_version, log=log) + scenario["steps"].append(step) + ok = _hop_ok(step) + all_ok = all_ok and ok + log(f"[staged] {step['label']}: {'OK' if ok else 'PROBLEM DETECTED'}") + + # Once every replica is on this hop's version, confirm ON CLUSTER DDL still + # works cluster-wide (a real thing OONI does during normal operation). + nodes = make_nodes() + try: + nodes[0].execute( + f"ALTER TABLE ooni.citizenlab ON CLUSTER oonidata_cluster " + f"ADD COLUMN IF NOT EXISTS test_marker_{hop_version.replace('.', '_')} String DEFAULT ''" + ) + ddl_ok = True + except Exception as e: + ddl_ok = False + log(f"[staged] ON CLUSTER ALTER failed after reaching {hop_version}: {e}") + scenario.setdefault("ddl_checks", []).append({"version": hop_version, "on_cluster_alter_ok": ddl_ok}) + all_ok = all_ok and ddl_ok + + scenario["verdict"] = "PASS -- rolling, node-by-node upgrade completed with no data loss, no replication errors, zero full-shard downtime" if all_ok else "FAIL -- see steps above for where it broke" + return scenario + + +def scenario_direct_jump(log=print) -> dict: + scenario = { + "name": "Direct one-hop rolling upgrade (skips all intermediate LTS releases)", + "description": ( + f"Rolling (one node at a time) upgrade straight from {BASE_VERSION} to " + f"{LATEST_VERSION}, the same way you'd do it if you just bumped the version " + "in Ansible and rolled it out host-by-host without stopping to think about " + "version skew. This intentionally spends time with replicas ~23 months apart " + "in version, well past ClickHouse's ~1 year documented compatibility window, " + "to observe what actually happens rather than assume." + ), + "steps": [], + } + env = fresh_cluster(BASE_VERSION, log=log) + load_schema_and_seed(log=log) + + all_ok = True + for node_name in NODE_ORDER: + step, env = _run_upgrade_step(env, node_name, LATEST_VERSION, log=log) + scenario["steps"].append(step) + ok = _hop_ok(step) + all_ok = all_ok and ok + log(f"[direct] {step['label']}: {'OK' if ok else 'PROBLEM DETECTED'}") + + scenario["verdict"] = ( + "PASS -- surprisingly, no issues observed (re-verify; ClickHouse still advises against this)" + if all_ok + else "FAIL -- confirms ClickHouse's guidance: do not skip >1 year of releases in a mixed-version cluster" + ) + return scenario diff --git a/scripts/clickhouse-upgrade-test/harness/seed_data.py b/scripts/clickhouse-upgrade-test/harness/seed_data.py new file mode 100644 index 00000000..f167f34e --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/seed_data.py @@ -0,0 +1,362 @@ +""" +Synthetic seed data generator. + +The task asked for this test to use OONI's initdb sample data from +ooni/backend. That repo's tests don't actually vendor the sample rows in +git -- ooniapi/services/oonimeasurements/tests/conftest.py downloads them at +test time from a public S3 bucket +(https://ooni-data-eu-fra.s3.eu-central-1.amazonaws.com/samples/*.sql.gz). +This sandbox's network egress is restricted to a small allowlist and cannot +reach S3 or Docker Hub, so those exact files couldn't be fetched while +building this harness. + +Instead, this module generates synthetic rows that conform *exactly* to the +production table schemas (see sql/001_schema.sql) with realistic-ish random +values and realistic cardinality/skew (a handful of repeated probe_cc/ASN +values, mostly-null optional columns, etc). This is enough to meaningfully +exercise ReplicatedMergeTree merges, replication, and wide-table part +formats across ClickHouse versions -- which is what the upgrade test cares +about, more than the specific content of the rows. + +If you're running this somewhere with S3 access, see README.md for how to +swap in the real dump instead (download the .sql.gz files referenced above, +`gunzip -c | docker compose exec -T ch1 clickhouse-client --database ooni`). +""" +from __future__ import annotations + +import random +from datetime import datetime, timedelta + +PROBE_CCS = ["US", "DE", "IR", "CN", "RU", "BR", "IN", "IT", "FR", "GB", "EG", "TR"] +ASNS = [7018, 3320, 12389, 4134, 24560, 1221, 3269, 4837, 8151, 6830] +TEST_NAMES = ["web_connectivity", "http_invalid_request_line", "signal", "telegram", "facebook_messenger"] +DOMAINS = ["example.com", "twitter.com", "facebook.com", "bbc.com", "wikipedia.org", "ooni.org"] +SOFTWARE_VERSIONS = ["3.19.0", "3.20.1", "3.21.0"] + + +def _rand_dt(start: datetime, end: datetime) -> datetime: + delta = end - start + return start + timedelta(seconds=random.randint(0, int(delta.total_seconds()))) + + +def _fmt_dt64(dt: datetime) -> str: + return dt.strftime("%Y-%m-%d %H:%M:%S.") + f"{dt.microsecond // 1000:03d}" + + +def _fmt_dt(dt: datetime) -> str: + return dt.strftime("%Y-%m-%d %H:%M:%S") + + +def _esc(s: str) -> str: + return s.replace("\\", "\\\\").replace("'", "\\'") + + +def _sql_str(s: str | None) -> str: + if s is None: + return "NULL" + return f"'{_esc(s)}'" + + +def _sql_arr(items: list[str]) -> str: + return "[" + ", ".join(_sql_str(i) for i in items) + "]" + + +def gen_citizenlab_rows(n: int, seed: int = 1) -> list[tuple]: + rnd = random.Random(seed) + rows = [] + for i in range(n): + domain = rnd.choice(DOMAINS) + rows.append((domain, f"https://{domain}/", rnd.choice(PROBE_CCS), rnd.choice(["GRP", "NEWS", "SRCH"]))) + return rows + + +def citizenlab_insert_statements(rows: list[tuple], batch_size: int = 500) -> list[str]: + cols = "(domain, url, cc, category_code)" + stmts = [] + for i in range(0, len(rows), batch_size): + batch = rows[i : i + batch_size] + values = ", ".join( + f"({_sql_str(d)}, {_sql_str(u)}, {_sql_str(cc)}, {_sql_str(cat)})" for d, u, cc, cat in batch + ) + stmts.append(f"INSERT INTO ooni.citizenlab {cols} VALUES {values}") + return stmts + + +def gen_fastpath_rows(n: int, seed: int = 2) -> list[dict]: + rnd = random.Random(seed) + start, end = datetime(2025, 1, 1), datetime(2025, 6, 1) + rows = [] + for i in range(n): + mst = _rand_dt(start, end) + rows.append( + { + "measurement_uid": f"seed-fp-{i:08d}", + "report_id": f"20250101T000000Z_{i:06d}", + "input": rnd.choice(DOMAINS), + "probe_cc": rnd.choice(PROBE_CCS), + "probe_asn": rnd.choice(ASNS), + "test_name": rnd.choice(TEST_NAMES), + "test_start_time": _fmt_dt(mst), + "measurement_start_time": _fmt_dt(mst), + "filename": f"{i}.json", + "scores": "{}", + "platform": rnd.choice(["android", "ios", "linux"]), + "anomaly": rnd.choice(["true", "false"]), + "confirmed": "false", + "msm_failure": "false", + "domain": rnd.choice(DOMAINS), + "software_name": "ooniprobe", + "software_version": rnd.choice(SOFTWARE_VERSIONS), + "control_failure": "", + "blocking_general": round(rnd.random(), 3), + "is_ssl_expected": rnd.choice([0, 1]), + "page_len": rnd.randint(100, 50000), + "page_len_ratio": round(rnd.random(), 3), + "server_cc": rnd.choice(PROBE_CCS), + "server_asn": rnd.randint(0, 100), + "server_as_name": "Example AS", + "test_version": "0.1.0", + "architecture": "amd64", + "engine_name": "ooniprobe-engine", + "engine_version": rnd.choice(SOFTWARE_VERSIONS), + "test_runtime": round(rnd.uniform(0.1, 30.0), 3), + "blocking_type": rnd.choice(["", "dns", "tcp_ip", "http-failure"]), + "test_helper_address": "https://th.ooni.org", + "test_helper_type": "https", + "ooni_run_link_id": None, + } + ) + return rows + + +def fastpath_insert_statements(rows: list[dict], batch_size: int = 500) -> list[str]: + cols = list(rows[0].keys()) + col_sql = "(" + ", ".join(cols) + ")" + stmts = [] + for i in range(0, len(rows), batch_size): + batch = rows[i : i + batch_size] + value_tuples = [] + for r in batch: + parts = [] + for c in cols: + v = r[c] + if v is None: + parts.append("NULL") + elif isinstance(v, (int, float)): + parts.append(str(v)) + else: + parts.append(_sql_str(str(v))) + value_tuples.append("(" + ", ".join(parts) + ")") + stmts.append(f"INSERT INTO ooni.fastpath {col_sql} VALUES {', '.join(value_tuples)}") + return stmts + + +def gen_analysis_rows(n: int, seed: int = 3) -> list[dict]: + rnd = random.Random(seed) + start, end = datetime(2025, 1, 1), datetime(2025, 6, 1) + rows = [] + for i in range(n): + mst = _rand_dt(start, end) + uid = f"{mst.strftime('%Y%m%d%H')}_seed_an_{i:08d}" + rows.append( + { + "domain": rnd.choice(DOMAINS), + "input": f"https://{rnd.choice(DOMAINS)}/", + "test_name": "web_connectivity", + "probe_asn": rnd.choice(ASNS), + "probe_as_org_name": "Example ISP", + "probe_cc": rnd.choice(PROBE_CCS), + "resolver_asn": rnd.choice(ASNS), + "resolver_as_cc": rnd.choice(PROBE_CCS), + "network_type": rnd.choice(["wifi", "mobile"]), + "measurement_start_time": _fmt_dt64(mst), + "measurement_uid": uid, + "ooni_run_link_id": "0", + "top_probe_analysis": rnd.choice([None, "ok", "blocked"]), + "top_dns_failure": None, + "top_tcp_failure": None, + "top_tls_failure": None, + "dns_blocked": round(rnd.random(), 3), + "dns_down": round(rnd.random(), 3), + "dns_ok": round(rnd.random(), 3), + "tcp_blocked": round(rnd.random(), 3), + "tcp_down": round(rnd.random(), 3), + "tcp_ok": round(rnd.random(), 3), + "tls_blocked": round(rnd.random(), 3), + "tls_down": round(rnd.random(), 3), + "tls_ok": round(rnd.random(), 3), + } + ) + return rows + + +def analysis_insert_statements(rows: list[dict], batch_size: int = 500) -> list[str]: + cols = list(rows[0].keys()) + col_sql = "(" + ", ".join(cols) + ")" + stmts = [] + for i in range(0, len(rows), batch_size): + batch = rows[i : i + batch_size] + value_tuples = [] + for r in batch: + parts = [] + for c in cols: + v = r[c] + if v is None: + parts.append("NULL") + elif isinstance(v, (int, float)): + parts.append(str(v)) + else: + parts.append(_sql_str(str(v))) + value_tuples.append("(" + ", ".join(parts) + ")") + stmts.append(f"INSERT INTO ooni.analysis_web_measurement {col_sql} VALUES {', '.join(value_tuples)}") + return stmts + + +def gen_obs_web_rows(n: int, seed: int = 4) -> list[dict]: + rnd = random.Random(seed) + start, end = datetime(2025, 1, 1), datetime(2025, 6, 1) + rows = [] + for i in range(n): + mst = _rand_dt(start, end) + uid = f"{mst.strftime('%Y%m%d%H')}_seed_ow_{i:08d}" + has_tls = rnd.random() > 0.3 + rows.append( + { + "measurement_uid": uid, + "observation_idx": i % 5, + "input": f"https://{rnd.choice(DOMAINS)}/", + "report_id": f"20250101T000000Z_{i:06d}", + "ooni_run_link_id": "0", + "measurement_start_time": _fmt_dt64(mst), + "software_name": "ooniprobe", + "software_version": rnd.choice(SOFTWARE_VERSIONS), + "test_name": "web_connectivity", + "test_version": "0.1.0", + "bucket_date": mst.strftime("%Y-%m-%d"), + "probe_asn": rnd.choice(ASNS), + "probe_cc": rnd.choice(PROBE_CCS), + "probe_as_org_name": "Example ISP", + "probe_as_cc": rnd.choice(PROBE_CCS), + "probe_as_name": "Example AS Name", + "network_type": rnd.choice(["wifi", "mobile"]), + "platform": rnd.choice(["android", "ios", "linux"]), + "origin": "probe", + "engine_name": "ooniprobe-engine", + "engine_version": rnd.choice(SOFTWARE_VERSIONS), + "architecture": "amd64", + "resolver_ip": f"8.8.{rnd.randint(0,255)}.{rnd.randint(0,255)}", + "resolver_asn": rnd.choice(ASNS), + "resolver_cc": rnd.choice(PROBE_CCS), + "resolver_as_org_name": "Example Resolver Org", + "resolver_as_cc": rnd.choice(PROBE_CCS), + "resolver_is_scrubbed": 0, + "resolver_asn_probe": rnd.choice(ASNS), + "resolver_as_org_name_probe": "Example Resolver Org", + "created_at": _fmt_dt(mst), + "target_id": None, + "hostname": rnd.choice(DOMAINS), + "transaction_id": i % 10, + "ip": f"93.184.{rnd.randint(0,255)}.{rnd.randint(0,255)}", + "port": 443, + "ip_asn": rnd.choice(ASNS), + "ip_as_org_name": "Example Host Org", + "ip_as_cc": rnd.choice(PROBE_CCS), + "ip_cc": rnd.choice(PROBE_CCS), + "ip_is_bogon": 0, + "dns_query_type": "A", + "dns_failure": None, + "dns_engine": "system", + "dns_engine_resolver_address": None, + "dns_answer_type": "A", + "dns_answer": f"93.184.{rnd.randint(0,255)}.{rnd.randint(0,255)}", + "dns_answer_asn": rnd.choice(ASNS), + "dns_answer_as_org_name": "Example Org", + "dns_t": round(rnd.uniform(0.01, 2.0), 4), + "tcp_failure": None, + "tcp_success": 1, + "tcp_t": round(rnd.uniform(0.01, 2.0), 4), + "tls_failure": None if has_tls else "connection_reset", + "tls_server_name": rnd.choice(DOMAINS) if has_tls else None, + "tls_outer_server_name": None, + "tls_echconfig": None, + "tls_version": "TLSv1.3" if has_tls else None, + "tls_cipher_suite": "TLS_AES_128_GCM_SHA256" if has_tls else None, + "tls_is_certificate_valid": 1 if has_tls else None, + "tls_end_entity_certificate_fingerprint": "deadbeef" if has_tls else None, + "tls_end_entity_certificate_subject": None, + "tls_end_entity_certificate_subject_common_name": None, + "tls_end_entity_certificate_issuer": None, + "tls_end_entity_certificate_issuer_common_name": None, + "tls_end_entity_certificate_san_list": [], + "tls_end_entity_certificate_not_valid_after": None, + "tls_end_entity_certificate_not_valid_before": None, + "tls_certificate_chain_length": 2 if has_tls else None, + "tls_certificate_chain_fingerprints": [], + "tls_handshake_read_count": None, + "tls_handshake_write_count": None, + "tls_handshake_read_bytes": None, + "tls_handshake_write_bytes": None, + "tls_handshake_last_operation": None, + "tls_handshake_time": round(rnd.uniform(0.01, 1.0), 4) if has_tls else None, + "tls_t": round(rnd.uniform(0.01, 2.0), 4) if has_tls else None, + "http_request_url": f"https://{rnd.choice(DOMAINS)}/", + "http_network": "tcp", + "http_alpn": "h2", + "http_failure": None, + "http_request_body_length": 0, + "http_request_method": "GET", + "http_runtime": round(rnd.uniform(0.01, 3.0), 4), + "http_response_body_length": rnd.randint(100, 90000), + "http_response_body_is_truncated": 0, + "http_response_body_sha1": "0" * 40, + "http_response_status_code": rnd.choice([200, 200, 200, 301, 403, 503]), + "http_response_header_location": None, + "http_response_header_server": "nginx", + "http_request_redirect_from": None, + "http_request_body_is_truncated": 0, + "http_t": round(rnd.uniform(0.01, 3.0), 4), + "probe_analysis": rnd.choice([None, "ok", "blocked"]), + } + ) + return rows + + +_ARRAY_COLS = {"tls_end_entity_certificate_san_list", "tls_certificate_chain_fingerprints"} + + +def obs_web_insert_statements(rows: list[dict], batch_size: int = 200) -> list[str]: + cols = list(rows[0].keys()) + col_sql = "(" + ", ".join(cols) + ")" + stmts = [] + for i in range(0, len(rows), batch_size): + batch = rows[i : i + batch_size] + value_tuples = [] + for r in batch: + parts = [] + for c in cols: + v = r[c] + if c in _ARRAY_COLS: + parts.append(_sql_arr(v or [])) + elif v is None: + parts.append("NULL") + elif isinstance(v, (int, float)): + parts.append(str(v)) + else: + parts.append(_sql_str(str(v))) + value_tuples.append("(" + ", ".join(parts) + ")") + stmts.append(f"INSERT INTO ooni.obs_web {col_sql} VALUES {', '.join(value_tuples)}") + return stmts + + +def build_all_seed_statements( + n_obs_web: int = 5000, + n_analysis: int = 1000, + n_fastpath: int = 2000, + n_citizenlab: int = 200, +) -> dict[str, list[str]]: + return { + "citizenlab": citizenlab_insert_statements(gen_citizenlab_rows(n_citizenlab)), + "fastpath": fastpath_insert_statements(gen_fastpath_rows(n_fastpath)), + "analysis_web_measurement": analysis_insert_statements(gen_analysis_rows(n_analysis)), + "obs_web": obs_web_insert_statements(gen_obs_web_rows(n_obs_web)), + } diff --git a/scripts/clickhouse-upgrade-test/harness/validate.py b/scripts/clickhouse-upgrade-test/harness/validate.py new file mode 100644 index 00000000..1d201610 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/validate.py @@ -0,0 +1,182 @@ +""" +Cluster health / correctness checks used before, during, and after each +upgrade step. +""" +from __future__ import annotations + +import time +import uuid + +from .ch_http import ChNode, ClickHouseError + +TABLES = ["citizenlab", "fastpath", "analysis_web_measurement", "obs_web"] + + +def wait_until_up(node: ChNode, timeout: float = 90.0) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if node.ping(): + return True + time.sleep(2) + return False + + +def wait_all_up(nodes: list[ChNode], timeout: float = 90.0) -> dict[str, bool]: + return {n.name: wait_until_up(n, timeout=timeout) for n in nodes} + + +def get_versions(nodes: list[ChNode]) -> dict[str, str | None]: + out = {} + for n in nodes: + try: + out[n.name] = n.version() + except Exception: + out[n.name] = None + return out + + +def cluster_replica_count(node: ChNode) -> int: + """How many replicas does system.clusters see as reachable for oonidata_cluster?""" + rows = node.query_rows( + "SELECT count() AS c FROM system.clusters WHERE cluster = 'oonidata_cluster'" + ) + return int(rows[0]["c"]) if rows else 0 + + +def replicas_readonly_status(node: ChNode) -> list[dict]: + """Per-table replication state as seen from this node.""" + return node.query_rows( + "SELECT database, table, is_readonly, is_session_expired, " + "future_parts, parts_to_check, queue_size, absolute_delay " + "FROM system.replicas WHERE database = 'ooni'" + ) + + +def row_counts(node: ChNode) -> dict[str, int | None]: + out = {} + for t in TABLES: + try: + out[t] = int(node.query_scalar(f"SELECT count() FROM ooni.{t}")) + except Exception: + out[t] = None + return out + + +def row_counts_all_nodes(nodes: list[ChNode]) -> dict[str, dict[str, int | None]]: + return {n.name: row_counts(n) for n in nodes} + + +def counts_converged(counts_by_node: dict[str, dict[str, int | None]]) -> bool: + """True if every table has the same non-None row count on every node.""" + node_names = list(counts_by_node.keys()) + if not node_names: + return False + for t in TABLES: + values = {counts_by_node[n].get(t) for n in node_names} + if len(values) != 1 or None in values: + return False + return True + + +def wait_for_convergence(nodes: list[ChNode], timeout: float = 120.0, interval: float = 3.0): + """Poll row counts on all nodes until they match (replication caught up).""" + deadline = time.time() + timeout + last = None + while time.time() < deadline: + last = row_counts_all_nodes(nodes) + if counts_converged(last): + return True, last + time.sleep(interval) + return False, last + + +def probe_write_then_read(write_node: ChNode, read_nodes: list[ChNode], timeout: float = 60.0) -> dict: + """ + Insert one uniquely identifiable row on `write_node`, then poll every + node in `read_nodes` until the row shows up (or timeout). This is the + most direct evidence of whether replication is actually working end to + end during a mixed-version state, independent of aggregate row counts. + """ + probe_id = f"probe-{uuid.uuid4().hex[:12]}" + result = {"probe_id": probe_id, "write_node": write_node.name, "write_ok": False, "read_back": {}} + try: + write_node.execute( + "INSERT INTO ooni.citizenlab (domain, url, cc, category_code) VALUES " + f"('{probe_id}.example.test', 'https://{probe_id}.example.test/', 'ZZ', 'PROBE')" + ) + result["write_ok"] = True + except ClickHouseError as e: + result["write_error"] = str(e) + return result + + deadline = time.time() + timeout + pending = {n.name: n for n in read_nodes} + while pending and time.time() < deadline: + for name in list(pending): + n = pending[name] + try: + c = n.query_scalar( + f"SELECT count() FROM ooni.citizenlab WHERE domain = '{probe_id}.example.test'" + ) + if c and int(c) > 0: + result["read_back"][name] = True + del pending[name] + except ClickHouseError: + pass + if pending: + time.sleep(2) + for name in pending: + result["read_back"][name] = False + result["fully_replicated"] = len(pending) == 0 + return result + + +def recent_replication_errors(node: ChNode, since_minutes: int = 30) -> list[dict]: + """ + Errors ClickHouse itself has logged for replication-related exception + codes since the upgrade step started. This is what actually shows up + when nodes run mismatched, incompatible versions (e.g. checksum + mismatches, unknown part format, protocol errors). + """ + try: + return node.query_rows( + f""" + SELECT name, value, last_error_message, last_error_time + FROM system.errors + WHERE last_error_time > now() - INTERVAL {since_minutes} MINUTE + AND ( + name LIKE '%REPLICA%' OR + name LIKE '%CHECKSUM%' OR + name LIKE '%UNKNOWN_FORMAT%' OR + name LIKE '%TOO_OLD%' OR + name LIKE '%NETWORK%' OR + name LIKE '%UNFINISHED%' OR + name LIKE '%NOT_ENOUGH_SPACE%' OR + name LIKE '%CANNOT_READ_ALL_DATA%' + ) + ORDER BY last_error_time DESC + """ + ) + except ClickHouseError: + return [] + + +def replication_queue_problems(node: ChNode) -> list[dict]: + try: + return node.query_rows( + "SELECT database, table, node_name, type, num_tries, last_exception " + "FROM system.replication_queue WHERE num_tries > 2" + ) + except ClickHouseError: + return [] + + +def full_health_snapshot(nodes: list[ChNode]) -> dict: + return { + "versions": get_versions(nodes), + "up": {n.name: n.ping() for n in nodes}, + "row_counts": row_counts_all_nodes(nodes), + "replicas": {n.name: replicas_readonly_status(n) for n in nodes}, + "errors": {n.name: recent_replication_errors(n) for n in nodes}, + "queue_problems": {n.name: replication_queue_problems(n) for n in nodes}, + } diff --git a/scripts/clickhouse-upgrade-test/harness/versions.py b/scripts/clickhouse-upgrade-test/harness/versions.py new file mode 100644 index 00000000..b1d2b099 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/harness/versions.py @@ -0,0 +1,41 @@ +""" +Version constants for the upgrade test. + +Sourced from: + - ooni/devops ansible/group_vars/clickhouse/vars.yml -> clickhouse_version: 24.8.6.70 + (this is BASE_VERSION -- what's running in production per issue ooni/devops#437) + - ClickHouse's own release history (https://clickhouse.com/docs/whats-new/changelog, + https://endoflife.date/clickhouse) as of 2026-08-10: + + 24.8.6.70 LTS, released 2024-08 <- current prod version + 25.3.14.14 LTS, released 2025-03-20 + 25.8.29.51 LTS, released 2025-08-29 + 26.3.17.110 LTS, released 2026-03-26 + 26.7.3.19 latest stable, released 2026-07-22 <- upgrade target + +ClickHouse documents a ~1 year mixed-version compatibility window for +replicated clusters (https://clickhouse.com/docs/operations/update): nodes +more than a year apart in version should not be run together mid-upgrade. +BASE_VERSION -> LATEST_VERSION spans ~23 months, so a single-hop rolling +upgrade is out of that window; each LTS_HOPS step individually stays inside +it (5-7 months apart). +""" + +BASE_VERSION = "24.8.6.70" # current production version (issue #437) +LATEST_VERSION = "26.7.3.19" # latest stable as of 2026-08-10 + +# Each entry: (version, months_since_previous) -- used for the staged-upgrade +# scenario, walking one LTS release at a time up to the latest stable. +LTS_HOPS = [ + ("24.8.6.70", None), # starting point + ("25.3.14.14", 7), + ("25.8.29.51", 5), + ("26.3.17.110", 7), + ("26.7.3.19", 4), # final hop lands on latest stable (not itself LTS) +] + +# For the "direct jump" scenario we go straight from BASE to LATEST. +DIRECT_JUMP = [ + ("24.8.6.70", None), + ("26.7.3.19", 23), # ~23 months apart -- exceeds the 1-year window +] diff --git a/scripts/clickhouse-upgrade-test/results/.gitkeep b/scripts/clickhouse-upgrade-test/results/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/scripts/clickhouse-upgrade-test/run_test.py b/scripts/clickhouse-upgrade-test/run_test.py new file mode 100644 index 00000000..859d36e6 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/run_test.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +""" +Entry point for the ClickHouse upgrade test. + +Usage: + python3 run_test.py --scenario staged # recommended path: LTS-hop rolling upgrade + python3 run_test.py --scenario direct # naive single-hop rolling upgrade + python3 run_test.py --scenario both # run both, one after another (default) + +Requires: Docker + the Docker Compose plugin, and network access to pull +clickhouse/clickhouse-server images from Docker Hub (the sandbox this +harness was authored in could not reach Docker Hub -- see README.md). + +Writes a Markdown + JSON report to results/report.md and results/report.json. +""" +from __future__ import annotations + +import argparse +import datetime +import json +import sys +import traceback +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from harness import compose, report +from harness.scenarios import scenario_direct_jump, scenario_staged_lts +from harness.versions import BASE_VERSION, LATEST_VERSION + +RESULTS_DIR = Path(__file__).resolve().parent / "results" + + +def log(msg: str) -> None: + ts = datetime.datetime.utcnow().strftime("%H:%M:%S") + print(f"[{ts}] {msg}", flush=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--scenario", choices=["staged", "direct", "both"], default="both") + parser.add_argument("--keep-up", action="store_true", help="Don't tear down the cluster when done") + args = parser.parse_args() + + log("Validating docker-compose.yml (no image pull required for this check)...") + try: + compose.config_check() + except Exception as e: + log(f"docker compose config failed -- fix docker-compose.yml before running: {e}") + return 2 + + results = { + "generated_at": datetime.datetime.utcnow().isoformat() + "Z", + "base_version": BASE_VERSION, + "latest_version": LATEST_VERSION, + "scenarios": [], + } + + scenarios_to_run = [] + if args.scenario in ("staged", "both"): + scenarios_to_run.append(("staged", scenario_staged_lts)) + if args.scenario in ("direct", "both"): + scenarios_to_run.append(("direct", scenario_direct_jump)) + + exit_code = 0 + for name, fn in scenarios_to_run: + log(f"=== Running scenario: {name} ===") + try: + scenario_result = fn(log=log) + results["scenarios"].append(scenario_result) + log(f"=== Scenario {name} verdict: {scenario_result['verdict']} ===") + if scenario_result["verdict"].startswith("FAIL"): + exit_code = 1 + except Exception as e: + log(f"Scenario {name} raised an exception: {e}") + traceback.print_exc() + results["scenarios"].append({"name": name, "verdict": f"ERROR: {e}", "steps": []}) + exit_code = 1 + + RESULTS_DIR.mkdir(exist_ok=True) + (RESULTS_DIR / "report.json").write_text(json.dumps(results, indent=2, default=str)) + report.write_report(results, RESULTS_DIR / "report.md") + log(f"Report written to {RESULTS_DIR / 'report.md'} and {RESULTS_DIR / 'report.json'}") + + if not args.keep_up: + log("Tearing down cluster (pass --keep-up to leave it running)...") + compose.down(volumes=True) + + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/clickhouse-upgrade-test/sql/001_schema.sql b/scripts/clickhouse-upgrade-test/sql/001_schema.sql new file mode 100644 index 00000000..a0b6b3c9 --- /dev/null +++ b/scripts/clickhouse-upgrade-test/sql/001_schema.sql @@ -0,0 +1,305 @@ +-- Schema for the test cluster. +-- +-- Sourced from two places: +-- * ooni/devops scripts/cluster-migration/schema.sql -- the actual +-- production DDL for `fastpath`, `citizenlab`, `jsonl`, +-- `analysis_web_measurement`, `event_detector_changepoints` and +-- `faulty_measurements`. These are reproduced close to verbatim +-- (ReplicatedReplacingMergeTree, ON CLUSTER oonidata_cluster, the +-- '/clickhouse/{cluster}/tables///{shard}' zk path convention). +-- * ooni/backend column definitions for `obs_web` and `analysis_web_measurement` +-- (ooniapi/services/oonimeasurements/tests/fixtures/initdb/clickhouse.sql) +-- and `fastpath`/`citizenlab`/`jsonl`/`event_detector_changepoints` +-- (ooniapi/services/oonimeasurements/tests/migrations/0_clickhouse_init_tables.sql). +-- The backend repo's copies are plain MergeTree/ReplacingMergeTree because +-- they're used for single-node CI tests; here they're converted to their +-- replicated equivalents so we exercise the same replication path prod +-- uses. `obs_web` does not appear in devops' schema.sql (it's created by +-- a different repo/pipeline not in scope here), so its DDL below is +-- derived from the backend fixture's column list. +-- +-- All DDL runs ON CLUSTER so it exercises ClickHouse's distributed_ddl queue +-- (the same mechanism prod uses to apply schema changes to all 3 replicas). + +CREATE DATABASE IF NOT EXISTS ooni ON CLUSTER oonidata_cluster; + +CREATE TABLE IF NOT EXISTS ooni.jsonl ON CLUSTER oonidata_cluster +( + `report_id` String, + `input` String, + `s3path` String, + `linenum` Int32, + `measurement_uid` String, + `date` Date, + `source` String, + `update_time` DateTime64(3) MATERIALIZED now64() +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/jsonl/{shard}', + '{replica}', + update_time +) +ORDER BY (report_id, input, measurement_uid) +SETTINGS index_granularity = 8192; + +CREATE TABLE IF NOT EXISTS ooni.fastpath ON CLUSTER oonidata_cluster +( + `measurement_uid` String, + `report_id` String, + `input` String, + `probe_cc` LowCardinality(String), + `probe_asn` Int32, + `test_name` LowCardinality(String), + `test_start_time` DateTime, + `measurement_start_time` DateTime, + `filename` String, + `scores` String, + `platform` String, + `anomaly` String, + `confirmed` String, + `msm_failure` String, + `domain` String, + `software_name` String, + `software_version` String, + `control_failure` String, + `blocking_general` Float32, + `is_ssl_expected` Int8, + `page_len` Int32, + `page_len_ratio` Float32, + `server_cc` String, + `server_asn` Int8, + `server_as_name` String, + `update_time` DateTime64(3) MATERIALIZED now64(), + `test_version` String, + `architecture` String, + `engine_name` LowCardinality(String), + `engine_version` String, + `test_runtime` Float32, + `blocking_type` String, + `test_helper_address` LowCardinality(String), + `test_helper_type` LowCardinality(String), + `ooni_run_link_id` Nullable(UInt64), + `is_verified` LowCardinality(String) DEFAULT 'u', + INDEX fastpath_rid_idx report_id TYPE minmax GRANULARITY 1, + INDEX measurement_uid_idx measurement_uid TYPE minmax GRANULARITY 8 +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/fastpath/{shard}', + '{replica}', + update_time +) +ORDER BY (measurement_start_time, report_id, input, measurement_uid) +SETTINGS index_granularity = 8192; + +CREATE TABLE IF NOT EXISTS ooni.citizenlab ON CLUSTER oonidata_cluster +( + `domain` String, + `url` String, + `cc` FixedString(32), + `category_code` String +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/citizenlab/{shard}', + '{replica}' +) +ORDER BY (domain, url, cc, category_code) +SETTINGS index_granularity = 4; + +CREATE TABLE IF NOT EXISTS ooni.analysis_web_measurement ON CLUSTER oonidata_cluster +( + `domain` String, + `input` String, + `test_name` String, + `probe_asn` UInt32, + `probe_as_org_name` String, + `probe_cc` String, + `resolver_asn` UInt32, + `resolver_as_cc` String, + `network_type` String, + `measurement_start_time` DateTime64(3, 'UTC'), + `measurement_uid` String, + `ooni_run_link_id` String, + `top_probe_analysis` Nullable(String), + `top_dns_failure` Nullable(String), + `top_tcp_failure` Nullable(String), + `top_tls_failure` Nullable(String), + `dns_blocked` Float32, + `dns_down` Float32, + `dns_ok` Float32, + `tcp_blocked` Float32, + `tcp_down` Float32, + `tcp_ok` Float32, + `tls_blocked` Float32, + `tls_down` Float32, + `tls_ok` Float32 +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/analysis_web_measurement/{shard}', + '{replica}' +) +PARTITION BY substring(measurement_uid, 1, 6) +PRIMARY KEY measurement_uid +ORDER BY (measurement_uid, measurement_start_time, probe_cc, probe_asn, domain) +SETTINGS index_granularity = 8192; + +-- Derived from ooni/backend ooniapi/services/oonimeasurements/tests/fixtures/initdb/clickhouse.sql +-- (converted from ReplacingMergeTree to its replicated equivalent). +CREATE TABLE IF NOT EXISTS ooni.obs_web ON CLUSTER oonidata_cluster +( + `measurement_uid` String, + `observation_idx` UInt16, + `input` Nullable(String), + `report_id` String, + `ooni_run_link_id` String DEFAULT '', + `measurement_start_time` DateTime64(3, 'UTC'), + `software_name` String, + `software_version` String, + `test_name` String, + `test_version` String, + `bucket_date` String, + `probe_asn` UInt32, + `probe_cc` String, + `probe_as_org_name` String, + `probe_as_cc` String, + `probe_as_name` String, + `network_type` String, + `platform` String, + `origin` String, + `engine_name` String, + `engine_version` String, + `architecture` String, + `resolver_ip` String, + `resolver_asn` UInt32, + `resolver_cc` String, + `resolver_as_org_name` String, + `resolver_as_cc` String, + `resolver_is_scrubbed` UInt8, + `resolver_asn_probe` UInt32, + `resolver_as_org_name_probe` String, + `created_at` Nullable(DateTime('UTC')), + `target_id` Nullable(String), + `hostname` Nullable(String), + `transaction_id` Nullable(UInt16), + `ip` Nullable(String), + `port` Nullable(UInt16), + `ip_asn` Nullable(UInt32), + `ip_as_org_name` Nullable(String), + `ip_as_cc` Nullable(String), + `ip_cc` Nullable(String), + `ip_is_bogon` Nullable(UInt8), + `dns_query_type` Nullable(String), + `dns_failure` Nullable(String), + `dns_engine` Nullable(String), + `dns_engine_resolver_address` Nullable(String), + `dns_answer_type` Nullable(String), + `dns_answer` Nullable(String), + `dns_answer_asn` Nullable(UInt32), + `dns_answer_as_org_name` Nullable(String), + `dns_t` Nullable(Float64), + `tcp_failure` Nullable(String), + `tcp_success` Nullable(UInt8), + `tcp_t` Nullable(Float64), + `tls_failure` Nullable(String), + `tls_server_name` Nullable(String), + `tls_outer_server_name` Nullable(String), + `tls_echconfig` Nullable(String), + `tls_version` Nullable(String), + `tls_cipher_suite` Nullable(String), + `tls_is_certificate_valid` Nullable(UInt8), + `tls_end_entity_certificate_fingerprint` Nullable(String), + `tls_end_entity_certificate_subject` Nullable(String), + `tls_end_entity_certificate_subject_common_name` Nullable(String), + `tls_end_entity_certificate_issuer` Nullable(String), + `tls_end_entity_certificate_issuer_common_name` Nullable(String), + `tls_end_entity_certificate_san_list` Array(String), + `tls_end_entity_certificate_not_valid_after` Nullable(DateTime64(3, 'UTC')), + `tls_end_entity_certificate_not_valid_before` Nullable(DateTime64(3, 'UTC')), + `tls_certificate_chain_length` Nullable(UInt16), + `tls_certificate_chain_fingerprints` Array(String), + `tls_handshake_read_count` Nullable(UInt16), + `tls_handshake_write_count` Nullable(UInt16), + `tls_handshake_read_bytes` Nullable(UInt32), + `tls_handshake_write_bytes` Nullable(UInt32), + `tls_handshake_last_operation` Nullable(String), + `tls_handshake_time` Nullable(Float64), + `tls_t` Nullable(Float64), + `http_request_url` Nullable(String), + `http_network` Nullable(String), + `http_alpn` Nullable(String), + `http_failure` Nullable(String), + `http_request_body_length` Nullable(UInt32), + `http_request_method` Nullable(String), + `http_runtime` Nullable(Float64), + `http_response_body_length` Nullable(Int32), + `http_response_body_is_truncated` Nullable(UInt8), + `http_response_body_sha1` Nullable(String), + `http_response_status_code` Nullable(UInt16), + `http_response_header_location` Nullable(String), + `http_response_header_server` Nullable(String), + `http_request_redirect_from` Nullable(String), + `http_request_body_is_truncated` Nullable(UInt8), + `http_t` Nullable(Float64), + `probe_analysis` Nullable(String) +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/obs_web/{shard}', + '{replica}' +) +PRIMARY KEY (measurement_uid, observation_idx) +ORDER BY (measurement_uid, observation_idx, measurement_start_time, probe_cc, probe_asn) +SETTINGS index_granularity = 8192; + +CREATE TABLE IF NOT EXISTS ooni.event_detector_changepoints ON CLUSTER oonidata_cluster +( + `probe_asn` UInt32, + `probe_cc` String, + `domain` String, + `ts` DateTime64(3, 'UTC'), + `count_isp_resolver` Nullable(UInt32), + `count_other_resolver` Nullable(UInt32), + `count` Nullable(UInt32), + `dns_isp_blocked` Nullable(Float32), + `dns_other_blocked` Nullable(Float32), + `tcp_blocked` Nullable(Float32), + `tls_blocked` Nullable(Float32), + `dns_isp_blocked_current_state` String DEFAULT 'ok', + `dns_isp_blocked_s_pos` Nullable(Float32), + `dns_isp_blocked_s_neg` Nullable(Float32), + `dns_other_blocked_current_state` String DEFAULT 'ok', + `dns_other_blocked_s_pos` Nullable(Float32), + `dns_other_blocked_s_neg` Nullable(Float32), + `tcp_blocked_current_state` String DEFAULT 'ok', + `tcp_blocked_s_pos` Nullable(Float32), + `tcp_blocked_s_neg` Nullable(Float32), + `tls_blocked_current_state` String DEFAULT 'ok', + `tls_blocked_s_pos` Nullable(Float32), + `tls_blocked_s_neg` Nullable(Float32), + `change_dir` Nullable(Int8), + `current_state` String DEFAULT 'ok', + `s_pos` Nullable(Float32), + `s_neg` Nullable(Float32), + `h` Nullable(Float32), + `block_type` String +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/event_detector_changepoints/{shard}', + '{replica}' +) +PARTITION BY toYYYYMM(ts) +ORDER BY (probe_asn, probe_cc, ts, domain) +SETTINGS index_granularity = 8192; + +CREATE TABLE IF NOT EXISTS ooni.faulty_measurements ON CLUSTER oonidata_cluster +( + `ts` DateTime64(3, 'UTC') DEFAULT now64(), + `type` String, + `uid` UUID DEFAULT generateUUIDv4(), + `probe_cc` String, + `probe_asn` UInt32, + `details` String +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/{cluster}/tables/ooni/faulty_measurements/{shard}', + '{replica}' +) +ORDER BY (ts, type, probe_cc, probe_asn, uid);