From 771e02b98835a9ae90e2ec93bbbc9c647fe15e4b Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Tue, 11 Aug 2026 17:57:02 +0200 Subject: [PATCH 01/12] Fix job statistics anonymization flag and improve schema probing logic --- cratedb_toolkit/cfr/cli.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cratedb_toolkit/cfr/cli.py b/cratedb_toolkit/cfr/cli.py index e52f1f67..825ef99e 100644 --- a/cratedb_toolkit/cfr/cli.py +++ b/cratedb_toolkit/cfr/cli.py @@ -109,7 +109,7 @@ def job_statistics(ctx: click.Context): @click.option( "--anonymize", type=str, - is_flag=True, + is_flag=False, flag_value="decoder_dictionary.json", # Use this value when flag is used without value default=None, # No anonymization by default help="Path to the decoder dictionary file for anonymizing SQL statements", @@ -196,7 +196,7 @@ def job_statistics_report(ctx: click.Context): import cratedb_toolkit.cfr.marimo address = DatabaseAddress.from_string(ctx.meta["cluster_url"]) - probe_database_schema(address, schema_name="stats") + probe_database_schema(address, schema_name=address.schema or "stats") os.environ["CRATEDB_CLUSTER_URL"] = address.dburi cratedb_toolkit.cfr.marimo.app.run() @@ -214,7 +214,7 @@ def job_statistics_ui(ctx: click.Context): import cratedb_toolkit.cfr.marimo address = DatabaseAddress.from_string(ctx.meta["cluster_url"]) - probe_database_schema(address, schema_name="stats") + probe_database_schema(address, schema_name=address.schema or "stats") os.environ["CRATEDB_CLUSTER_URL"] = address.dburi server = marimo.create_asgi_app() server = server.with_app(path="/", root=cratedb_toolkit.cfr.marimo.__file__) From 569c1b83ccb48ebbab6139f9e8296b3f0546b5ad Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Tue, 11 Aug 2026 18:00:24 +0200 Subject: [PATCH 02/12] Update job statistics descriptions and correct percentile calculation in job logs --- cratedb_toolkit/info/library.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cratedb_toolkit/info/library.py b/cratedb_toolkit/info/library.py index a5276fa6..993fbc56 100644 --- a/cratedb_toolkit/info/library.py +++ b/cratedb_toolkit/info/library.py @@ -170,7 +170,7 @@ class JobInfo: sys.jobs_log; """, transform=get_single_value("job_count"), - description="Total number of queries on this node.", + description="Total number of queries recorded in `sys.jobs_log`.", ) performance15min = InfoElement( name="performance15min", @@ -230,7 +230,7 @@ class JobInfo: MAX((ended::LONG - started::LONG) ) AS max_duration, MIN((ended::LONG - started::LONG) ) AS min_duration, AVG((ended::LONG - started::LONG) ) AS avg_duration, - PERCENTILE((ended::LONG - started::LONG), 0.99) AS p90 + PERCENTILE((ended::LONG - started::LONG), 0.99) AS p99 FROM sys.jobs_log GROUP BY stmt ORDER BY stmt_count DESC From face63d9ef2a017e2b7926372b78f74df45817ff Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Tue, 11 Aug 2026 18:00:45 +0200 Subject: [PATCH 03/12] Update SQL query to use dynamic schema for job statistics retrieval --- cratedb_toolkit/cfr/marimo.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/cratedb_toolkit/cfr/marimo.py b/cratedb_toolkit/cfr/marimo.py index 2cf92909..11b01af0 100644 --- a/cratedb_toolkit/cfr/marimo.py +++ b/cratedb_toolkit/cfr/marimo.py @@ -49,12 +49,16 @@ def _(limitsl): import pandas as pd import sqlalchemy as sa + from cratedb_toolkit.model import DatabaseAddress + sqlalchemy_url = os.getenv("CRATEDB_CLUSTER_URL", "crate://?schema=stats") engine = sa.create_engine(sqlalchemy_url) + + schema = DatabaseAddress.from_string(sqlalchemy_url).schema or "stats" df = pd.read_sql( sql=f""" SELECT stmt, username, query_type, last_used, avg_duration, bucket - FROM stats.jobstats_statements + FROM "{schema}".jobstats_statements ORDER BY last_used, avg_duration DESC LIMIT {limitsl.value}""", con=engine, From eb6af3bf2c411ab9db1a0c829dc62aa70759cedc Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 08:46:55 +0200 Subject: [PATCH 04/12] Introduce new tests for cli --- tests/info/test_cli.py | 80 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 74 insertions(+), 6 deletions(-) diff --git a/tests/info/test_cli.py b/tests/info/test_cli.py index bbfae8ac..f7983cd8 100644 --- a/tests/info/test_cli.py +++ b/tests/info/test_cli.py @@ -5,6 +5,8 @@ from click.testing import CliRunner from cratedb_toolkit.info.cli import cli +from cratedb_toolkit.info.core import JobInfoContainer +from tests.info.test_model import JOB_ELEMENT_NAMES @pytest.fixture @@ -101,9 +103,75 @@ def test_info_jobs(request, runner_factory): assert "meta" in info assert "data" in info - data_keys = list(info["data"].keys()) - assert "by_user" in data_keys - assert "top100_count" in data_keys - assert "top100_duration_individual" in data_keys - assert "top100_duration_total" in data_keys - assert "performance15min" in data_keys + # Both sections describe the very same elements. + assert sorted(info["data"]) == JOB_ELEMENT_NAMES + assert sorted(info["meta"]["elements"]) == JOB_ELEMENT_NAMES + + # The elements which reduce to a single value. + assert isinstance(info["data"]["history_count"], int) + assert isinstance(info["data"]["running_count"], int) + + # The elements which return rows. + assert isinstance(info["data"]["by_user"], list) + assert isinstance(info["data"]["top100_count"], list) + + +def test_info_jobs_history(cratedb, runner_standalone): + """ + Verify `ctk info jobs` reports the statements which have been invoked. + """ + + marker = "ctk-info-jobs-marker" + cratedb.database.run_sql(f"SELECT '{marker}' AS marker") + + result = runner_standalone.invoke(cli, args="jobs", catch_exceptions=False) + assert result.exit_code == 0 + + data = json.loads(result.output)["data"] + + assert data["history_count"] > 0 + assert any(marker in record["stmt"] for record in data["history"]) + + # The query history is reported in chronological order, oldest first. + timestamps = [record["time"] for record in data["history"]] + assert timestamps == sorted(timestamps) + + # The query frequency reports the 99th percentile of the query duration, per statement. + assert sorted(data["top100_count"][0]) == [ + "avg_duration", + "max_duration", + "min_duration", + "p99", + "stmt", + "stmt_count", + ] + + # Durations are reported in milliseconds. + assert sorted(data["top100_duration_individual"][0]) == ["duration", "stmt"] + + +@pytest.mark.parametrize("element_name", JOB_ELEMENT_NAMES) +def test_info_jobs_element(cratedb, element_name): + """ + Verify each element of `ctk info jobs` runs against CrateDB on its own. + + """ + + container = JobInfoContainer(adapter=cratedb.database) + element = container.elements.index[element_name] + container.evaluate_element(element) + + +def test_info_serve(mocker): + """ + Verify `ctk info serve` starts the HTTP service, and hands over its options. + """ + + pytest.importorskip("fastapi") + start = mocker.patch("cratedb_toolkit.info.http.start") + + runner = CliRunner(env={"CRATEDB_CLUSTER_URL": "crate://localhost:4200/"}) + result = runner.invoke(cli, args="serve --listen 0.0.0.0:8042 --reload", catch_exceptions=False) + assert result.exit_code == 0 + + start.assert_called_once_with("0.0.0.0:8042", reload=True) From 474714d6bccf11d95ae80e010f3815c3da7b79d6 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 08:54:27 +0200 Subject: [PATCH 05/12] Improve job statistics implementations with new unit tests and fixtures --- tests/cfr/test_jobstats.py | 273 +++++++++++++++++++++++++++- tests/cfr/test_jobstats_unit.py | 310 ++++++++++++++++++++++++++++++++ 2 files changed, 575 insertions(+), 8 deletions(-) create mode 100644 tests/cfr/test_jobstats_unit.py diff --git a/tests/cfr/test_jobstats.py b/tests/cfr/test_jobstats.py index 37d8afbf..1f8999fe 100644 --- a/tests/cfr/test_jobstats.py +++ b/tests/cfr/test_jobstats.py @@ -1,4 +1,7 @@ +# ruff: noqa: S608 import json +import os +import uuid import pytest from click.testing import CliRunner @@ -8,6 +11,51 @@ pytestmark = pytest.mark.cfr +STATEMENTS_TABLE = f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements" +LAST_TABLE = f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last" + + +@pytest.fixture +def runner(cratedb): + """ + Provide a Click runner which collects into the `testdrive-ext` schema. + """ + return CliRunner(env={"CRATEDB_CLUSTER_URL": cratedb.database.dburi + f"?schema={TESTDRIVE_EXT_SCHEMA}"}) + + +@pytest.fixture(autouse=True) +def reset_collector_state(): + """ + Discard collector state, which lives in module-global variables, between test cases. + """ + from cratedb_toolkit.cfr import jobstats + + jobstats.reset_state() + yield + jobstats.reset_state() + + +def marker_statement(cratedb, label: str) -> str: + """ + Run a uniquely identifiable statement, so it can be found in the collected statistics. + + The marker gets a unique suffix, because `sys.jobs_log` also holds the statements of + previous test runs when the database container is reused. The collector skips + statements which touch `sys.` or `information_schema.`. + """ + marker = f"{label}-{uuid.uuid4().hex[:8]}" + cratedb.database.run_sql(f"SELECT '{marker}' AS marker") + return marker + + +def collected_statements(cratedb, table: str = STATEMENTS_TABLE): + """ + Return all statements from the collected statistics. + """ + cratedb.database.refresh_table(table) + quoted = cratedb.database.quote_relation_name(table) + return [record["stmt"] for record in cratedb.database.run_sql(f"SELECT stmt FROM {quoted}", records=True)] + def test_cfr_jobstats_collect_self(cratedb, caplog): """ @@ -17,6 +65,8 @@ def test_cfr_jobstats_collect_self(cratedb, caplog): # Configure database URI. dburi = cratedb.database.dburi + f"?schema={TESTDRIVE_EXT_SCHEMA}" + marker_statement(cratedb, "jobstats-collect-self-marker") + # Invoke command. runner = CliRunner(env={"CRATEDB_CLUSTER_URL": dburi}) result = runner.invoke( @@ -35,8 +85,9 @@ def test_cfr_jobstats_collect_self(cratedb, caplog): assert {"table_name": "jobstats_last"} in results assert {"table_name": "jobstats_statements"} in results + # How many statements are collected depends on the activity on the cluster. cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") - assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") >= 19 + assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") >= 1 cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") == 1 @@ -45,18 +96,24 @@ def test_cfr_jobstats_collect_self(cratedb, caplog): def test_cfr_jobstats_collect_anonymized(cratedb, caplog): """ Verify `ctk cfr jobstats collect` into the same database, using the `--anonymize` option. + + Without a value, the option uses `decoder_dictionary.json` in the current directory, + so the command runs on an isolated filesystem here. """ # Configure database URI. dburi = cratedb.database.dburi + f"?schema={TESTDRIVE_EXT_SCHEMA}" + marker = marker_statement(cratedb, "jobstats-anonymized-default") + # Invoke command. runner = CliRunner(env={"CRATEDB_CLUSTER_URL": dburi}) - result = runner.invoke( - cli, - args="jobstats collect --once --anonymize", - catch_exceptions=False, - ) + with runner.isolated_filesystem(): + result = runner.invoke( + cli, + args="jobstats collect --once --anonymize", + catch_exceptions=False, + ) assert result.exit_code == 0, result.output # Verify outcome: Log output. @@ -69,11 +126,14 @@ def test_cfr_jobstats_collect_anonymized(cratedb, caplog): assert {"table_name": "jobstats_statements"} in results cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") - assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") >= 19 + assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") >= 1 cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") == 1 + # Verify outcome: No statement has been stored in clear text. + assert not any(marker in stmt for stmt in collected_statements(cratedb)) + def test_cfr_jobstats_collect_reportdb(cratedb, caplog): """ @@ -104,8 +164,9 @@ def test_cfr_jobstats_collect_reportdb(cratedb, caplog): assert {"table_name": "jobstats_last"} in results assert {"table_name": "jobstats_statements"} in results + # How many statements are collected depends on the activity on the cluster. cratedb.database.refresh_table(f"{schema_reportdb}.jobstats_statements") - assert cratedb.database.count_records(f"{schema_reportdb}.jobstats_statements") >= 10 + assert cratedb.database.count_records(f"{schema_reportdb}.jobstats_statements") >= 1 cratedb.database.refresh_table(f"{schema_reportdb}.jobstats_last") assert cratedb.database.count_records(f"{schema_reportdb}.jobstats_last") == 1 @@ -135,3 +196,199 @@ def test_cfr_jobstats_view(cratedb): data_keys = list(info["data"].keys()) assert "stats" in data_keys + + +def test_cfr_jobstats_collect_records_statements(cratedb, runner): + """ + Verify `ctk cfr jobstats collect` records statements verbatim, when not anonymizing. + """ + + marker = marker_statement(cratedb, "jobstats-plain") + + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + assert any(marker in stmt for stmt in collected_statements(cratedb)) + + +def test_cfr_jobstats_collect_anonymize_with_path(cratedb, runner, tmp_path): + """ + Verify `ctk cfr jobstats collect --anonymize` accepts a decoder dictionary path. + + Passing a path used to fail with `Got unexpected extra argument`, because the option + was declared as a boolean flag, while both the help text and the docs advertised a path. + """ + + marker = marker_statement(cratedb, "jobstats-anonymize") + decoder_dictionary = tmp_path / "decoder_dictionary.json" + + result = runner.invoke( + cli, + args=f"jobstats collect --once --anonymize {decoder_dictionary}", + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + # Verify outcome: The marker statement has been stored, but not in clear text. + statements = collected_statements(cratedb) + assert statements + assert not any(marker in stmt for stmt in statements) + + +def test_cfr_jobstats_anonymize_roundtrip(cratedb, runner, tmp_path): + """ + Verify `collect --anonymize` and `view --deanonymize` are inverse operations. + """ + + marker = marker_statement(cratedb, "jobstats-roundtrip") + decoder_dictionary = tmp_path / "decoder_dictionary.json" + + result = runner.invoke( + cli, + args=f"jobstats collect --once --anonymize {decoder_dictionary}", + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + assert not any(marker in stmt for stmt in collected_statements(cratedb)) + + result = runner.invoke( + cli, + args=f"jobstats view --deanonymize {decoder_dictionary}", + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + # Verify outcome: The statement is legible again. + statements = json.loads(result.output)["data"]["stats"] + assert any(marker in stmt for stmt in statements) + + # Verify outcome: Each statement is reported once, not once per anonymization state. + assert len(statements) == len(collected_statements(cratedb)) + + +def test_cfr_jobstats_collect_resumes_from_watermark(cratedb, runner, caplog): + """ + Verify a second `ctk cfr jobstats collect --once` resumes from the recorded watermark. + + The `jobstats_last` table records how far the collector has come. Without honoring it, + a restarted collector counts the same jobs again. + """ + + marker = marker_statement(cratedb, "jobstats-watermark") + + # Collect once, and remember the watermark. + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + cratedb.database.refresh_table(LAST_TABLE) + quoted_last = cratedb.database.quote_relation_name(LAST_TABLE) + first_watermark = cratedb.database.run_sql(f"SELECT last_execution FROM {quoted_last}", records=True)[0][ + "last_execution" + ] + assert first_watermark > 0 + + def marker_calls(): + """ + Return the recorded number of calls of the marker statement. + + Match the statement exactly. The verification queries of this test case are + recorded in `sys.jobs_log`, too, and mention the marker as well. + """ + cratedb.database.refresh_table(STATEMENTS_TABLE) + quoted = cratedb.database.quote_relation_name(STATEMENTS_TABLE) + records = cratedb.database.run_sql(f"SELECT stmt, calls FROM {quoted}", records=True) + return [record["calls"] for record in records if record["stmt"] == f"SELECT '{marker}' AS marker"] + + assert marker_calls() == [1] + + # Collect again. The marker job ran before the watermark, so it must not be counted twice. + caplog.clear() + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + assert f"Resuming from recorded watermark: {first_watermark}" in caplog.text + + assert marker_calls() == [1] + + cratedb.database.refresh_table(LAST_TABLE) + second_watermark = cratedb.database.run_sql(f"SELECT last_execution FROM {quoted_last}", records=True)[0][ + "last_execution" + ] + assert second_watermark > first_watermark + + +def test_cfr_jobstats_view_without_data(cratedb): + """ + Verify `ctk cfr jobstats view` on a schema without collected statistics. + + The command creates its tables on demand, so it reports an empty result instead of failing. + """ + + schema = "testdrive-ext-empty" + runner = CliRunner(env={"CRATEDB_CLUSTER_URL": cratedb.database.dburi + f"?schema={schema}"}) + result = runner.invoke(cli, args="jobstats view", catch_exceptions=False) + assert result.exit_code == 0, result.output + + assert json.loads(result.output)["data"]["stats"] == {} + + cratedb.reset(tables=[f'"{schema}".jobstats_statements', f'"{schema}".jobstats_last']) + + +def test_cfr_jobstats_report(cratedb, runner, mocker): + """ + Verify `ctk cfr jobstats report` reads the schema the statistics were collected into. + """ + + pytest.importorskip("marimo") + + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + captured = {} + + def capture(): + captured["cluster_url"] = os.environ.get("CRATEDB_CLUSTER_URL") + + app_run = mocker.patch("cratedb_toolkit.cfr.marimo.app.run", side_effect=capture) + + result = runner.invoke(cli, args="jobstats report", catch_exceptions=False) + assert result.exit_code == 0, result.output + + app_run.assert_called_once() + assert TESTDRIVE_EXT_SCHEMA in captured["cluster_url"] + + +def test_cfr_jobstats_ui(cratedb, runner, mocker): + """ + Verify `ctk cfr jobstats ui` launches a web server for the collected statistics. + """ + + pytest.importorskip("marimo") + pytest.importorskip("uvicorn") + + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + mocker.patch("marimo.create_asgi_app") + uvicorn_run = mocker.patch("uvicorn.run") + + result = runner.invoke(cli, args="jobstats ui", catch_exceptions=False) + assert result.exit_code == 0, result.output + + uvicorn_run.assert_called_once() + assert uvicorn_run.call_args.kwargs["port"] == 7777 + + +def test_cfr_jobstats_report_without_data(cratedb): + """ + Verify `ctk cfr jobstats report` reports missing statistics in an actionable way. + """ + + pytest.importorskip("marimo") + + runner = CliRunner(env={"CRATEDB_CLUSTER_URL": cratedb.database.dburi + "?schema=testdrive-ext-missing"}) + result = runner.invoke(cli, args="jobstats report") + + assert result.exit_code == 1 + assert isinstance(result.exception, FileNotFoundError) + assert "testdrive-ext-missing" in str(result.exception) + assert "ctk cfr jobstats collect" in str(result.exception) diff --git a/tests/cfr/test_jobstats_unit.py b/tests/cfr/test_jobstats_unit.py new file mode 100644 index 00000000..9e61f097 --- /dev/null +++ b/tests/cfr/test_jobstats_unit.py @@ -0,0 +1,310 @@ +""" +Unit tests for the job statistics collector, which do not need a database. + +The collector accumulates its statistics in Python, not in SQL, so the bucket +assignment and the average computation are verified here explicitly. +""" + +import pytest + +from cratedb_toolkit.cfr import jobstats +from cratedb_toolkit.model import DatabaseAddress + +pytestmark = pytest.mark.cfr + + +@pytest.fixture(autouse=True) +def reset_collector_state(): + """ + Provide each test case with a pristine collector, and leave no state behind. + """ + jobstats.reset_state() + jobstats.anonymize_sql = False + jobstats.deanonymize_sql = False + yield + jobstats.reset_state() + jobstats.anonymize_sql = False + jobstats.deanonymize_sql = False + + +def job(started=1_000, duration=50, stmt="SELECT 1", query_type="SELECT", username="crate", node_name="node-1"): + """ + Produce a single `sys.jobs_log` record, in the shape `scrape_db` yields it. + """ + return (started, started + duration, {"type": query_type}, stmt, username, {"id": "n1", "name": node_name}) + + +# Bucket assignment. + + +@pytest.mark.parametrize( + ("duration", "expected"), + [ + (0, "10"), + (9, "10"), + (10, "50"), + (49, "50"), + (50, "100"), + (1_999, "2000"), + (19_999, "20000"), + (20_000, "INF"), + (999_999, "INF"), + ], +) +def test_assign_to_bucket(duration, expected): + """ + Verify durations land in the bucket of the next larger threshold. Thresholds are exclusive. + """ + bucket = dict(jobstats.bucket_dict) + outcome = jobstats.assign_to_bucket(bucket, duration) + assert outcome[expected] == 1 + assert sum(outcome.values()) == 1 + + +def test_assign_to_bucket_mutates_in_place(): + """ + Verify `assign_to_bucket` updates and returns the very same dictionary. + """ + bucket = dict(jobstats.bucket_dict) + assert jobstats.assign_to_bucket(bucket, 5) is bucket + + +def test_bucket_dict_matches_bucket_list(): + """ + Verify the bucket thresholds and the bucket template do not drift apart. + """ + assert list(jobstats.bucket_dict) == [str(threshold) for threshold in jobstats.bucket_list] + ["INF"] + + +# Statistics accumulation. + + +def test_update_statistics_new_statement(): + """ + Verify a previously unseen statement is recorded with a single call. + """ + jobstats.update_statistics([job(started=1_000, duration=50)]) + + assert list(jobstats.sys_jobs_log) == ["SELECT 1"] + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["calls"] == 1 + assert entry["type"] == "SELECT" + assert entry["user"] == "crate" + assert entry["last_used"] == 1_000 + assert entry["avg_duration"] == 50 + assert entry["bucket"]["100"] == 1 + assert entry["in_db"] is False + assert entry["changed"] is True + assert entry["id"] + + +def test_update_statistics_repeated_statement(): + """ + Verify repeated statements accumulate, and that `avg_duration` is a decaying average. + + The collector computes `(previous + current) / 2` per sample, which weights recent + executions much more heavily than an arithmetic mean would. + """ + jobstats.update_statistics([job(duration=50)]) + jobstats.update_statistics([job(started=2_000, duration=100)]) + + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["calls"] == 2 + assert entry["last_used"] == 2_000 + assert entry["avg_duration"] == 75.0 + assert entry["bucket"]["100"] == 1 + assert entry["bucket"]["500"] == 1 + + +def test_update_statistics_deduplicates_nodes(): + """ + Verify each node is recorded only once per statement, no matter how often it runs there. + """ + jobstats.update_statistics( + [ + job(node_name="node-1"), + job(node_name="node-1"), + job(node_name="node-2"), + ] + ) + + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["calls"] == 3 + assert len(entry["nodes"]) == 2 + + +def test_update_statistics_separates_statements(): + """ + Verify statistics are keyed by statement. + """ + jobstats.update_statistics([job(stmt="SELECT 1"), job(stmt="SELECT 2", query_type="SELECT")]) + + assert sorted(jobstats.sys_jobs_log) == ["SELECT 1", "SELECT 2"] + assert jobstats.sys_jobs_log["SELECT 1"]["calls"] == 1 + assert jobstats.sys_jobs_log["SELECT 2"]["calls"] == 1 + + +# State handling. + + +def test_reset_state(): + """ + Verify `reset_state` discards accumulated statistics. + + Without it, a second `boot()` in the same process would consider statements to be + stored already, and update rows which do not exist. + """ + jobstats.update_statistics([job()]) + jobstats.last_execution_ts = 42 + assert jobstats.sys_jobs_log + + jobstats.reset_state() + + assert jobstats.sys_jobs_log == {} + assert jobstats.last_execution_ts == 0 + + +def test_init_stmts_keeps_accumulated_statistics(): + """ + Verify reading rows from the database does not clobber freshly accumulated statistics. + """ + jobstats.update_statistics([job(duration=50)]) + jobstats.init_stmts([("id-1", "SELECT 1", 99, dict(jobstats.bucket_dict), "crate", "SELECT", 1.0, [], 1_000)]) + + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["calls"] == 1 + assert entry["in_db"] is False + + +def test_init_stmts_adopts_database_rows(): + """ + Verify rows read from the database are adopted as already persisted. + """ + jobstats.init_stmts([("id-1", "SELECT 1", 99, dict(jobstats.bucket_dict), "crate", "SELECT", 1.5, ["n1"], 1_000)]) + + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["id"] == "id-1" + assert entry["calls"] == 99 + assert entry["avg_duration"] == 1.5 + assert entry["in_db"] is True + assert entry["changed"] is False + + +# Anonymization. + + +def test_anonymize_statement_disabled(): + """ + Verify statements pass through unchanged while anonymization is disabled. + """ + assert jobstats.anonymize_statement("SELECT * FROM foobar") == "SELECT * FROM foobar" + + +def test_deanonymize_statement_disabled(): + """ + Verify statements pass through unchanged while deanonymization is disabled. + """ + assert jobstats.deanonymize_statement("SELECT * FROM foobar") == "SELECT * FROM foobar" + + +def test_redacted_statement(): + """ + Verify redaction discloses nothing but remains stable per statement. + """ + redacted = jobstats.redacted_statement("SELECT * FROM secrets") + assert redacted == jobstats.redacted_statement("SELECT * FROM secrets") + assert redacted != jobstats.redacted_statement("SELECT * FROM other_secrets") + assert "secrets" not in redacted + assert redacted.startswith(" Date: Wed, 12 Aug 2026 08:56:40 +0200 Subject: [PATCH 06/12] Add unit tests for job information elements and containers --- tests/info/test_model.py | 206 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) create mode 100644 tests/info/test_model.py diff --git a/tests/info/test_model.py b/tests/info/test_model.py new file mode 100644 index 00000000..d08fe449 --- /dev/null +++ b/tests/info/test_model.py @@ -0,0 +1,206 @@ +""" +Unit tests for the information elements and containers. + +`ctk info jobs` and friends are assembled from `InfoElement` instances, and the element +names are the keys of the emitted JSON document. +""" + +import io +import typing as t +from pathlib import Path + +import pytest + +from cratedb_toolkit.info.core import InfoContainer, JobInfoContainer, LogContainer +from cratedb_toolkit.info.library import Library +from cratedb_toolkit.info.model import ElementStore, InfoContainerBase, InfoElement, LogElement +from cratedb_toolkit.info.util import get_single_value +from cratedb_toolkit.util.database import DatabaseAdapter + +# The output keys of `ctk info jobs`. Changing them breaks consumers of the JSON document. +JOB_ELEMENT_NAMES = [ + "age_range", + "by_user", + "duration_buckets", + "duration_percentiles", + "history", + "history_count", + "performance15min", + "running", + "running_count", + "top100_count", + "top100_duration_individual", + "top100_duration_total", +] + + +class FakeAdapter(DatabaseAdapter): + """ + Record SQL statements instead of running them, and reply with a canned response. + + Deliberately does not invoke `DatabaseAdapter.__init__` to not connect DB. + """ + + def __init__(self, response=None): + self.response = response if response is not None else [{"job_count": 42}] + self.queries: t.List[str] = [] + + def run_sql( + self, + sql: t.Union[str, Path, io.IOBase], + parameters: t.Optional[t.Mapping[str, str]] = None, + records: bool = False, + ignore: t.Optional[str] = None, + ): + self.queries.append(str(sql)) + return self.response + + +def test_job_info_container_elements(): + """ + Verify `ctk info jobs` emits exactly the well-known set of elements. + """ + + container = JobInfoContainer(adapter=FakeAdapter()) + assert sorted(container.elements.index) == JOB_ELEMENT_NAMES + assert len(container.elements.items) == len(JOB_ELEMENT_NAMES) + + +def test_job_info_element_names_differ_from_attributes(): + """ + Verify the `history100` attribute is emitted as `history`. + """ + + assert Library.JobInfo.history100.name == "history" + + +def test_job_info_elements_are_documented(): + """ + Verify each job element carries a label and a description, for the docs to stay truthful. + """ + container = JobInfoContainer(adapter=FakeAdapter()) + for element in container.elements.items: + assert element.label, f"Element without label: {element.name}" + assert element.description, f"Element without description: {element.name}" + assert element.sql.strip(), f"Element without SQL: {element.name}" + + +@pytest.mark.parametrize( + ("container_factory", "count"), + [(InfoContainer, 18), (JobInfoContainer, 12), (LogContainer, 1)], + ids=["cluster", "jobs", "logs"], +) +def test_container_element_registration(container_factory, count): + """ + Verify all containers register their elements without name collisions. + """ + container = container_factory(adapter=FakeAdapter()) + assert len(container.elements.items) == count + assert len(container.elements.index) == count + + +def test_element_store_rejects_duplicates(): + """ + Verify duplicate element names are refused, because they would shadow each other in the output. + """ + store = ElementStore() + element = InfoElement(name="foo", label="Foo", sql="SELECT 1;") + store.add(element) + with pytest.raises(KeyError) as ex: + store.add(element) + assert "Duplicate key/label: foo" in str(ex.value) + + +def test_info_element_to_dict(): + """ + Verify element serialization, as emitted per the `meta.elements` section. + """ + element = InfoElement(name="foo", label="Foo", sql=" SELECT 1; ", description="Foo element", unit="ms") + data = element.to_dict() + assert data["name"] == "foo" + assert data["label"] == "Foo" + assert data["sql"] == "SELECT 1;" + assert data["description"] == "Foo element" + assert data["unit"] == "ms" + assert data["transform"] == "None" + + +def test_log_element_limit_templating(): + """ + Verify `LogElement` interpolates its row limit into the SQL statement. + """ + adapter = FakeAdapter() + element = LogElement(name="foo", label="Foo", sql="SELECT 1 LIMIT {limit};", limit=42) + container = LogContainer(adapter=adapter) + container.evaluate_element(element) + assert adapter.queries == ["SELECT 1 LIMIT 42;"] + + +def test_evaluate_element_applies_transform(): + """ + Verify element transformations are applied to the SQL result. + """ + adapter = FakeAdapter(response=[{"job_count": 42}]) + element = InfoElement(name="foo", label="Foo", sql="SELECT 1;", transform=get_single_value("job_count")) + container = JobInfoContainer(adapter=adapter) + assert container.evaluate_element(element) == 42 + + +def test_history_transform_reverses_rows(): + """ + Verify the query history is emitted in chronological order, oldest first. + + The SQL statement selects the most recent jobs first, so the rows are reversed. + """ + rows = [{"time": 3}, {"time": 2}, {"time": 1}] + transform = Library.JobInfo.history100.transform + assert transform is not None + assert transform(rows) == [{"time": 1}, {"time": 2}, {"time": 3}] + + +def test_get_single_value(): + """ + Verify scalar reduction of a single-row, single-column SQL result. + """ + assert get_single_value("job_count")([{"job_count": 42}]) == 42 + + +def test_container_document_shape(): + """ + Verify the `meta`/`data` document shape, and that both sections describe the same elements. + """ + container = JobInfoContainer(adapter=FakeAdapter()) + document = container.to_dict() + + assert sorted(document) == ["data", "meta"] + assert sorted(document["meta"]) == ["application_name", "application_version", "elements", "system_time"] + assert sorted(document["meta"]["elements"]) == JOB_ELEMENT_NAMES + assert sorted(document["data"]) == JOB_ELEMENT_NAMES + + +def test_container_meta_includes_sql(): + """ + Verify the emitted metadata echoes the SQL statement of each element, so users can re-run it. + """ + container = JobInfoContainer(adapter=FakeAdapter()) + elements = container.to_dict()["meta"]["elements"] + assert elements["running_count"]["sql"].startswith("SELECT") + assert "sys.jobs" in elements["running_count"]["sql"] + + +def test_container_by_table_not_implemented(): + """ + Verify the unimplemented per-table inquiry reports itself as such. + """ + container = JobInfoContainer(adapter=FakeAdapter()) + with pytest.raises(NotImplementedError): + container.by_table(schema="doc", table="foo") + + +def test_container_base_needs_builtins(): + """ + Verify containers must register their elements. + """ + with pytest.raises(NotImplementedError) as ex: + InfoContainerBase(adapter=FakeAdapter()) + assert ex.match("Method needs to be implemented by child class") From e01a1df5bd665db5067bf9b1963b091e99453c72 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 09:06:01 +0200 Subject: [PATCH 07/12] Consolidate job statistics implementations and enhance state management --- cratedb_toolkit/cfr/jobstats.py | 109 ++++++++++++++++++++++++-------- tests/info/test_model.py | 2 +- 2 files changed, 83 insertions(+), 28 deletions(-) diff --git a/cratedb_toolkit/cfr/jobstats.py b/cratedb_toolkit/cfr/jobstats.py index 2814e9db..763030f7 100644 --- a/cratedb_toolkit/cfr/jobstats.py +++ b/cratedb_toolkit/cfr/jobstats.py @@ -2,11 +2,14 @@ # Distributed under the terms of the AGPLv3 license, see LICENSE. # ruff: noqa: S608 +import hashlib +import io import json import logging import os import time import typing as t +from contextlib import redirect_stdout from uuid import uuid4 import urllib3 @@ -19,9 +22,11 @@ TRACING = False +# How far back to look for jobs when no watermark has been recorded yet, in seconds. +INITIAL_LOOKBACK_SECONDS = 600 last_execution_ts = 0 -sys_jobs_log = {} +sys_jobs_log: t.Dict[str, t.Dict[str, t.Any]] = {} bucket_list = [10, 50, 100, 500, 1000, 2000, 5000, 10000, 15000, 20000] bucket_dict = { "10": 0, @@ -45,7 +50,21 @@ interval: float anonymize_sql: bool = False deanonymize_sql: bool = False # Added global flag for deanonymization -decoder_dict_path: str +decoder_dict_path: str = "" + + +def reset_state(): + """ + Discard all accumulated in-memory statistics. + + The collector keeps its statistics in module-global state. Without resetting it, + a second `boot()` within the same process starts off with statements flagged as + already stored (`in_db=True`), which makes `write_stats_to_db` issue an `UPDATE` + against a table that has no such row, silently dropping the statistics. + """ + global last_execution_ts + last_execution_ts = 0 + sys_jobs_log.clear() def boot( @@ -66,6 +85,7 @@ def boot( anonymize_sql, \ deanonymize_sql, \ decoder_dict_path + reset_state() anonymize_sql = anonymize_statements deanonymize_sql = deanonymize_statements @@ -83,6 +103,7 @@ def boot( logger.info(f"SQL deanonymization is enabled, using dictionary: {decoder_dict_path}") interval = float(os.getenv("INTERVAL", 10)) + initial_lookback = float(os.getenv("INITIAL_LOOKBACK_SECONDS", INITIAL_LOOKBACK_SECONDS)) stmt_log_table = os.getenv("STMT_TABLE", f'"{schema}".jobstats_statements') last_exec_table = os.getenv("LAST_EXEC_TABLE", f'"{schema}".jobstats_last') @@ -116,24 +137,45 @@ def boot( # If no separate reporting DB, use the same cursor for both report_cursor = cursor - last_scrape = int(time.time() * 1000) - int(interval * 60000) - dbinit() + # Resume from the recorded watermark, so a restarted collector neither re-counts jobs + # it has already processed, nor misses jobs which ran while it was not running. + if isinstance(last_execution_ts, (int, float)) and last_execution_ts > 0: + last_scrape = int(last_execution_ts) + logger.info(f"Resuming from recorded watermark: {last_scrape}") + else: + last_scrape = int(time.time() * 1000) - int(initial_lookback * 1000) + logger.info(f"No watermark recorded yet, looking back {initial_lookback} seconds") + + +def redacted_statement(statement: str) -> str: + """ + Return a placeholder for a statement which could not be anonymized. + """ + digest = hashlib.sha256(statement.encode("utf-8")).hexdigest()[:16] + return f"" + def anonymize_statement(statement: str) -> str: - """Anonymize SQL statement using queryanonymizer.""" - if anonymize_sql and anonymize is not None: + """ + Anonymize SQL statement using queryanonymizer. + + When anonymization fails, the statement is redacted rather than stored in clear text. + """ + if not anonymize_sql: + return statement + try: + # Load dictionary file each time + encoder_dict = {} try: - # Load dictionary file each time - encoder_dict = {} - try: - with open(decoder_dict_path, "r") as f: - encoder_dict = json.load(f) - except (FileNotFoundError, json.JSONDecodeError) as e: - logger.warning(f"Could not load encoder dictionary: {e}") - - # Call anonymize and extract only the anonymized statement (first item) + with open(decoder_dict_path, "r") as f: + encoder_dict = json.load(f) + except (FileNotFoundError, json.JSONDecodeError) as e: + logger.warning(f"Could not load encoder dictionary: {e}") + + # Call anonymize and extract only the anonymized statement (first item). + with redirect_stdout(io.StringIO()): result = anonymize( query=statement, keywords_group="SQL", @@ -143,13 +185,13 @@ def anonymize_statement(statement: str) -> str: path_to_decoder_dictionary_file=decoder_dict_path, custom_encoder_dictionary=encoder_dict, ) - # Return only the anonymized statement string - if isinstance(result, tuple) and len(result) > 0: - return result[0] - return result - except Exception as e: - logger.warning(f"Failed to anonymize statement: {e}") - return statement + # Return only the anonymized statement string + if isinstance(result, tuple) and len(result) > 0: + return result[0] + return result + except Exception as e: + logger.warning(f"Failed to anonymize statement, redacting it instead: {e}") + return redacted_statement(statement) def deanonymize_statement(statement: str) -> str: @@ -161,11 +203,16 @@ def deanonymize_statement(statement: str) -> str: except (FileNotFoundError, json.JSONDecodeError) as e: logger.warning(f"Could not load decoder dictionary: {e}") - # Call anonymize to decode the statement - result = deanonymize( - statement, - path_to_decoder_dictionary_file=decoder_dict_path, - ) + # Call anonymize to decode the statement, again discarding its output on stdout. + try: + with redirect_stdout(io.StringIO()): + result = deanonymize( + statement, + path_to_decoder_dictionary_file=decoder_dict_path, + ) + except Exception as e: + logger.warning(f"Failed to deanonymize statement: {e}") + return statement # Return only the deanonymized statement string if isinstance(result, tuple) and len(result) > 0: @@ -181,11 +228,14 @@ def dbinit(): f"username TEXT, query_type TEXT, avg_duration FLOAT, nodes ARRAY(TEXT))" ) report_cursor.execute(stmt) + # Refresh before reading, so statistics written by a previous run are seen. + report_cursor.execute(f"REFRESH TABLE {stmt_log_table}") stmt = f"SELECT id, stmt, calls, bucket, username, query_type, avg_duration, nodes, last_used FROM {stmt_log_table}" report_cursor.execute(stmt) init_stmts(report_cursor.fetchall()) stmt = f"CREATE TABLE IF NOT EXISTS {last_exec_table} (last_execution TIMESTAMP)" report_cursor.execute(stmt) + report_cursor.execute(f"REFRESH TABLE {last_exec_table}") stmt = f"SELECT last_execution FROM {last_exec_table}" report_cursor.execute(stmt) init_last_execution(report_cursor.fetchall()) @@ -197,6 +247,8 @@ def init_last_execution(last_execution): last_execution_ts = 0 stmt = f"INSERT INTO {last_exec_table} (last_execution) VALUES (?)" report_cursor.execute(stmt, (0,)) + # Refresh, so the `UPDATE` of `write_stats_to_db` finds the record just inserted. + report_cursor.execute(f"REFRESH TABLE {last_exec_table}") else: last_execution_ts = last_execution[0][0] @@ -296,6 +348,9 @@ def read_stats(): deanonymized_results.append(tuple(row_list)) results = deanonymized_results + # The records just read are authoritative. Without discarding what `dbinit` has read + # before, deanonymized statements would be reported next to their anonymized form. + sys_jobs_log.clear() init_stmts(results) return sys_jobs_log diff --git a/tests/info/test_model.py b/tests/info/test_model.py index d08fe449..e4da4d6b 100644 --- a/tests/info/test_model.py +++ b/tests/info/test_model.py @@ -60,7 +60,7 @@ def test_job_info_container_elements(): """ Verify `ctk info jobs` emits exactly the well-known set of elements. """ - + container = JobInfoContainer(adapter=FakeAdapter()) assert sorted(container.elements.index) == JOB_ELEMENT_NAMES assert len(container.elements.items) == len(JOB_ELEMENT_NAMES) From d81a50a4aa1808a55c0398be3f778afafc73f575 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 09:24:31 +0200 Subject: [PATCH 08/12] Improve docs for job statistics and cluster information commands --- doc/cfr/index.md | 3 +- doc/cfr/jobstats.md | 136 +++++++++++++++++++++++++++++++++++++---- doc/info/index.md | 146 ++++++++++++++++++++++++++++++++++---------- 3 files changed, 241 insertions(+), 44 deletions(-) diff --git a/doc/cfr/index.md b/doc/cfr/index.md index 61651321..69e4a7c7 100644 --- a/doc/cfr/index.md +++ b/doc/cfr/index.md @@ -7,7 +7,8 @@ information collection and recording per `ctk cfr`. The three areas differ in how raw vs. interpreted their output is: - `sys-export` / `sys-import` — a true raw copy of every `sys.*` table. See {ref}`cfr-systable`. - `jobstats collect` / `view` — raw, time-series query statistics, persisted to a schema. - `report` / `ui` additionally launch an interpreted. See {ref}`cfr-jobstats`. + `report` / `ui` additionally launch an interpreted, interactive dashboard over the same + collected data. See {ref}`cfr-jobstats`. - `info record` — a raw snapshot of `ctk info cluster` and `ctk info jobs`. See {ref}`cfr-info`. ```{toctree} diff --git a/doc/cfr/jobstats.md b/doc/cfr/jobstats.md index e01ce656..53e19190 100644 --- a/doc/cfr/jobstats.md +++ b/doc/cfr/jobstats.md @@ -1,37 +1,149 @@ (cfr-jobstats)= # Job statistics collector -Collect and display job statistics. This is a separate, continuously-collected time series -of query statistics, distinct from the one-shot `ctk info jobs` snapshot. -See {ref}`cluster-info`. +Collect query statistics from `sys.jobs_log` continuously, and keep them beyond the +retention of that table. `collect` and `view` handle the raw statistics, `report` and `ui` +launch an interpreted, interactive dashboard on top of the same collected data. -`collect` and `view` return raw, time-series JSON persisted to the `stats` schema. `report` -and `ui` instead launch an interpreted, interactive dashboard on top of the same collected -data. +This is distinct from the one-shot `ctk info jobs` snapshot. For a side-by-side comparison +of both, see {ref}`jobs-vs-jobstats`. +## Install +```shell +pip install --upgrade 'cratedb-toolkit[cfr]' +``` +:::{tip} +Alternatively, use the Docker image per `ghcr.io/crate/cratedb-toolkit`. +For more information about installing CrateDB Toolkit, see {ref}`install`. +::: + +## Synopsis + +The collector stores its statistics in the schema of the cluster URL, `stats` by default. ```shell export CRATEDB_CLUSTER_URL=crate://crate@localhost:4200/?schema=stats ``` + +Collect statistics, then display or explore them. ```shell ctk cfr jobstats collect +``` +```shell ctk cfr jobstats view +``` +```shell ctk cfr jobstats report +``` +```shell ctk cfr jobstats ui ``` -Note: Please collect statistics first using `ctk cfr jobstats collect`, -then use the other commands to display or explore them. +:::{note} +Please collect statistics first using `ctk cfr jobstats collect`, then use the other +commands to display or explore them. `view` creates its tables on demand, so an empty +result means nothing has been collected into that schema yet. +::: + +## How collection works + +`collect` polls `sys.jobs_log` for jobs which finished since the last poll, and folds them +into per-statement statistics. Statements against `sys.*` and `information_schema.*` are +skipped, so the collector does not account for its own queries. + +How far the collector has come is recorded as a watermark, so a restarted collector picks +up where it left off, instead of counting the same jobs again. + +Per distinct statement, the collector maintains: +- `calls` — how often the statement has been executed +- `bucket` — a histogram of the execution durations, in milliseconds. A duration is counted + into the first bucket whose threshold it stays below. The thresholds are 10, 50, 100, 500, + 1000, 2000, 5000, 10000, 15000, and 20000, plus `INF` for everything slower. +- `avg_duration` — a *decaying* average, updated as `(previous + current) / 2` per + execution. Recent executions therefore weigh much more heavily than an arithmetic mean + over all executions would. +- `nodes` — the nodes which have run the statement, without duplicates +- `last_used` — when the statement was last seen +- `username`, `query_type` — as reported by CrateDB + +## Tables + +Two tables are created in the configured schema. + +`"".jobstats_statements` holds one record per distinct statement: + +| Column | Type | Description | +|---|---|---| +| `id` | `TEXT` | identifier assigned by the collector | +| `stmt` | `TEXT` | the statement, anonymized when collected with `--anonymize` | +| `calls` | `INT` | number of executions counted so far | +| `bucket` | `OBJECT` | duration histogram, keyed by threshold | +| `last_used` | `TIMESTAMP` | when the statement was last seen | +| `username` | `TEXT` | user which ran the statement | +| `query_type` | `TEXT` | statement classification reported by CrateDB | +| `avg_duration` | `FLOAT` | decaying average duration, in milliseconds | +| `nodes` | `ARRAY(TEXT)` | nodes which have run the statement, as JSON strings of the `node` object of `sys.jobs_log` | + +`"".jobstats_last` holds a single record, the watermark: + +| Column | Type | Description | +|---|---|---| +| `last_execution` | `TIMESTAMP` | up to when jobs have been collected | + +## Configuration + +The cluster address is taken from `--cluster-url` / `CRATEDB_CLUSTER_URL`, and the schema +from its `?schema=` parameter, defaulting to `stats`. Additionally, these environment +variables are recognized. + +- `INTERVAL` — how long to sleep between two collection cycles, in seconds. Default: `10`. +- `INITIAL_LOOKBACK_SECONDS` — how far back to look for jobs when no watermark has been + recorded yet, in seconds. Default: `600`. +- `STMT_TABLE` — full-qualified name of the statistics table, overriding the default + `"".jobstats_statements`. +- `LAST_EXEC_TABLE` — full-qualified name of the watermark table, overriding the default + `"".jobstats_last`. :::{rubric} Options ::: `ctk cfr jobstats collect`: - `--once` — record only one sample, then exit, instead of collecting continuously -- `--reportdb` / `-r` — a separate database URL to store report data in - (`crate://crate@localhost:4200/?sslmode=require`) +- `--reportdb` / `-r` — a separate database URL to store the statistics in + (`crate://crate@localhost:4200/?schema=stats&sslmode=require`). Jobs are read from the + cluster URL, and written to this one. - `--anonymize` — path to a decoder dictionary file for anonymizing SQL statements before - they're stored; using the flag without a value defaults to `decoder_dictionary.json` + they are stored; using the flag without a value defaults to `decoder_dictionary.json` in + the current working directory `ctk cfr jobstats view`: -- `--reportdb` / `-r` — a separate database URL to read report data from +- `--reportdb` / `-r` — a separate database URL to read the statistics from - `--deanonymize` — path to the decoder dictionary file used to reverse `--anonymize`, to view statements in their original form + +`ctk cfr jobstats report` and `ctk cfr jobstats ui` read the statistics from the same +schema `collect` wrote them to. `ui` serves the dashboard on `localhost:7777`. + +## Anonymization + +With `--anonymize`, statements are anonymized before they are stored, and the substitutions +are recorded in the decoder dictionary file. Keep that file: it is the only way to make the +collected statements legible again, using `view --deanonymize`. + +```shell +ctk cfr jobstats collect --once --anonymize ./decoder_dictionary.json +``` +```shell +ctk cfr jobstats view --deanonymize ./decoder_dictionary.json +``` + +:::{warning} +The decoder dictionary maps anonymized tokens back to the original identifiers and string +literals. Treat it as confidential, and do not ship it together with the collected +statistics. +::: + +:::{note} +Anonymization fails closed: when a statement cannot be anonymized, it is stored as +`>` rather than in clear text. Statistics per distinct statement remain +meaningful, but such statements cannot be recovered with `--deanonymize`. The event is +reported as a warning. +::: diff --git a/doc/info/index.md b/doc/info/index.md index bdf70b99..92b100a2 100644 --- a/doc/info/index.md +++ b/doc/info/index.md @@ -8,8 +8,8 @@ A bundle of information inquiry utilities, for diagnostics and more. - `ctk info cluster` / `ctk info jobs` — a curated, one-shot snapshot of hand-picked health, shard, and query metrics. Good for a quick look at what's going on right now. - `ctk cfr info record` — the same snapshot, persisted over time. See {ref}`cfr-info`. -- `ctk cfr jobstats collect` / `view` — a separate, continuously-collected time series of - query statistics, not the same data as `ctk info jobs`. See {ref}`cfr-jobstats`. +- `ctk cfr jobstats collect` / `view` — query statistics accumulated continuously over time. + See {ref}`jobs-vs-jobstats` for how it compares to `ctk info jobs`. - `ctk cfr sys-export` — a true raw dump of every system table, no interpretation. This is the one to reach for when collecting diagnostics for a CrateDB support case. See {ref}`cfr-systable`. @@ -28,18 +28,24 @@ For more information about installing CrateDB Toolkit, see {ref}`install`. Define CrateDB database cluster address per command-line option. Choose one of both alternatives. ```shell -ctk cfr --cluster-url "https://username:password@localhost:4200/?schema=ext" jobstats collect +ctk info --cluster-url "https://username:password@localhost:4200/" jobs ``` ```shell -ctk cfr --cluster-url "crate://username:password@localhost:4200/?schema=ext&ssl=true" jobstats collect +ctk info --cluster-url "crate://username:password@localhost:4200/?ssl=true" jobs ``` Define CrateDB database cluster address per environment variable. Choose one of both alternatives. ```shell -export CRATEDB_CLUSTER_URL=https://username:password@localhost:4200/?schema=ext +export CRATEDB_CLUSTER_URL=https://username:password@localhost:4200/ ``` ```shell -export CRATEDB_CLUSTER_URL=crate://username:password@localhost:4200/?schema=ext&ssl=true +export CRATEDB_CLUSTER_URL=crate://username:password@localhost:4200/?ssl=true +``` + +On CrateDB Cloud, address the cluster by name or by identifier instead, using +`--cluster-name` / `CRATEDB_CLUSTER_NAME`, or `--cluster-id` / `CRATEDB_CLUSTER_ID`. +```shell +ctk info --cluster-name hotzenplotz jobs ``` :::{note} For some commands, both options might not be available yet, just one of them. @@ -82,31 +88,9 @@ Shards: - **Total uncommitted translog size** — a large uncommitted total can indicate issues with shard replication -Display database cluster job information: a one-shot, ad hoc snapshot, not persisted over -time. Contrast with `ctk cfr jobstats collect`, which collects the same kind of information -continuously, see {ref}`cfr-jobstats`. -```shell -ctk info jobs -``` - -:::{rubric} Elements -::: -- **Query age range** — timestamps of first and last job -- **Queries by user** — total number of queries per user -- **Query Duration Distribution (Buckets)** — distribution of query durations, bucketed -- **Query Duration Distribution (Percentiles)** — distribution of query durations, percentiles -- **Query History** — statements and durations of the 100 most recent queries / jobs -- **Query History Count** — total number of queries on this node -- **Query performance 15min** — query performance within the last 15 minutes: queries per - second, and query speed (ms) -- **Currently Running Queries** — statements and durations of currently running queries / jobs -- **Number of running queries** — total number of currently running queries -- **Query frequency** — the 100 most frequent queries -- **Individual Query Duration** — the 100 queries by individual duration (ms) -- **Total Query Duration** — the 100 queries by total duration (ms) - -Display database cluster log messages: a raw, limited passthrough of the most recent +Display database cluster log messages: a raw, limited passthrough of the 100 most recent `sys.jobs_log` rows, filtered to exclude queries against `sys.*`/`information_schema.*`. +The row limit is not adjustable. ```shell ctk info logs ``` @@ -119,6 +103,100 @@ ctk tail -n 3 sys.jobs_log ``` +## Job information + +Display database cluster job information: a one-shot, ad hoc snapshot, computed on the +spot and not persisted anywhere. +```shell +ctk info jobs +``` + +Every element is an individual SQL statement against `sys.jobs_log` (jobs which have +finished) or `sys.jobs` (jobs which are still running), evaluated when you invoke the +command. All durations are reported in milliseconds. + +:::{rubric} Elements +::: +| Output key | Label | Source | Result | +|---|---|---|---| +| `age_range` | Query age range | `sys.jobs_log` | `first_job`, `last_job` | +| `by_user` | Queries by user | `sys.jobs_log` | `username`, `count` per user | +| `duration_buckets` | Query Duration Distribution (Buckets) | `sys.jobs_log` | `bucket`, `count`, `duration` per percentile bucket | +| `duration_percentiles` | Query Duration Distribution (Percentiles) | `sys.jobs_log` | `min`, `p50`, `p90`, `p99`, `max` | +| `history` | Query History | `sys.jobs_log` | the 100 most recent jobs, oldest first: `time`, `stmt`, `duration`, `username` | +| `history_count` | Query History Count | `sys.jobs_log` | single value: total number of recorded jobs | +| `performance15min` | Query performance 15min | `sys.jobs_log` | per 10-second interval and query type: `qps`, `duration` | +| `running` | Currently Running Queries | `sys.jobs` | `time`, `stmt`, `duration`, `username` | +| `running_count` | Number of running queries | `sys.jobs` | single value: number of running jobs | +| `top100_count` | Query frequency | `sys.jobs_log` | the 100 most frequent statements: `stmt`, `stmt_count`, `min_duration`, `max_duration`, `avg_duration`, `p99` | +| `top100_duration_individual` | Individual Query Duration | `sys.jobs_log` | the 100 slowest single executions: `duration`, `stmt` | +| `top100_duration_total` | Total Query Duration | `sys.jobs_log` | the 100 statements with the highest total duration: `total_duration`, `stmt`, `stmt_count` | + +`history` and `running` omit statements mentioning `snapshot`, to keep backup activity out +of the query history. + +:::{rubric} Output +::: +The command emits a single JSON document with two sections. `data` holds one entry per +element, keyed by its output key. `meta` describes the elements, including the SQL +statement each one ran, so you can re-run an individual element by hand. +```json +{ + "meta": { + "system_time": "2026-08-11T12:00:00.000000", + "application_name": "CrateDB Toolkit", + "application_version": "0.0.0", + "elements": { + "running_count": { + "name": "running_count", + "label": "Number of running queries", + "sql": "SELECT\n COUNT(*) AS job_count\nFROM\n sys.jobs;", + "description": "Total number of currently running queries.", + "transform": "functools.partial(...)", + "unit": null + } + } + }, + "data": { + "running_count": 1, + "by_user": [{"username": "crate", "count": 42}] + } +} +``` + +:::{note} +`sys.jobs_log` is a bounded, in-memory record of finished jobs. It is governed by the +`stats.enabled`, `stats.jobs_log_size`, and `stats.jobs_log_expiration` cluster settings, +and it does not survive a node restart. With statistics disabled, `ctk info jobs` reports +empty results. Because the table only ever holds a recent window, a snapshot cannot answer +questions about last week — that is what the continuous collector is for. +::: + +(jobs-vs-jobstats)= +### Comparison with `ctk cfr jobstats` + +Both commands report on the jobs of a cluster, but they are separate implementations with +different purposes. See {ref}`cfr-jobstats` for the collector. + +| | `ctk info jobs` | `ctk cfr jobstats collect` / `view` | +|---|---|---| +| Mode | one shot, stateless | polls the cluster continuously, `--once` for a single sample | +| Source | `sys.jobs_log` and `sys.jobs` | `sys.jobs_log` | +| Statements considered | all; `history` and `running` skip `snapshot` statements | skips statements against `sys.*` and `information_schema.*` | +| Persistence | none, JSON on stdout | tables in the configured schema, default `stats` | +| Retention | whatever `sys.jobs_log` currently holds | unbounded, keeps history beyond `sys.jobs_log` | +| Aggregation | computed in SQL by CrateDB, per invocation | accumulated in the collector: call counters, duration buckets, decaying average | +| Anonymization | not available | `--anonymize` / `--deanonymize` | +| Cluster address | `--cluster-url`, `--cluster-name`, `--cluster-id` | `--cluster-url` | +| Output | `meta` / `data`, one entry per element | `meta` / `data.stats`, one entry per statement | + +Rules of thumb: +- Use `ctk info jobs` to inspect a cluster you are looking at right now, or to hand a + self-contained snapshot to someone else. +- Use `ctk cfr jobstats collect` when you need query statistics to outlive `sys.jobs_log`, + for example to find out which statements are slow over the course of a week. + + ## HTTP API Install. @@ -127,7 +205,8 @@ pip install --upgrade 'cratedb-toolkit[service]' ``` Expose collected status information. An HTTP wrapper around the same data as -`ctk info cluster`, the only endpoint is `GET /info/all`. +`ctk info cluster`, the only endpoint is `GET /info/all`. Job information is not +served over HTTP. ```shell ctk info serve @@ -142,6 +221,11 @@ Make the service listen on a specific address. ctk info serve --listen 0.0.0.0:8042 ``` +:::{note} +The HTTP service reads the cluster address from the `CRATEDB_CLUSTER_URL` environment +variable only. The `--cluster-name` and `--cluster-id` options are not honored here. +::: + :::{note} The `--reload` option is suitable for development scenarios where you intend to have the changes to the code become available while editing, in near From 7f69554ce3a54b59e7bbbb8648005613f98f3a69 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 11:56:10 +0200 Subject: [PATCH 09/12] Refactor job statistics implementation for improved state management and add unit tests for data integrity --- cratedb_toolkit/cfr/jobstats.py | 94 +++++++++++----------- doc/cfr/jobstats.md | 12 +-- tests/cfr/test_jobstats.py | 38 +++++++++ tests/cfr/test_jobstats_unit.py | 136 ++++++++++++++++++++++++++++++-- 4 files changed, 225 insertions(+), 55 deletions(-) diff --git a/cratedb_toolkit/cfr/jobstats.py b/cratedb_toolkit/cfr/jobstats.py index 763030f7..44971c9c 100644 --- a/cratedb_toolkit/cfr/jobstats.py +++ b/cratedb_toolkit/cfr/jobstats.py @@ -42,12 +42,13 @@ "INF": 0, } -stmt_log_table: str -last_exec_table: str -cursor: t.Any -report_cursor: t.Any -last_scrape: int -interval: float +# All of those are assigned by `boot()`. +stmt_log_table: str = "" +last_exec_table: str = "" +cursor: t.Any = None +report_cursor: t.Any = None +last_scrape: int = 0 +interval: float = 10.0 anonymize_sql: bool = False deanonymize_sql: bool = False # Added global flag for deanonymization decoder_dict_path: str = "" @@ -232,51 +233,56 @@ def dbinit(): report_cursor.execute(f"REFRESH TABLE {stmt_log_table}") stmt = f"SELECT id, stmt, calls, bucket, username, query_type, avg_duration, nodes, last_used FROM {stmt_log_table}" report_cursor.execute(stmt) - init_stmts(report_cursor.fetchall()) + init_stmts(fetch_records(report_cursor)) stmt = f"CREATE TABLE IF NOT EXISTS {last_exec_table} (last_execution TIMESTAMP)" report_cursor.execute(stmt) report_cursor.execute(f"REFRESH TABLE {last_exec_table}") - stmt = f"SELECT last_execution FROM {last_exec_table}" + # Aggregate, so the outcome does not depend on the order records are returned in. + # The table can hold more than one record, for example after an interrupted startup. + stmt = f"SELECT MAX(last_execution) AS last_execution, COUNT(*) AS record_count FROM {last_exec_table}" report_cursor.execute(stmt) - init_last_execution(report_cursor.fetchall()) + init_last_execution(fetch_records(report_cursor)[0]) -def init_last_execution(last_execution): +def fetch_records(db_cursor) -> t.List[t.Dict[str, t.Any]]: + """ + Return the result of the most recent query as records, keyed by column name. + """ + column_names = [column[0] for column in db_cursor.description] + return [dict(zip(column_names, row)) for row in db_cursor.fetchall()] + + +def init_last_execution(watermark: t.Dict[str, t.Any]): + """ + Adopt the recorded watermark, and create it when it does not exist yet. + """ global last_execution_ts - if len(last_execution) == 0: - last_execution_ts = 0 + last_execution_ts = watermark.get("last_execution") or 0 + if not watermark.get("record_count"): stmt = f"INSERT INTO {last_exec_table} (last_execution) VALUES (?)" - report_cursor.execute(stmt, (0,)) + report_cursor.execute(stmt, (last_execution_ts,)) # Refresh, so the `UPDATE` of `write_stats_to_db` finds the record just inserted. report_cursor.execute(f"REFRESH TABLE {last_exec_table}") - else: - last_execution_ts = last_execution[0][0] - -def init_stmts(stmts): - for stmt in stmts: - stmt_id = stmt[0] - stmt_column = stmt[1] - calls = stmt[2] - bucket = stmt[3] - user = stmt[4] - stmt_type = stmt[5] - avg_duration = stmt[6] - nodes = stmt[7] - last_used = stmt[8] +def init_stmts(records: t.Iterable[t.Dict[str, t.Any]]): + """ + Adopt statistics read from the database, without clobbering accumulated ones. + """ + for record in records: + stmt_column = record["stmt"] if stmt_column not in sys_jobs_log: sys_jobs_log[stmt_column] = { - "id": stmt_id, + "id": record["id"], "size": 0, "info": [], - "calls": calls, - "bucket": bucket, - "user": user, - "type": stmt_type, - "avg_duration": avg_duration, - "nodes": nodes, - "last_used": last_used, + "calls": record["calls"], + "bucket": record["bucket"], + "user": record["username"], + "type": record["query_type"], + "avg_duration": record["avg_duration"], + "nodes": record["nodes"], + "last_used": record["last_used"], "in_db": True, "changed": False, } @@ -332,21 +338,17 @@ def write_stats_to_db(): def read_stats(): stmt = ( - f"SELECT id, stmt, calls, avg_duration, bucket, username, query_type, nodes, last_used " + f"SELECT id, stmt, calls, bucket, username, query_type, avg_duration, nodes, last_used " f"FROM {stmt_log_table} ORDER BY calls DESC, avg_duration DESC;" ) report_cursor.execute(stmt) - results = report_cursor.fetchall() + results = fetch_records(report_cursor) # Deanonymize statements if needed if deanonymize_sql and decoder_dict_path: - deanonymized_results = [] - for row in results: - row_list = list(row) - if row_list[1]: # Check if stmt (at index 1) exists - row_list[1] = deanonymize_statement(row_list[1]) - deanonymized_results.append(tuple(row_list)) - results = deanonymized_results + for record in results: + if record["stmt"]: + record["stmt"] = deanonymize_statement(record["stmt"]) # The records just read are authoritative. Without discarding what `dbinit` has read # before, deanonymized statements would be reported next to their anonymized form. @@ -417,7 +419,9 @@ def scrape_db(): f"WHERE " f"stmt NOT LIKE '%sys.%' AND " f"stmt NOT LIKE '%information_schema.%' " - f"AND ended BETWEEN {last_scrape} AND {next_scrape} " + # Half-open interval: The watermark itself has been processed already, so a job + # ending exactly on it must not be counted a second time. + f"AND ended > {last_scrape} AND ended <= {next_scrape} " f"ORDER BY ended DESC" ) diff --git a/doc/cfr/jobstats.md b/doc/cfr/jobstats.md index 53e19190..c9a1068d 100644 --- a/doc/cfr/jobstats.md +++ b/doc/cfr/jobstats.md @@ -51,7 +51,8 @@ into per-statement statistics. Statements against `sys.*` and `information_schem skipped, so the collector does not account for its own queries. How far the collector has come is recorded as a watermark, so a restarted collector picks -up where it left off, instead of counting the same jobs again. +up where it left off, instead of counting the same jobs again. Each cycle considers jobs +which ended after the watermark, up to and including the current moment. Per distinct statement, the collector maintains: - `calls` — how often the statement has been executed @@ -119,8 +120,10 @@ variables are recognized. - `--deanonymize` — path to the decoder dictionary file used to reverse `--anonymize`, to view statements in their original form -`ctk cfr jobstats report` and `ctk cfr jobstats ui` read the statistics from the same -schema `collect` wrote them to. `ui` serves the dashboard on `localhost:7777`. +`ctk cfr jobstats report` and `ctk cfr jobstats ui` read the statistics from the schema of +the cluster URL. They do not accept `--reportdb`, so when `collect --reportdb` was used, +address that database per `--cluster-url` here. `ui` serves the dashboard on +`localhost:7777`. ## Anonymization @@ -137,8 +140,7 @@ ctk cfr jobstats view --deanonymize ./decoder_dictionary.json :::{warning} The decoder dictionary maps anonymized tokens back to the original identifiers and string -literals. Treat it as confidential, and do not ship it together with the collected -statistics. +literals. Treat it as confidential, do not ship it together with the collected statistics. ::: :::{note} diff --git a/tests/cfr/test_jobstats.py b/tests/cfr/test_jobstats.py index 1f8999fe..1fc97b8c 100644 --- a/tests/cfr/test_jobstats.py +++ b/tests/cfr/test_jobstats.py @@ -13,6 +13,7 @@ STATEMENTS_TABLE = f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements" LAST_TABLE = f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last" +BUCKET_KEYS = ["10", "50", "100", "500", "1000", "2000", "5000", "10000", "15000", "20000", "INF"] @pytest.fixture @@ -198,6 +199,43 @@ def test_cfr_jobstats_view(cratedb): assert "stats" in data_keys +def test_cfr_jobstats_view_values(cratedb, runner): + """ + Verify `ctk cfr jobstats view` reports each value in its own field. + """ + + marker = marker_statement(cratedb, "jobstats-view-values") + + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + result = runner.invoke(cli, args="jobstats view", catch_exceptions=False) + assert result.exit_code == 0, result.output + + entry = json.loads(result.output)["data"]["stats"][f"SELECT '{marker}' AS marker"] + + assert isinstance(entry["calls"], int) + assert isinstance(entry["avg_duration"], (int, float)) + assert isinstance(entry["bucket"], dict) + assert sorted(entry["bucket"]) == sorted(BUCKET_KEYS) + assert entry["user"] == "crate" + assert entry["type"] == "SELECT" + assert isinstance(entry["nodes"], list) + assert isinstance(entry["last_used"], int) + + # Verify outcome: The reported values are the stored ones. + cratedb.database.refresh_table(STATEMENTS_TABLE) + quoted = cratedb.database.quote_relation_name(STATEMENTS_TABLE) + records = cratedb.database.run_sql( + f"SELECT id, calls, avg_duration, username, query_type FROM {quoted}", records=True + ) + stored = {record["id"]: record for record in records}[entry["id"]] + assert stored["calls"] == entry["calls"] + assert stored["avg_duration"] == entry["avg_duration"] + assert stored["username"] == entry["user"] + assert stored["query_type"] == entry["type"] + + def test_cfr_jobstats_collect_records_statements(cratedb, runner): """ Verify `ctk cfr jobstats collect` records statements verbatim, when not anonymizing. diff --git a/tests/cfr/test_jobstats_unit.py b/tests/cfr/test_jobstats_unit.py index 9e61f097..8c20329e 100644 --- a/tests/cfr/test_jobstats_unit.py +++ b/tests/cfr/test_jobstats_unit.py @@ -5,6 +5,8 @@ assignment and the average computation are verified here explicitly. """ +import typing as t + import pytest from cratedb_toolkit.cfr import jobstats @@ -34,6 +36,42 @@ def job(started=1_000, duration=50, stmt="SELECT 1", query_type="SELECT", userna return (started, started + duration, {"type": query_type}, stmt, username, {"id": "n1", "name": node_name}) +def db_record(stmt="SELECT 1", **overrides): + """ + Produce a single record of the statistics table, in the shape `fetch_records` yields it. + """ + record = { + "id": "id-1", + "stmt": stmt, + "calls": 99, + "bucket": dict(jobstats.bucket_dict), + "username": "crate", + "query_type": "SELECT", + "avg_duration": 1.5, + "nodes": ["n1"], + "last_used": 1_000, + } + record.update(overrides) + return record + + +class FakeCursor: + """ + Record statements instead of executing them, and reply with canned records. + """ + + def __init__(self, column_names=None, rows=None): + self.description = [(name,) for name in column_names or []] + self.rows = rows or [] + self.statements: t.List[str] = [] + + def execute(self, statement, parameters=None): + self.statements.append(statement) + + def fetchall(self): + return self.rows + + # Bucket assignment. @@ -166,21 +204,21 @@ def test_reset_state(): def test_init_stmts_keeps_accumulated_statistics(): """ - Verify reading rows from the database does not clobber freshly accumulated statistics. + Verify reading records from the database does not clobber freshly accumulated statistics. """ jobstats.update_statistics([job(duration=50)]) - jobstats.init_stmts([("id-1", "SELECT 1", 99, dict(jobstats.bucket_dict), "crate", "SELECT", 1.0, [], 1_000)]) + jobstats.init_stmts([db_record()]) entry = jobstats.sys_jobs_log["SELECT 1"] assert entry["calls"] == 1 assert entry["in_db"] is False -def test_init_stmts_adopts_database_rows(): +def test_init_stmts_adopts_database_records(): """ - Verify rows read from the database are adopted as already persisted. + Verify records read from the database are adopted as already persisted. """ - jobstats.init_stmts([("id-1", "SELECT 1", 99, dict(jobstats.bucket_dict), "crate", "SELECT", 1.5, ["n1"], 1_000)]) + jobstats.init_stmts([db_record()]) entry = jobstats.sys_jobs_log["SELECT 1"] assert entry["id"] == "id-1" @@ -190,6 +228,94 @@ def test_init_stmts_adopts_database_rows(): assert entry["changed"] is False +def test_init_stmts_maps_columns_by_name(): + """ + Verify each column ends up in its own field. + + The readers of the statistics table use different `SELECT` orders. Addressing the + columns by position made `jobstats view` report the average duration as the duration + histogram, the histogram as the user, the user as the query type, and so on. + """ + jobstats.init_stmts( + [ + db_record( + bucket={"10": 7}, + username="hotzenplotz", + query_type="DDL", + avg_duration=12.5, + nodes=["node-1"], + last_used=1_700_000_000_000, + ) + ] + ) + + entry = jobstats.sys_jobs_log["SELECT 1"] + assert entry["bucket"] == {"10": 7} + assert entry["user"] == "hotzenplotz" + assert entry["type"] == "DDL" + assert entry["avg_duration"] == 12.5 + assert entry["nodes"] == ["node-1"] + assert entry["last_used"] == 1_700_000_000_000 + + +def test_fetch_records(): + """ + Verify query results are keyed by column name. + """ + cursor = FakeCursor(column_names=["id", "stmt"], rows=[("id-1", "SELECT 1"), ("id-2", "SELECT 2")]) + assert jobstats.fetch_records(cursor) == [ + {"id": "id-1", "stmt": "SELECT 1"}, + {"id": "id-2", "stmt": "SELECT 2"}, + ] + + +def test_init_last_execution_adopts_watermark(mocker): + """ + Verify a recorded watermark is adopted, without touching the table. + """ + cursor = FakeCursor() + mocker.patch.object(jobstats, "report_cursor", cursor) + mocker.patch.object(jobstats, "last_exec_table", '"testdrive".jobstats_last') + + jobstats.init_last_execution({"last_execution": 1_700_000_000_000, "record_count": 1}) + + assert jobstats.last_execution_ts == 1_700_000_000_000 + assert cursor.statements == [] + + +def test_init_last_execution_creates_watermark(mocker): + """ + Verify a missing watermark record is created, so it can be updated later on. + """ + cursor = FakeCursor() + mocker.patch.object(jobstats, "report_cursor", cursor) + mocker.patch.object(jobstats, "last_exec_table", '"testdrive".jobstats_last') + + jobstats.init_last_execution({"last_execution": None, "record_count": 0}) + + assert jobstats.last_execution_ts == 0 + assert [statement.split()[0] for statement in cursor.statements] == ["INSERT", "REFRESH"] + + +def test_scrape_db_uses_half_open_interval(mocker): + """ + Verify the collection window excludes the watermark, and includes the current moment. + + An inclusive lower bound would count a job ending exactly on the watermark twice. + """ + cursor = FakeCursor() + mocker.patch.object(jobstats, "cursor", cursor) + mocker.patch.object(jobstats, "last_scrape", 1_700_000_000_000) + + jobstats.scrape_db() + + statement = cursor.statements[0] + assert "ended > 1700000000000" in statement + assert f"ended <= {jobstats.last_scrape}" in statement + assert "BETWEEN" not in statement + assert jobstats.last_scrape > 1_700_000_000_000 + + # Anonymization. From 1191697174cf28cbf4f559ec94792d18787b5901 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Wed, 12 Aug 2026 12:00:35 +0200 Subject: [PATCH 10/12] Update changelog with recent fixes and breaking changes for job statistics --- CHANGES.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGES.md b/CHANGES.md index 22aa96de..2e88a094 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,6 +1,11 @@ # Changelog ## Unreleased +- Fixed `ctk cfr jobstats` bugs related anonymization, views, ui, report + and collect arguments. +- Breaking change: `ctk info jobs` reported the 99th percentile of the query duration as + `p90`. The field of the `top100_count` element is called `p99` now. +- Documentation: Explained `ctk info jobs` and `ctk cfr jobstats`, and how they differ - Breaking change: Dropped support for Python 3.8 and 3.9, which have reached end-of-life. The minimum supported Python version is now 3.10. - Fixed a failing SQL statement during `ctk info cluster` From ecbbd05b9de8c3c98d9c59831cba4844d567fb88 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 17 Aug 2026 10:29:21 +0200 Subject: [PATCH 11/12] Fix PR comments: improve logging for decoder dictionary loading, ensure last_used reflects most recent execution, and maintain pending records on write failures --- cratedb_toolkit/cfr/jobstats.py | 29 +++++-- doc/cfr/jobstats.md | 44 ++++++++-- tests/cfr/test_jobstats.py | 146 +++++++++++++++++++++++++++++++- 3 files changed, 204 insertions(+), 15 deletions(-) diff --git a/cratedb_toolkit/cfr/jobstats.py b/cratedb_toolkit/cfr/jobstats.py index 44971c9c..3c7a26a2 100644 --- a/cratedb_toolkit/cfr/jobstats.py +++ b/cratedb_toolkit/cfr/jobstats.py @@ -172,8 +172,10 @@ def anonymize_statement(statement: str) -> str: try: with open(decoder_dict_path, "r") as f: encoder_dict = json.load(f) - except (FileNotFoundError, json.JSONDecodeError) as e: - logger.warning(f"Could not load encoder dictionary: {e}") + except FileNotFoundError: + logger.info(f"No decoder dictionary found yet, creating a new one: {decoder_dict_path}") + except json.JSONDecodeError as e: + logger.warning(f"Could not load encoder dictionary, continuing without it: {e}") # Call anonymize and extract only the anonymized statement (first item). with redirect_stdout(io.StringIO()): @@ -299,6 +301,7 @@ def write_stats_to_db(): f"UPDATE {stmt_log_table} SET calls = ?, avg_duration = ?, nodes = ?, bucket = ?, last_used = ? WHERE id = ?" ) write_params = [] + write_keys = [] for key in sys_jobs_log.keys(): if not sys_jobs_log[key]["in_db"]: write_params.append( @@ -314,8 +317,7 @@ def write_stats_to_db(): sys_jobs_log[key]["last_used"], ] ) - sys_jobs_log[key]["in_db"] = True - sys_jobs_log[key]["changed"] = False + write_keys.append(key) elif sys_jobs_log[key]["changed"]: report_cursor.execute( update_query_stmt, @@ -330,7 +332,17 @@ def write_stats_to_db(): ) sys_jobs_log[key]["changed"] = False if len(write_params) > 0: - report_cursor.executemany(write_query_stmt, write_params) + results = report_cursor.executemany(write_query_stmt, write_params) or [] + + outcomes = list(results) + [None] * (len(write_params) - len(results)) + for key, outcome in zip(write_keys, outcomes): + if outcome is not None and outcome.get("rowcount", 0) < 0: + logger.warning( + f"Storing statistics failed, retrying on the next cycle: {outcome.get('error_message') or outcome}" + ) + continue + sys_jobs_log[key]["in_db"] = True + sys_jobs_log[key]["changed"] = False stmt = f"UPDATE {last_exec_table} SET last_execution = ?" report_cursor.execute(stmt, (last_scrape,)) @@ -400,7 +412,8 @@ def update_statistics(query_results): sys_jobs_log[stmt]["changed"] = True sys_jobs_log[stmt]["avg_duration"] = (sys_jobs_log[stmt]["avg_duration"] + duration) / 2 sys_jobs_log[stmt]["bucket"] = assign_to_bucket(sys_jobs_log[stmt]["bucket"], duration) - sys_jobs_log[stmt]["last_used"] = started + # Keep the most recent execution, independently of the order records arrive in. + sys_jobs_log[stmt]["last_used"] = max(sys_jobs_log[stmt]["last_used"] or 0, started) sys_jobs_log[stmt]["calls"] += 1 sys_jobs_log[stmt]["nodes"].append(node) sys_jobs_log[stmt]["nodes"] = list(set(sys_jobs_log[stmt]["nodes"])) # only save unique nodes @@ -422,7 +435,9 @@ def scrape_db(): # Half-open interval: The watermark itself has been processed already, so a job # ending exactly on it must not be counted a second time. f"AND ended > {last_scrape} AND ended <= {next_scrape} " - f"ORDER BY ended DESC" + # Oldest first, so the *decaying* average of `update_statistics` ends up weighing + # the most recent execution most heavily. + f"ORDER BY ended ASC" ) cursor.execute(stmt) diff --git a/doc/cfr/jobstats.md b/doc/cfr/jobstats.md index c9a1068d..d6a2834b 100644 --- a/doc/cfr/jobstats.md +++ b/doc/cfr/jobstats.md @@ -24,16 +24,26 @@ The collector stores its statistics in the schema of the cluster URL, `stats` by export CRATEDB_CLUSTER_URL=crate://crate@localhost:4200/?schema=stats ``` -Collect statistics, then display or explore them. +Collects statistics on an ongoing basis: + ```shell ctk cfr jobstats collect ``` + +Prints collected statistics as a JSON document: + ```shell ctk cfr jobstats view ``` + +Shows the top 10 collected statements, sorted by runtime descending: + ```shell ctk cfr jobstats report ``` + +Launches a web interface with visualisations for interactive exploration of statistics: + ```shell ctk cfr jobstats ui ``` @@ -48,7 +58,18 @@ result means nothing has been collected into that schema yet. `collect` polls `sys.jobs_log` for jobs which finished since the last poll, and folds them into per-statement statistics. Statements against `sys.*` and `information_schema.*` are -skipped, so the collector does not account for its own queries. +skipped, which excludes the collector's own poll query, but not its writes. + +:::{note} +The collector observes itself. Reading `sys.jobs_log` is filtered out, however the +statements which store the statistics are regular DDL and DML against the tables below, +so the next cycle collects them like any other query. On a quiet cluster, the statistics +therefore consist mostly of the collector's own `CREATE TABLE IF NOT EXISTS`, +`REFRESH TABLE`, `SELECT`, `INSERT`, and `UPDATE` statements. Take that into account when +interpreting `calls`. Pointing `--reportdb` at a *separate cluster* avoids it, because the +statistics are then written outside the cluster being observed. A `--reportdb` which only +differs in the schema does not help, as those writes still show up in `sys.jobs_log`. +::: How far the collector has come is recorded as a watermark, so a restarted collector picks up where it left off, instead of counting the same jobs again. Each cycle considers jobs @@ -63,8 +84,11 @@ Per distinct statement, the collector maintains: execution. Recent executions therefore weigh much more heavily than an arithmetic mean over all executions would. - `nodes` — the nodes which have run the statement, without duplicates -- `last_used` — when the statement was last seen -- `username`, `query_type` — as reported by CrateDB +- `last_used` — when the most recent execution of the statement *started* +- `username`, `query_type` — as reported by CrateDB, taken from the execution which + introduced the statement. Neither is updated afterwards, so when the same statement is + run by several users, the record keeps the user which ran it first and does not grow a + second record for the others. Use `ctk info jobs` for a per-user breakdown. ## Tables @@ -113,7 +137,8 @@ variables are recognized. cluster URL, and written to this one. - `--anonymize` — path to a decoder dictionary file for anonymizing SQL statements before they are stored; using the flag without a value defaults to `decoder_dictionary.json` in - the current working directory + the current working directory. The file does not need to exist: it is created on the + first run, and extended as further identifiers are encountered. `ctk cfr jobstats view`: - `--reportdb` / `-r` — a separate database URL to read the statistics from @@ -128,8 +153,9 @@ address that database per `--cluster-url` here. `ui` serves the dashboard on ## Anonymization With `--anonymize`, statements are anonymized before they are stored, and the substitutions -are recorded in the decoder dictionary file. Keep that file: it is the only way to make the -collected statements legible again, using `view --deanonymize`. +are recorded in the decoder dictionary file. The file is created on the first run and grows +as new identifiers are encountered. Keep it: it is the only way to make the collected +statements legible again, using `view --deanonymize`. ```shell ctk cfr jobstats collect --once --anonymize ./decoder_dictionary.json @@ -148,4 +174,8 @@ Anonymization fails closed: when a statement cannot be anonymized, it is stored `>` rather than in clear text. Statistics per distinct statement remain meaningful, but such statements cannot be recovered with `--deanonymize`. The event is reported as a warning. + +Expect this to happen occasionally, and more often as the dictionary grows: the underlying +`queryanonymizer` can fail with a `PatternError` on a statement it anonymized successfully +when the dictionary was still small. ::: diff --git a/tests/cfr/test_jobstats.py b/tests/cfr/test_jobstats.py index 1fc97b8c..0c797811 100644 --- a/tests/cfr/test_jobstats.py +++ b/tests/cfr/test_jobstats.py @@ -1,6 +1,7 @@ # ruff: noqa: S608 import json import os +import time import uuid import pytest @@ -66,7 +67,7 @@ def test_cfr_jobstats_collect_self(cratedb, caplog): # Configure database URI. dburi = cratedb.database.dburi + f"?schema={TESTDRIVE_EXT_SCHEMA}" - marker_statement(cratedb, "jobstats-collect-self-marker") + marker = marker_statement(cratedb, "jobstats-collect-self-marker") # Invoke command. runner = CliRunner(env={"CRATEDB_CLUSTER_URL": dburi}) @@ -90,6 +91,9 @@ def test_cfr_jobstats_collect_self(cratedb, caplog): cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_statements") >= 1 + # The record count alone is satisfied by unrelated cluster activity. + assert any(marker in stmt for stmt in collected_statements(cratedb)) + cratedb.database.refresh_table(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") assert cratedb.database.count_records(f"{TESTDRIVE_EXT_SCHEMA}.jobstats_last") == 1 @@ -354,6 +358,146 @@ def marker_calls(): assert second_watermark > first_watermark +def test_cfr_jobstats_collect_last_used_is_most_recent_execution(cratedb, runner): + """ + Verify `last_used` reports the most recent execution, also within a single cycle. + + A cycle hands all its jobs to `update_statistics` in one batch. Assigning `last_used` + per record unconditionally leaves the value of whichever record is processed last, + which is not the most recent execution. + """ + + marker = f"jobstats-last-used-{uuid.uuid4().hex[:8]}" + statement = f"SELECT '{marker}' AS marker" + escaped = statement.replace("'", "''") + + # Several executions of the very same statement, all within one collection cycle. + for _ in range(3): + cratedb.database.run_sql(statement) + time.sleep(0.2) + + result = runner.invoke(cli, args="jobstats collect --once", catch_exceptions=False) + assert result.exit_code == 0, result.output + + # Ground truth, as recorded by CrateDB itself. Compare the statement exactly: the + # verification queries mention the marker as well, but are not equal to it. + executions = cratedb.database.run_sql( + f"SELECT MIN(started) AS first_started, MAX(started) AS last_started, COUNT(*) AS execution_count " + f"FROM sys.jobs_log WHERE stmt = '{escaped}'", + records=True, + )[0] + assert executions["execution_count"] == 3 + assert executions["last_started"] > executions["first_started"] + + cratedb.database.refresh_table(STATEMENTS_TABLE) + quoted = cratedb.database.quote_relation_name(STATEMENTS_TABLE) + stored = cratedb.database.run_sql(f"SELECT calls, last_used FROM {quoted} WHERE stmt = '{escaped}'", records=True)[ + 0 + ] + + assert stored["calls"] == 3 + assert stored["last_used"] == executions["last_started"] + + +def test_cfr_jobstats_last_used_does_not_move_backwards(mocker): + """ + Verify `last_used` does not regress when a later cycle reports an older execution. + + A long-running job may start before the watermark and only end after it, so it is + collected one cycle later than shorter jobs which started after it. Ordering the + records of a cycle is therefore not sufficient on its own. + """ + + from cratedb_toolkit.cfr import jobstats + + mocker.patch.object(jobstats, "anonymize_sql", False) + statement = "SELECT 'backwards'" + classification = {"type": "SELECT"} + node = {"id": "n1"} + + # One cycle, holding a short job. + jobstats.update_statistics([(5000, 5050, classification, statement, "crate", node)]) + assert jobstats.sys_jobs_log[statement]["last_used"] == 5000 + + # The next cycle, holding a job which started earlier, but only ended now. + jobstats.update_statistics([(4000, 6000, classification, statement, "crate", node)]) + assert jobstats.sys_jobs_log[statement]["last_used"] == 5000 + + +def pending_statistics(identifier: str) -> dict: + """ + Return an in-memory statistics record which has not been stored yet. + """ + return { + "id": identifier, + "calls": 1, + "bucket": dict.fromkeys(BUCKET_KEYS, 0), + "user": "crate", + "type": "SELECT", + "avg_duration": 1.0, + "nodes": [], + "last_used": 1, + "in_db": False, + "changed": True, + } + + +def test_cfr_jobstats_write_stats_keeps_failed_records_pending(mocker): + """ + Verify statistics are only flagged as stored once the record has really been written. + + A bulk operation does not raise when individual records fail, it reports a negative + `rowcount` for them. Flagging such a statement as stored anyway would downgrade it to + an `UPDATE` from the next cycle on, which matches no record, so its statistics would + be dropped silently. + """ + + from cratedb_toolkit.cfr import jobstats + + cursor = mocker.Mock() + cursor.executemany.return_value = [{"rowcount": 1}, {"rowcount": -2, "error_message": "nope"}] + mocker.patch.object(jobstats, "report_cursor", cursor) + mocker.patch.object(jobstats, "stmt_log_table", '"testdrive".jobstats_statements') + mocker.patch.object(jobstats, "last_exec_table", '"testdrive".jobstats_last') + + jobstats.sys_jobs_log["SELECT 'stored'"] = pending_statistics("id-stored") + jobstats.sys_jobs_log["SELECT 'failed'"] = pending_statistics("id-failed") + + jobstats.write_stats_to_db() + + stored = jobstats.sys_jobs_log["SELECT 'stored'"] + assert stored["in_db"] is True + assert stored["changed"] is False + + # Still pending, so the next cycle inserts it again instead of updating nothing. + failed = jobstats.sys_jobs_log["SELECT 'failed'"] + assert failed["in_db"] is False + assert failed["changed"] is True + + +def test_cfr_jobstats_write_stats_without_record_outcomes(mocker): + """ + Verify statistics are flagged as stored when the driver reports no per-record outcome. + + Assuming failure instead would insert the very same record again on each cycle. + """ + + from cratedb_toolkit.cfr import jobstats + + cursor = mocker.Mock() + cursor.executemany.return_value = None + mocker.patch.object(jobstats, "report_cursor", cursor) + mocker.patch.object(jobstats, "stmt_log_table", '"testdrive".jobstats_statements') + mocker.patch.object(jobstats, "last_exec_table", '"testdrive".jobstats_last') + + jobstats.sys_jobs_log["SELECT 'unknown'"] = pending_statistics("id-unknown") + + jobstats.write_stats_to_db() + + assert jobstats.sys_jobs_log["SELECT 'unknown'"]["in_db"] is True + assert jobstats.sys_jobs_log["SELECT 'unknown'"]["changed"] is False + + def test_cfr_jobstats_view_without_data(cratedb): """ Verify `ctk cfr jobstats view` on a schema without collected statistics. From 24d3eac31dfa54c6abfb88f4d466283dd34c7310 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 17 Aug 2026 10:31:24 +0200 Subject: [PATCH 12/12] fix formatting --- cratedb_toolkit/cfr/jobstats.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cratedb_toolkit/cfr/jobstats.py b/cratedb_toolkit/cfr/jobstats.py index 3c7a26a2..44dfb5d6 100644 --- a/cratedb_toolkit/cfr/jobstats.py +++ b/cratedb_toolkit/cfr/jobstats.py @@ -333,7 +333,7 @@ def write_stats_to_db(): sys_jobs_log[key]["changed"] = False if len(write_params) > 0: results = report_cursor.executemany(write_query_stmt, write_params) or [] - + outcomes = list(results) + [None] * (len(write_params) - len(results)) for key, outcome in zip(write_keys, outcomes): if outcome is not None and outcome.get("rowcount", 0) < 0: