Add structured telemetry with ClickHouse sink to flytekit task execution steps

[ryanjwong]

Why are the changes needed?

Flytekit's task execution pipeline has many distinct phases (task loading, input download, deserialization, user code execution, output conversion, upload, etc.) but only a few are instrumented with timing today. This makes it hard to diagnose where time is spent during task execution.

These changes emit structured telemetry at each step boundary. When FLYTE_TELEMETRY_CLICKHOUSE_URL is set, each event is immediately inserted as a row in ClickHouse via a background HTTP POST. Otherwise, structured JSON logs are emitted to stderr for Loki/LogQL ingestion as a fallback.

What changes were proposed in this pull request?

5 files changed:

1. flytekit/loggers.py — New telemetry_logger (flytekit.telemetry), controlled by FLYTE_TELEMETRY_ENABLED env var (default: on). New ClickHouseTelemetrySink class that inserts each event as a single row via a fire-and-forget background thread HTTP POST (FORMAT JSONEachRow). No buffering, no batching — every step is immediately visible in ClickHouse. Zero new dependencies (stdlib urllib + threading only).

2. flytekit/core/utils.py — Enhanced timeit context manager:

   • New _emit_telemetry() method emits structured data with event, step, wall_time_s, process_time_s, status, error_type, execution_id, task_name, project, domain.
   • Routes to ClickHouseTelemetrySink when enabled, falls back to telemetry_logger (stderr JSON) otherwise.
   • New **extras kwarg for passing additional metadata (e.g. input_size_bytes).
   • Entire _emit_telemetry body is wrapped in try/except Exception: pass so telemetry can never break execution or mask real exceptions from __exit__.

3. flytekit/bin/entrypoint.py — Wrapped 7 previously uninstrumented steps in _dispatch_execute: load_task, download_inputs, deserialize_inputs, task_dispatch_execute, output_offloading, upload_outputs, output_deck.

4. flytekit/core/base_task.py — Wrapped 3 previously uninstrumented steps in PythonTask.dispatch_execute: pre_execute, post_execute, write_decks.

5. tests/flytekit/unit/core/test_utils.py — 17 new tests (30 total, all passing): 8 for stderr fallback path, 9 for ClickHouse sink (per-row POST, background thread firing, routing, error silencing, enable/disable).

ClickHouse env vars

Env var	Default	Purpose
FLYTE_TELEMETRY_ENABLED	"1"	Master toggle for all telemetry
FLYTE_TELEMETRY_CLICKHOUSE_URL	(none)	e.g. https://host:8443 — enables ClickHouse sink
FLYTE_TELEMETRY_CLICKHOUSE_USER	"default"	ClickHouse username
FLYTE_TELEMETRY_CLICKHOUSE_PASSWORD	""	ClickHouse password
FLYTE_TELEMETRY_CLICKHOUSE_DATABASE	"default"	Target database
FLYTE_TELEMETRY_CLICKHOUSE_TABLE	"flytekit_telemetry"	Target table

Example telemetry event (ClickHouse row)

{"event": "flytekit_step", "step": "execute_user_code", "wall_time_s": 1.234, "process_time_s": 0.89, "status": "success", "task_name": "my_task", "execution_id": "f12abc", "project": "ml", "domain": "production", "timestamp": "2025-02-12 10:30:45.123"}

Fallback: LogQL queries (when ClickHouse not configured)

# p99 user code execution time by task
{app="flytekit"} | json | event="flytekit_step" | step="execute_user_code" | unwrap wall_time_s | quantile_over_time(0.99, [5m]) by (task_name)

# error rate per step
sum by (step) (rate({app="flytekit"} | json | event="flytekit_step" | status="error" [5m]))

How was this patch tested?

30 unit tests in tests/flytekit/unit/core/test_utils.py, all passing:

Test	What it verifies
test_timeit_telemetry_success_fallback_to_logger	Structured log emitted on success with correct fields
test_timeit_telemetry_error_fallback_to_logger	status="error" and error_type set on exception
test_timeit_telemetry_extras	Custom kv pairs via **extras pass through
test_timeit_telemetry_context_enrichment	Context (execution_id, task_name, etc.) pulled from FlyteContext
test_telemetry_disabled	FLYTE_TELEMETRY_ENABLED=0 silences telemetry
test_telemetry_enabled_by_default	Telemetry enabled when env var unset
test_timeit_telemetry_json_format	Output parses as valid JSON with expected fields
test_timeit_all_steps_in_task_execution	Local task run emits pre_execute, Execute user level code, post_execute
test_clickhouse_sink_disabled_without_url	Sink disabled when URL not set
test_clickhouse_sink_enabled_with_url	Sink enabled when URL is set
test_clickhouse_sink_send_noop_when_disabled	send() is no-op when disabled
test_clickhouse_sink_send_fires_background_thread	send() spawns daemon thread targeting _post_row
test_clickhouse_sink_post_row_sends_json	HTTP POST body is single-line JSON with correct URL
test_timeit_routes_to_clickhouse_sink	timeit routes to sink when enabled
test_timeit_routes_to_clickhouse_on_error	Error status/type captured on exception
test_clickhouse_sink_post_row_silences_errors	Network errors don't propagate
test_default_clickhouse_sink_is_disabled	Default sink is disabled (no URL configured)

⚠️ Human review checklist

• Telemetry is ON by default — FLYTE_TELEMETRY_ENABLED defaults to "1". This will add telemetry overhead for all existing deployments. Confirm this is acceptable or change default to off.
• Silent exception swallowing — _emit_telemetry and _post_row have bare except Exception: pass. Telemetry configuration issues and ClickHouse connectivity problems will be completely silent. No way to know if telemetry is actually reaching ClickHouse.
• No ClickHouse table DDL provided — The table flytekit_telemetry must exist before telemetry works. Consider adding a migration script or documenting the required schema.
• Thread-per-event overhead — Each step spawns a background thread + HTTP connection. For workflows with many tasks/steps, this could be significant. Consider if connection pooling or batching is needed at scale.

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

---

Link to Devin run: https://app.devin.ai/sessions/c6fb4462bb7e4072b175e7fc664aad4b
Requested by: @ryanjwong

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no potential bugs to report.

View in Devin Review to see 6 additional findings.

[devin-ai-integration]

Devin Review found 1 new potential issue.

View 10 additional findings in Devin Review.

[devin-ai-integration]

Devin Review found 1 new potential issue.

View 17 additional findings in Devin Review.

[Carlos-Marques]

if these don't exist does it fail correctly by not emitting any metrics?

[devin-ai-integration]

Yes — if the env vars aren't set, it degrades gracefully:

1. ClickHouseTelemetrySink.__init__ defaults self._url to "" (line 222)
2. self._enabled = bool(self._url) → False when URL is empty (line 227)
3. send() returns immediately if not enabled (line 235-236)
4. If FLYTE_TELEMETRY_ENABLED is explicitly set to "0"/"false", the sink is set to None entirely (line 271)

So no metrics emitted, no errors, no HTTP calls — completely silent when unconfigured.