Pipeline Timing

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

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:

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:

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:

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.