fix(inc-2141): only check essential clusters in liveness healthcheck

[onewland]

Depends on #7924

Per INC-2141: previously sanity_check_clickhouse_connections returned True as soon as any clickhouse cluster answered. The intent was to keep the API alive when low-priority datasets (e.g. replays) had a cluster outage, but it backfired when the events_analytics_platform (EAP) cluster held onto stalled connections — some other cluster always answered fast, so health stayed green and pods that should have been recycled weren't.

This change scopes the liveness check to a small set of essential clusters and fails health if any of them is unreachable.

Identifying essential clusters

Match on storage_set_key (which maps 1:1 to a cluster in settings.CLUSTERS)

ESSENTIAL_STORAGE_SET_KEYS = {
    StorageSetKey.EVENTS,                     # errors storage
    StorageSetKey.EVENTS_RO,                  # errors_ro storage
    StorageSetKey.EVENTS_ANALYTICS_PLATFORM,
}

Picking storages by storage_set_key rather than storage key is also more robust against renaming individual storages (e.g. eap_items) — the cluster topology is the load-bearing concept here.

Behavior change

• sanity_check_clickhouse_connections filters to essential storages only, then requires every unique cluster to return successfully within timeout_seconds.
• Returns False if any cluster raises, any cluster fails to respond within the timeout, any essential storage has an undefined cluster, or no essential storages are found at all (defensive — should never happen in prod).
• The thorough path (check_all_tables_present, gated by ?thorough=true) is unchanged.
• The health_check_ignore_clickhouse config knob and the health_check.timeout_override_seconds override are unchanged.

ThreadPoolExecutor: don't block on stalled futures

Previously the executor was used as a context manager:

with ThreadPoolExecutor(...) as executor:
    ...

__exit__ calls shutdown(wait=True), which blocks until every submitted future finishes — including stalled ones. That defeats the timeout entirely: the very scenario INC-2141 is about (EAP holding broken connections) is exactly when we'd block on the way out.

The executor is now managed manually with shutdown(wait=False) in a finally:

executor = ThreadPoolExecutor(...)
try:
    ...
finally:
    executor.shutdown(wait=False)

Stalled worker threads run to completion in the background; that's acceptable because the pod is failing health and will be recycled by k8s. The function itself is now bounded by the timeout regardless of cluster behavior.

Tests

• test_sanity_check_clickhouse_connections — unchanged, still passes (errors_ro is essential and reachable).
• test_single_bad_dataset_passes_healthcheck → renamed test_non_essential_bad_storage_ignored with updated comment: a broken non-essential storage doesn't affect health.
• New test_no_essential_storages_fails: only non-essential storage keys configured → fails defensively.
• New test_unreachable_essential_cluster_fails_healthcheck: mocks _execute_show_tables to raise → fails (the key INC-2141 regression test).
• New test_stalled_cluster_does_not_block_healthcheck: mocks _execute_show_tables with a 10s sleep, asserts the function returns within 5s under a 0.2s timeout. Without shutdown(wait=False) this would block ~10s. This is the regression test for the executor change.

[linear-code]

INC-2141

[sentry]

Bug: The test passes due to a KeyError from a mocked storage key, which aborts the function under test and bypasses the logic it's meant to validate.
_{Severity: MEDIUM}

Suggested Fix

Modify the test setup to prevent the KeyError. Either properly register the mocked storage in the _storage_factory or patch the get_storage call to return a valid mock storage object. This will ensure the filter_checked_storages function executes completely and the intended logic is tested.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: tests/utils/test_check_clickhouse.py#L199-L203

Potential issue: The tests `test_no_essential_storages_fails` and
`test_non_essential_bad_storage_ignored` pass for the wrong reason. The test setup uses
a mocked `FakeStorageKey` which is not found in the global storage factory, causing
`get_storage` to raise a `KeyError`. This exception is caught and silently passed in
`sanity_check_clickhouse_connections`, causing the `filter_checked_storages` function to
abort prematurely. Consequently, the new logic intended to filter non-essential storages
is never actually executed or tested, making the test's success misleading.

_{Did we get this right? 👍 / 👎 to inform future reviews.}