etl-prometheus2clickhouse

ETL job that reads metrics from Prometheus, writes them to ClickHouse and uses same ClickHouse to store job state.

Overview

The job:

• reads all metrics from Prometheus using query_range with {__name__=~".+"} selector (exports all metrics matching this selector);
• writes rows into a single ClickHouse table:
  • timestamp (DateTime64(6, 'UTC')) – metric timestamp with microseconds precision;
  • name (String) – metric name;
  • labels (Nested(key, value)) – metric labels as Nested structure with key and value arrays;
  • value (Float64) – metric value;
• stores job state in ClickHouse ETL table:
  • timestamp_progress – current processing progress timestamp;
  • timestamp_start – job start timestamp;
  • timestamp_end – job completion timestamp;
  • batch_window_seconds – size of processed window;
  • batch_rows – number of metric value pairs (rows) successfully written to ClickHouse in this batch;
  • batch_skipped_count – number of value pairs skipped due to format errors (non-numeric values that cannot be parsed as Float64).

All connection settings are provided via environment variables. Job state is stored in ClickHouse ETL table.

Important: Before the first run, the timestamp_progress value must be set in ClickHouse ETL table to specify the starting point for data processing. The job will fail if this value is not found.

Note: timestamp_start and timestamp_end may be absent on the first run (they will be created automatically). Only timestamp_progress is required before the first execution.

Requirements

• uv package manager

Installation

make install

Configuration

All connection settings are defined in environment variables. See env.example for the full list. The most important variables:

• PROMETHEUS_URL – Prometheus/Mimir base URL;
  • Optional basic auth: set PROMETHEUS_USER and PROMETHEUS_PASSWORD;
  • Set PROMETHEUS_INSECURE=1 to disable TLS verification;
  • PROMETHEUS_QUERY_STEP_SECONDS – step resolution for query_range in seconds (default: 15). Should match scrape_interval to get all data points;
  • PROMETHEUS_TIMEOUT – HTTP request timeout in seconds (default: 10);
• CLICKHOUSE_URL – ClickHouse HTTP URL (required);
  • CLICKHOUSE_TABLE_METRICS – table name for metrics (default: default.metrics);
  • CLICKHOUSE_TABLE_ETL – table name for ETL state (default: default.etl);
  • Optional: CLICKHOUSE_USER and CLICKHOUSE_PASSWORD for authentication;
  • Set CLICKHOUSE_INSECURE=1 to disable TLS verification;
  • CLICKHOUSE_CONNECT_TIMEOUT – HTTP connection timeout in seconds (default: 10);
  • CLICKHOUSE_SEND_RECEIVE_TIMEOUT – HTTP send/receive timeout in seconds for insert operations (default: 300);
• BATCH_WINDOW_SIZE_SECONDS – processing window size in seconds (default: 300);
• BATCH_WINDOW_OVERLAP_SECONDS – overlap in seconds to avoid missing data at boundaries (default: 0);
• MIN_WINDOW_START_TIMESTAMP – minimum allowed window_start timestamp. Window start will be clamped to this value if calculated start is earlier (default: 0, Unix epoch);
• TEMP_DIR – temporary directory for intermediate data files (default: /tmp);
• LOG_LEVEL – logging level (default: INFO).

Running

Run locally:

make run

Run checks:

make all

Debugging

For debugging purposes, you can run the ETL job periodically to monitor its behavior:

while true; do make docker-run; sleep 10; done

This will run the job every 10 seconds. Adjust the sleep interval as needed.

ClickHouse Table Requirements

The target ClickHouse tables must be configured to handle potential duplicate inserts. This is required because the job is designed to be idempotent: if state update fails after successful data write, the job will reprocess the same window on the next run, which may result in duplicate rows.

Note: Idempotency is ensured by the storage layer (ReplacingMergeTree engine), not by the ETL job itself. The job may insert duplicate rows, and ClickHouse handles deduplication during merge operations.

You can use the default database or create a custom database. To create a custom database:

CREATE DATABASE IF NOT EXISTS metrics;

Recommended table engine: ReplacingMergeTree or a table with deduplication enabled.

Note on deduplication: ReplacingMergeTree performs deduplication asynchronously during background merge operations. Duplicate rows may be visible in queries until merges complete. Use FINAL keyword in SELECT queries to see deduplicated results immediately, though this may impact query performance.

The table schemas should match the following structure:

Metrics table:

CREATE TABLE default.metrics (
    id UInt64 MATERIALIZED cityHash64(
        timestamp,
        name,
        labels.key,
        labels.value,
        value
    ),
    timestamp DateTime64(6, 'UTC'),
    name String CODEC(ZSTD(3)),
    labels Nested(
        key String,
        value String
    ) CODEC(ZSTD(3)),
    value Float64
) ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (
    timestamp,
    name,
    arraySort(
        arrayMap((k, v) -> tuple(k, v), labels.key, labels.value)
    )
);

Note: Table is partitioned by date (toYYYYMMDD(timestamp)) to improve merge performance. Without partitioning, all data goes into a single partition, causing slow merges as data volume grows. With daily partitioning, merges only affect small date ranges, significantly improving performance.

Note on id column: The id field is a MATERIALIZED column that is always auto-generated based on data content using cityHash64(). It provides unique identifier within the table (8 bytes, 50% smaller than UUID). Hash values have high entropy and do not compress well, so no compression codec is applied to this column. It is stored on disk and can be used in WHERE clauses, JOINs, and ORDER BY for efficient queries. By default, MATERIALIZED columns are not included in SELECT * results. To include id in results, either explicitly specify it (SELECT id, * FROM ...) or use the setting SET asterisk_include_materialized_columns=1.

ETL state table:

CREATE TABLE default.etl (
    id UInt64 MATERIALIZED cityHash64(
        timestamp_start,
        coalesce(timestamp_end, toDateTime(0)),
        coalesce(timestamp_progress, toDateTime(0)),
        coalesce(batch_window_seconds, 0),
        coalesce(batch_rows, 0),
        coalesce(batch_skipped_count, 0)
    ),
    timestamp_start DateTime,
    timestamp_end Nullable(DateTime),
    timestamp_progress Nullable(DateTime),
    batch_window_seconds Nullable(Int64) CODEC(ZSTD(3)),
    batch_rows Nullable(Int64) CODEC(ZSTD(3)),
    batch_skipped_count Nullable(Int64) CODEC(ZSTD(3))
) ENGINE = ReplacingMergeTree()
ORDER BY (timestamp_start);

Note on id column: Same as in metrics table - MATERIALIZED column that is always auto-generated based on data content using cityHash64(), providing unique identifier within the table (8 bytes, no compression codec applied as hash values have high entropy and do not compress well).

Note: ETL state operates in second-level resolution (DateTime type), while metrics table uses microsecond precision (DateTime64(6)). This means state timestamps are rounded to seconds, which affects window boundaries and overlap calculations.

Logging

Logging is implemented with logging-objects-with-schema and ECS (Elastic Common Schema) formatter for structured JSON logging.

Format:

• All logs are formatted as JSON using ECS (Elastic Common Schema) format
• Timestamps are automatically formatted in UTC timezone
• Structured fields are validated against schema defined in logging_objects_with_schema.json

Output:

• Non-error messages (DEBUG, INFO, WARNING) are sent to stdout
• Error messages (ERROR and above) are sent to stderr

This separation allows easy filtering and routing in containerized environments and log aggregation systems.

Container environment: This logging model is designed for containerized environments where stdout/stderr streams are aggregated by log collection systems (e.g., Docker, Kubernetes, log aggregators).

Troubleshooting

Job Stuck: TimestampStart Exists but TimestampEnd Missing

Symptoms:

• Job logs show: "Previous job is still running (TimestampStart exists but TimestampEnd is missing), skipping run"
• Job cannot start on subsequent runs

Cause: This happens when the job successfully marks its start (timestamp_start is saved to ClickHouse) but fails before completing the batch (e.g., error loading timestamp_progress, network failure, or application crash). The job's safety mechanism prevents concurrent runs by checking that the previous run completed.

Solution:

Warning: Manual INSERT operations into the ETL state table can violate job invariants and may cause data reprocessing or inconsistent state. Use these operations only when necessary and ensure you understand the implications. After manual intervention, verify that the job processes correctly on the next run.

Set timestamp_end value in ClickHouse ETL table to a value greater than timestamp_start. Since the table uses ReplacingMergeTree with ORDER BY (timestamp_start), you need to insert a record with the same timestamp_start as the running job. This marks the previous job as completed and allows the new job to start:

# Get timestamp_start from the running job (last record with timestamp_end IS NULL)
TIMESTAMP_START=$(clickhouse-client --query "SELECT timestamp_start FROM default.etl FINAL WHERE timestamp_end IS NULL ORDER BY timestamp_start DESC LIMIT 1 FORMAT TabSeparated")
TIMESTAMP_END=$(date +%s)
clickhouse-client --query "INSERT INTO default.etl (timestamp_start, timestamp_end) VALUES (toDateTime($TIMESTAMP_START), toDateTime($TIMESTAMP_END))"

Or using HTTP interface:

# Get timestamp_start from the running job
TIMESTAMP_START=$(curl -s "http://clickhouse:8123/?query=SELECT+timestamp_start+FROM+default.etl+FINAL+WHERE+timestamp_end+IS+NULL+ORDER+BY+timestamp_start+DESC+LIMIT+1+FORMAT+TabSeparated")
TIMESTAMP_END=$(date +%s)
curl -X POST "http://clickhouse:8123/?query=INSERT+INTO+default.etl+(timestamp_start,timestamp_end)+VALUES+(toDateTime($TIMESTAMP_START),toDateTime($TIMESTAMP_END))"

Note: After setting timestamp_end, the job will be able to pass the start check on the next run. However, if this was the first run and the job never completed successfully, timestamp_progress may be missing. In that case, the job will fail with "TimestampProgress not found in ClickHouse" error. You will need to set timestamp_progress manually as described in the "TimestampProgress Not Found in ClickHouse" section below.

If there was a previous successful run, the job will continue from the last successful timestamp_progress value (which is stored in ClickHouse).

TimestampProgress Not Found in ClickHouse

Symptoms:

• Job logs show: "TimestampProgress not found in ClickHouse"
• Job fails to start with error: "TimestampProgress not found in ClickHouse"

Cause:

This happens when:

• This is the first run and the job has never completed successfully (no timestamp_progress was ever set)
• The timestamp_progress value was deleted from ClickHouse ETL table
• The ETL table doesn't exist or is empty

Solution:

Set all three required fields in ClickHouse ETL table: timestamp_start, timestamp_end, and timestamp_progress. The timestamp_start and timestamp_end values are "fake" (dummy values) for initialization, but they must be set to satisfy state validation. Only timestamp_progress is the real value that determines the starting point for processing.

Convert any date/time to Unix timestamp:

Current time:

export TIMESTAMP_PROGRESS=$(date +%s)

Specific date (Linux):

export TIMESTAMP_PROGRESS=$(date -d "2024-01-01 00:00:00" +%s)

Specific date (macOS):

export TIMESTAMP_PROGRESS=$(date -j -f "%Y-%m-%d %H:%M:%S" "2024-01-01 00:00:00" +%s)

N days ago (Linux):

export TIMESTAMP_PROGRESS=$(date -d "30 days ago" +%s)

N days ago (macOS):

export TIMESTAMP_PROGRESS=$(date -v-30d +%s)

Set all three fields in ClickHouse (using dummy values for start and end):

# Use timestamp_progress as base, set start to same value, end to start + 1
# Convert Unix timestamps to DateTime using toDateTime() function
clickhouse-client --query "INSERT INTO default.etl (timestamp_start, timestamp_end, timestamp_progress) VALUES (toDateTime($TIMESTAMP_PROGRESS), toDateTime($((TIMESTAMP_PROGRESS + 1))), toDateTime($TIMESTAMP_PROGRESS))"

Or using HTTP interface:

curl -X POST "http://clickhouse:8123/?query=INSERT+INTO+default.etl+(timestamp_start,timestamp_end,timestamp_progress)+VALUES+(toDateTime($TIMESTAMP_PROGRESS),toDateTime($((TIMESTAMP_PROGRESS+1))),toDateTime($TIMESTAMP_PROGRESS))"

Note: After setting all three fields, the job will be able to start on the next run. The job will process data starting from timestamp_progress value. The timestamp_start and timestamp_end values are dummy values for initialization and will be replaced with real values on the first successful run.