# Pipeline Timing

LLMS index: [llms.txt](/llms.txt) | Full content: [llms-full.txt](/llms-full.txt)

---

Independent SQS worker for queryable RPM delivery timing and cross-system lineage.
It shares the Status events database but owns only the `pipeline_timing` schema.

## Features

- Current state, timestamps, and separate attempts for GitLab jobs and Konflux PipelineRuns
- Commit, merge request, PipelineRun, TaskRun, Snapshot, and Release lineage
- Deterministic S3 event references without raw JSON in reporting tables
- Alembic-managed schema, dedicated queue, and worker metrics

## Prerequisites

- Python 3.11+ and PostgreSQL 16
- A dedicated timing SQS queue subscribed to the environment's SNS topic with
  `RawMessageDelivery: false`
- The S3 archive bucket for the same environment

The existing Status worker and event forwarders keep their own images and queues.
Timeline queue provisioning, IAM, and alerts are owned by HUM-7535.

## Installation

```bash
cd pipeline-timing
pip install -e ".[dev]"
```

The `pipeline-timing` Containerfile builds the separate
`quay.io/hummingbird-ci/pipeline-timing` image. Both the initializer and worker
run from this image.

## Usage

### Schema initialization

Run the initializer as the init container of the single-replica timing worker
Deployment. Use a `Recreate` strategy so Kubernetes stops the old worker before
starting a Pod whose init container may upgrade the schema:

```bash
DATABASE_URL="$DATABASE_URL" python3 -m pipeline_timing.worker.init_db
```

The initializer creates the `pipeline_timing` schema and runs
`alembic upgrade head` each time a Pod starts. This is a no-op at the expected
revision; a new image can apply a forward migration during rollout. If migration
fails, the initializer exits nonzero and Kubernetes does not start the worker.
The worker checks the revision again before polling. Fix migration errors with
a forward revision; do not downgrade or drop the schema during rollback.

### Timing worker

Deploy one worker replica with the environment's dedicated queue and archive
bucket. Kubernetes starts it only after the init container succeeds:

```bash
SQS_QUEUE_URL="$TIMING_QUEUE_URL" \
  S3_BUCKET="$TIMING_ARCHIVE_BUCKET" python3 -m pipeline_timing.worker
```

The worker polls when it starts. Accepted events write the event reference and
reporting rows in one committed transaction before SQS acknowledgement. Malformed
events remain for retry and eventual dead-letter-queue redrive; database or
revision failures stop polling without affecting legacy Status. Scale the timing
worker to zero replicas to pause ingestion. The queue retains messages for 14
days, so monitor queue age while paused. A new rollout uses `Recreate`: the old
worker stops, Alembic upgrades, and the new worker starts. The S3 archiver
receives its own SNS copy, so a database reference may exist briefly before its
archive object; monitor archiver failures separately.

## Configuration

| Variable        | Default | Description                                |
| --------------- | ------- | ------------------------------------------ |
| `DATABASE_URL`  | -       | Existing Status events database URL        |
| `SQS_QUEUE_URL` | -       | Environment-specific timing queue URL      |
| `S3_BUCKET`     | -       | Environment-specific SNS archive bucket    |
| `SQS_WAIT_TIME` | `20`    | SQS long-poll interval in seconds          |
| `METRICS_PORT`  | `9090`  | Prometheus port when the worker is enabled |

`POSTGRES_URL` is also accepted when `DATABASE_URL` is not set. For local
development only, the database URL defaults to
`postgresql://postgres:dev@localhost:5432/events`.

### Schema and lineage

| Table                   | Identity                       | Purpose                              |
| ----------------------- | ------------------------------ | ------------------------------------ |
| `events`                | SNS message ID                 | Source, timestamp, S3 bucket and key |
| `merge_requests`        | GitLab project ID and MR IID   | Current state and lifecycle times    |
| `commits`               | GitLab project ID and SHA      | Commit metadata                      |
| `merge_request_commits` | Project ID, IID, SHA, relation | Head and merge-commit links          |
| `gitlab_pipelines`      | Project ID and pipeline ID     | Pipeline status and timing           |
| `gitlab_jobs`           | Project ID and job ID          | Individual job attempts              |
| `pipeline_runs`         | Cluster and Kubernetes UID     | Build, test, and Release attempts    |
| `task_runs`             | Cluster and Kubernetes UID     | Task timing and parent UID           |
| `snapshots`             | Cluster and Kubernetes UID     | Build-to-test bridge                 |
| `releases`              | Cluster and Kubernetes UID     | Snapshot-to-Release bridge           |

Every domain row stores `latest_event_id` pointing to `events`. The schema
stores selected reporting fields, not raw event JSON. Join MR head and merge
commits to GitLab pipelines by canonical HTTPS repository URL and SHA; join
jobs by project and pipeline ID. Konflux PipelineRuns use the same repository
URL and SHA, then fan out by component and execution UID. TaskRuns link through
PipelineRun owner UIDs; Snapshot source-run names, test-run Snapshot labels,
and Release references connect the remaining phases. The Release completion
time closes the current timeline. Distinct job IDs and PipelineRun UIDs retain
each retry's status and timing.

Metrics include `hummingbird_pipeline_timing_records_total{kind,result}`,
`hummingbird_pipeline_timing_failures_total{reason}`,
`hummingbird_pipeline_timing_missing_lineage_total{kind}`, and
`hummingbird_pipeline_timing_worker_stopped`. Monitor timing queue age and
dead-letter-queue depth separately from the legacy Status queue. Fix malformed
events before redriving them; SNS message IDs make processing idempotent.

## Development

Run tests against the PostgreSQL service provided in CI or a local database:

```bash
DATABASE_URL=postgresql://postgres:dev@localhost:5432/events \
  python3 -m unittest discover -s tests -t .
```

Tests also load the canonical archive-key fixture in
`sns-s3-archiver/tests/fixtures/archive-key-contract.json`. CI provides
`DATABASE_URL` for the PostgreSQL integration tests. The timing image has a
Kubernetes import smoke test in `tests-k8s.yml`.

## License

This project is licensed under the GNU General Public License v3.0 or later.
See the [LICENSE](https://gitlab.com/redhat/hummingbird/tools/-/blob/main/LICENSE).