Add SiphonFinishedBuildsFinder for CI job analytics (!242689) · Merge requests · GitLab.org / GitLab

What does this MR do and why?

This MR adds ClickHouse::Finders::Ci::SiphonFinishedBuildsFinder, a finder that reads finished CI builds from the siphon_p_ci_builds ClickHouse table — the Siphon CDC replica of the Postgres p_ci_builds table — as the siphon-backed counterpart to the existing FinishedBuildsDeduplicatedFinder.

siphon_p_ci_builds is a ReplacingMergeTree that holds every replicated version of every build (one row per _siphon_replicated_at), so the finder:

Deduplicates to the latest version per (id, partition_id) using argMax over _siphon_replicated_at.
Restricts results to finished Ci::Build rows (status IN the completed statuses, type = 'Ci::Build', _siphon_deleted = 0), mirroring the legacy flow that only seeds ci_finished_builds once a build reaches a finished status.
Supports container scoping (project exact-match / group prefix-match), job-name and date filters, pipeline source/ref filters via a commit_id IN (...) subquery, an opt-in stages LEFT JOIN for stage_name, and aggregations (mean_duration, percentiles, status rates, counts).

Because siphon_p_ci_builds has no duration column, duration aggregations are derived from age('ms', started_at, finished_at). Shared duration and group-by logic is hoisted into FinishedBuildsAggregations behind field_expression / duration_expression hooks, so the siphon finder only overrides the parts that genuinely differ.

The finder is not wired into any caller yet; this MR delivers the finder and its test coverage in isolation.

Database changes

No schema migrations. This adds a new read path against existing ClickHouse siphon_p_ci_* tables. The emitted SQL (CTE-wrapped dedup + soft-delete + finished-status filter, optional stages join scoped by traversal_path and the builds CTE's stage_id set) is asserted in spec/lib/click_house/finders/ci/siphon_finished_builds_finder_spec.rb and can be inspected with the validation script below.

References

Related to #601321 (closed)

Screenshots or screen recordings

N/A — no UI changes.

How to set up and validate locally

Prerequisites:

Start ClickHouse: gdk start clickhouse
Apply ClickHouse migrations: bundle exec rake gitlab:clickhouse:migrate

Then copy the script below into a local file (for example /tmp/verify_siphon.rb) and run it from your GDK gitlab/ directory:

bundle exec rails runner /tmp/verify_siphon.rb

The script seeds a throwaway project + builds into Postgres and the siphon_p_ci_* ClickHouse tables, exercises the finder's public surface, prints the emitted SQL alongside the results, and cleans everything up in an ensure block. Compare each printed result against its EXPECT: annotation:

Scenario	Method(s)	Expected
Aggregate by job name	`for_container` + `select(:name)` + `total_count`	`compile=2`, `rspec=2`; running build and `Ci::Bridge` excluded
Duration from timestamps	`mean_duration`	`rspec` ~ 4.0s
Success/failed rates	`rate_of_success` / `rate_of_failed`	`rspec` 50% / 50%
Stage name join	`with_stages` + `select(:name, :stage_name)`	`compile -> build`, `rspec -> test`
Pipeline filter	`filter_by_pipeline_attrs(source: 'web')`	only `compile`, `total_count=1`
Date window	`within_dates`	empty window -> 0 rows; full window -> `compile`, `rspec`
Soft-delete + dedup	newer `_siphon_replicated_at` with `_siphon_deleted`	`rspec` count drops to 1

verify_siphon.rb

# frozen_string_literal: true

# Validation script for ClickHouse::Finders::Ci::SiphonFinishedBuildsFinder
#
# Usage:
#   1. Copy this whole script into a local file, e.g. /tmp/verify_siphon.rb
#   2. Run it from your GDK gitlab/ directory:
#        bundle exec rails runner /tmp/verify_siphon.rb
#   3. Compare each printed result against its `EXPECT:` annotation.
#
# Prerequisites:
#   - GDK ClickHouse running (gdk start clickhouse)
#   - ClickHouse migrations applied:
#       bundle exec rake gitlab:clickhouse:migrate
#
# What it does:
#   1. Creates a throwaway project, pipelines, stages and builds in Postgres.
#   2. Seeds the corresponding rows into the siphon_p_ci_* ClickHouse tables
#      (mirroring what the Siphon CDC replication would do).
#   3. Exercises the finder's public surface (for_container, select,
#      aggregations, with_stages, filter_by_pipeline_attrs, within_dates) and
#      prints the emitted SQL + results so a reviewer can eyeball correctness.
#   4. Cleans up all Postgres + ClickHouse rows it created.

require 'securerandom'

finder_class = ClickHouse::Finders::Ci::SiphonFinishedBuildsFinder

def section(title)
  puts "\n#{'=' * 78}\n#{title}\n#{'=' * 78}"
end

def quote(value)
  case value
  when nil then 'NULL'
  when Time, ActiveSupport::TimeWithZone then "'#{value.utc.strftime('%Y-%m-%d %H:%M:%S.%6N')}'"
  when true then '1'
  when false then '0'
  when Numeric then value.to_s
  else "'#{value.to_s.gsub("'", "\\\\'")}'"
  end
end

def insert_rows(table, rows)
  return if rows.empty?

  columns = rows.first.keys
  values = rows.map { |row| "(#{columns.map { |c| quote(row[c]) }.join(', ')})" }.join(', ')
  ClickHouse::Client.execute("INSERT INTO #{table} (#{columns.join(', ')}) VALUES #{values}", :main)
end

def seed_builds(builds, replicated_at:, deleted: false)
  insert_rows('siphon_p_ci_builds', builds.map do |build|
    project = build.project
    {
      id: build.id,
      partition_id: build.partition_id || 100,
      project_id: project&.id || 0,
      commit_id: build.commit_id || build.pipeline_id,
      status: build.status,
      name: build.name,
      stage_id: build.stage_id,
      type: build.type || 'Ci::Build',
      started_at: build.started_at,
      finished_at: build.finished_at,
      created_at: build.created_at,
      updated_at: build.updated_at || build.created_at,
      traversal_path: project.project_namespace.traversal_path(with_organization: true),
      _siphon_replicated_at: replicated_at,
      _siphon_deleted: deleted
    }
  end)
end

def seed_stages(stages, replicated_at:)
  insert_rows('siphon_p_ci_stages', stages.map do |stage|
    project = stage.project
    {
      id: stage.id,
      partition_id: stage.partition_id || 100,
      project_id: project&.id || 0,
      pipeline_id: stage.pipeline_id,
      name: stage.name,
      status: stage.try(:status_value) || 0,
      position: stage.position,
      created_at: stage.created_at,
      updated_at: stage.updated_at || stage.created_at,
      traversal_path: project.project_namespace.traversal_path(with_organization: true),
      _siphon_replicated_at: replicated_at,
      _siphon_deleted: false
    }
  end)
end

def seed_pipelines(pipelines, replicated_at:)
  insert_rows('siphon_p_ci_pipelines', pipelines.map do |pipeline|
    project = pipeline.project
    {
      id: pipeline.id,
      partition_id: pipeline.partition_id || 100,
      project_id: project&.id || 0,
      ref: pipeline.ref,
      source: ::Ci::Pipeline.sources[pipeline.source],
      status: pipeline.status,
      started_at: pipeline.started_at,
      finished_at: pipeline.finished_at,
      created_at: pipeline.created_at,
      updated_at: pipeline.updated_at || pipeline.created_at,
      traversal_path: project.project_namespace.traversal_path(with_organization: true),
      _siphon_replicated_at: replicated_at,
      _siphon_deleted: false
    }
  end)
end

uid = SecureRandom.hex(4)
base_time = Time.current.utc.beginning_of_minute
now = Time.current

project = nil

begin
  section "1. Building Postgres fixtures (project, pipelines, stages, builds)"

  namespace = FactoryBot.create(:namespace, path: "siphon-finder-#{uid}")
  project = FactoryBot.create(:project, namespace: namespace, path: "siphon-finder-#{uid}")

  pipeline_push = FactoryBot.create(:ci_pipeline, project: project, ref: 'master', source: 'push',
    started_at: base_time, finished_at: base_time + 30.seconds)
  pipeline_web = FactoryBot.create(:ci_pipeline, project: project, ref: 'feature', source: 'web',
    started_at: base_time, finished_at: base_time + 30.seconds)

  stage_build = FactoryBot.create(:ci_stage, project: project, pipeline: pipeline_push, name: 'build')
  stage_test  = FactoryBot.create(:ci_stage, project: project, pipeline: pipeline_push, name: 'test')

  b_compile = FactoryBot.create(:ci_build, :success, project: project, pipeline: pipeline_push,
    ci_stage: stage_build, name: 'compile', started_at: base_time, finished_at: base_time + 2.seconds)
  b_rspec_failed = FactoryBot.create(:ci_build, :failed, project: project, pipeline: pipeline_push,
    ci_stage: stage_test, name: 'rspec', started_at: base_time, finished_at: base_time + 5.seconds)
  b_rspec_success = FactoryBot.create(:ci_build, :success, project: project, pipeline: pipeline_push,
    ci_stage: stage_test, name: 'rspec', started_at: base_time, finished_at: base_time + 3.seconds)
  b_web_compile = FactoryBot.create(:ci_build, :success, project: project, pipeline: pipeline_web,
    ci_stage: stage_build, name: 'compile', started_at: base_time, finished_at: base_time + 1.second)
  b_running = FactoryBot.create(:ci_build, :running, project: project, pipeline: pipeline_push,
    ci_stage: stage_build, name: 'deploy', started_at: base_time, finished_at: nil)
  bridge = FactoryBot.create(:ci_bridge, project: project, pipeline: pipeline_push,
    ci_stage: stage_build, name: 'trigger-downstream', started_at: base_time, finished_at: base_time + 1.second)

  puts "project: #{project.full_path} (id=#{project.id})"
  puts "traversal_path: #{project.project_namespace.traversal_path(with_organization: true)}"
  puts "builds: compile(success), rspec(failed), rspec(success), compile/web(success), deploy(running), bridge"

  section "2. Seeding siphon_p_ci_* ClickHouse tables"

  seed_builds([b_compile, b_rspec_failed, b_rspec_success, b_web_compile, b_running, bridge], replicated_at: now)
  seed_stages([stage_build, stage_test], replicated_at: now)
  seed_pipelines([pipeline_push, pipeline_web], replicated_at: now)
  puts "seeded 6 builds, 2 stages, 2 pipelines"

  section "3a. Aggregate by job name — total_count (Ci::Build only, finished only)"
  finder = finder_class.for_container(project).select(:name).total_count
  puts "SQL:\n#{finder.to_sql}\n"
  rows = finder.execute
  rows.each { |r| puts "  name=#{r['name'].inspect} total_count=#{r['total_count']}" }
  puts "EXPECT: compile=2, rspec=2; 'deploy' (running) and 'trigger-downstream' (bridge) absent"

  section "3b. mean_duration from timestamps (no `duration` column on siphon_p_ci_builds)"
  finder = finder_class.for_container(project).select(:name).mean_duration
  puts "SQL:\n#{finder.to_sql}\n"
  finder.execute.each { |r| puts "  name=#{r['name'].inspect} mean_duration=#{r['mean_duration']}s" }
  puts "EXPECT: rspec ≈ (5s + 3s) / 2 = 4.0s".encode('UTF-8')

  section "3c. rate_of_success / rate_of_failed"
  finder = finder_class.for_container(project).select(:name).rate_of_success.rate_of_failed
  finder.execute.each do |r|
    puts "  name=#{r['name'].inspect} success=#{r['rate_of_success']}% failed=#{r['rate_of_failed']}%"
  end
  puts "EXPECT: rspec success=50%, failed=50%"

  section "3d. with_stages — stage_name via the stages LEFT JOIN"
  finder = finder_class.for_container(project).with_stages(project).select(:name, :stage_name).total_count
  puts "SQL:\n#{finder.to_sql}\n"
  finder.execute.each { |r| puts "  name=#{r['name'].inspect} stage_name=#{r['stage_name'].inspect}" }
  puts "EXPECT: compile->build, rspec->test"

  section "3e. filter_by_pipeline_attrs — source: 'web'"
  finder = finder_class.for_container(project).select(:name).total_count
    .filter_by_pipeline_attrs(container: project, from_time: base_time - 1.hour, source: 'web')
  finder.execute.each { |r| puts "  name=#{r['name'].inspect} total_count=#{r['total_count']}" }
  puts "EXPECT: only compile (from the web pipeline), total_count=1"

  section "3f. within_dates — empty window vs full window"
  empty = finder_class.for_container(project).select(:name).total_count
    .within_dates(base_time + 10.seconds, base_time + 1.hour).execute
  full = finder_class.for_container(project).select(:name).total_count
    .within_dates(base_time - 1.minute, base_time + 1.hour).execute
  puts "  empty window rows: #{empty.size} (EXPECT 0)"
  full_names = full.map { |r| r['name'] } # rubocop:disable Rails/Pluck -- CH rows are plain Hashes
  puts "  full window names: #{full_names.sort.inspect} (EXPECT compile, rspec)"

  section "3g. soft-delete + dedup by _siphon_replicated_at"
  seed_builds([b_rspec_failed], replicated_at: now + 1.minute, deleted: true)
  rows = finder_class.for_container(project).select(:name).total_count.execute
  rspec = rows.find { |r| r['name'] == 'rspec' }
  puts "  rspec total_count after soft-deleting 1 failed build: #{rspec['total_count']} (EXPECT 1)"

  section "DONE — review the SQL + results above"
ensure
  section "Cleanup"

  if project
    path = project.project_namespace.traversal_path(with_organization: true)
    # Lightweight DELETE is blocked on these tables (they carry projections),
    # so use a synchronous heavy mutation instead.
    %w[siphon_p_ci_builds siphon_p_ci_stages siphon_p_ci_pipelines].each do |table|
      ClickHouse::Client.execute(
        "ALTER TABLE #{table} DELETE WHERE traversal_path = #{quote(path)} SETTINGS mutations_sync = 1", :main
      )
    end
    puts "deleted ClickHouse rows for #{path}"
    project.namespace.destroy!
    puts "destroyed Postgres project + namespace"
  end
end

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist.

Suggested labels

~"type::maintenance" ~backend ~database ~"ClickHouse" ~"group::ci platform" ~"devops::verify"

Add SiphonFinishedBuildsFinder for CI job analytics