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. Infrastructure provisions the timeline queues, IAM, and alerts separately from the Status worker. Each environment has a separate queue named arr-hummingbird-<env>-pipeline-timing-sqs and a matching -sqs-dlq queue. The source queue retains messages for 4 days; its DLQ retains them for 14 days from the original enqueue time when SQS automatically moves a message to the DLQ. Operator redrive from the DLQ resets the enqueue timestamp. The source queue moves a message to its DLQ after 3 receives and uses a 300-second visibility timeout, matching Status. The SNS subscription preserves the SNS envelope (RawMessageDelivery: false) and does not filter events. See Hummingbird Events Topic for queue policy and redrive details. Provision both environments’ queues and SNS subscriptions before deploying the workers. Messages collected before a worker starts may expire after four days; this pre-launch loss is accepted. A queue without a consumer does not move messages to its DLQ because no receives occur. Validate the staging worker before deploying the production worker or enabling new production event sources.

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. After launch, monitor queue age against the four-day source retention limit during a pause to avoid losing events. SQS preserves the original enqueue time for automatic DLQ transfer, and an operator redrive resets it. 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.

The worker follows Status’s polling pattern: a 20-second long poll receives up to 10 messages, processed sequentially with boto3’s default request retries. Receive errors are logged and polling continues. An acknowledgement error leaves the committed event available for idempotent redelivery and processing continues with the remaining batch. Database connection or schema failures stop the worker for Kubernetes to restart it. SQS visibility controls when an unacknowledged message becomes available again.

Queue age and DLQ depth are monitored independently from the Status queue. Fix malformed messages before using the documented redrive procedure. If the worker is paused or not yet deployed, remove only a known test message after confirming that redrive returned it to the timing queue.

Configuration

Infrastructure supplies matching queue and archive-bucket values for each environment:

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

Source timestamps accept ISO 8601 and GitLab’s YYYY-MM-DD HH:MM:SS UTC format and are normalized to UTC. Timestamps without a timezone are interpreted as UTC.

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.

PipelineRun SHAs come from supported labels and annotations, including pac.test.appstudio.openshift.io/sha, before falling back to the revision parameter.

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.