Read CI job analytics from siphon behind a feature flag

Narendranrequested to merge
narendran-kannan/601321/job-analytics-siphon-integration into master

What does this MR do and why?

This wires ClickHouse::Finders::Ci::SiphonFinishedBuildsFinder (added in the parent MR) into Ci::JobAnalytics::QueryBuilder, behind the new job_analytics_siphon feature flag. When the flag is enabled for a project, the job analytics read path queries the Siphon CDC replica (siphon_p_ci_builds) instead of the ci_finished_builds-backed FinishedBuildsFinder / FinishedBuildsDeduplicatedFinder.

The two finders expose intentionally different APIs (the siphon table has no stage_name/duration columns and scopes by traversal_path), so the divergence is reconciled with small adapters inside QueryBuilder rather than by changing either finder:

  • project scope: for_container(project) vs for_project(project.id)
  • date window: within_dates(from_time, to_time) vs apply_finished_at_lower_bound(from_time)
  • pipeline attrs: container: vs project: keyword
  • stages LEFT JOIN: with_stages(project) is added only when stage_name is requested (selected or sorted). The siphon finder raises if stage_name is referenced without the join, and the join is pruning-expensive, so it is gated.

The siphon branch takes precedence over the backfill/deduplicated branch when the flag is on. The flag defaults to false.

Feature flag

  • job_analytics_siphon (beta, project-scoped, default_enabled: false)
  • Rollout issue: #604585 (closed)

References

  • Feature issue: #601321 (closed)
  • Parent MR (adds the finder): !242689 (merged) — this MR is stacked on top and targets that branch; it should be retargeted to master after the parent merges.

Database changes

No schema migrations. This adds a new read path against existing ClickHouse siphon_p_ci_* tables, gated by a feature flag. The emitted SQL (argMax 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/ci/job_analytics/query_builder_spec.rb and can be inspected with the validation script below.

Screenshots or screen recordings

N/A — no UI changes.

How to set up and validate locally

Prerequisites:

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

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

bundle exec rails runner /tmp/verify_job_analytics_siphon.rb

The script seeds a throwaway project + builds into Postgres and the siphon_p_ci_* ClickHouse tables, then shows that QueryBuilder#build_finder selects the legacy finder with the flag off and the siphon finder with the flag on, prints the emitted siphon SQL, and asserts the aggregated result. It cleans everything up in an ensure block. Expected output: FinishedBuildsFinder with the flag off, SiphonFinishedBuildsFinder with the flag on, and a single row name="rspec" stage_name="test" mean_duration=4.0s whose SQL reads from siphon_p_ci_builds with a LEFT OUTER JOIN on siphon_p_ci_stages.

verify_job_analytics_siphon.rb 
# frozen_string_literal: true

# Validation for Ci::JobAnalytics::QueryBuilder wiring behind job_analytics_siphon.
# Usage: bundle exec rails runner /tmp/verify_job_analytics_siphon.rb

require 'securerandom'

def section(t) = puts("\n#{'=' * 72}\n#{t}\n#{'=' * 72}")

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

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

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

uid = SecureRandom.hex(4)
base = Time.current.utc.beginning_of_minute
project = nil

begin
  section '1. Seed Postgres fixtures'
  ns = FactoryBot.create(:namespace, path: "ja-siphon-#{uid}")
  project = FactoryBot.create(:project, namespace: ns, path: "ja-siphon-#{uid}")
  user = FactoryBot.create(:user, username: "jauser-#{uid}", email: "jauser-#{uid}@example.com")
  project.add_maintainer(user)
  ApplicationSetting.current_without_cache.update!(use_clickhouse_for_analytics: true)
  pipeline = FactoryBot.create(:ci_pipeline, project: project, ref: 'master', source: 'push',
    started_at: base, finished_at: base + 30.seconds)
  stage = FactoryBot.create(:ci_stage, project: project, pipeline: pipeline, name: 'test')

  b1 = FactoryBot.create(:ci_build, :success, project: project, pipeline: pipeline, ci_stage: stage,
    name: 'rspec', started_at: base, finished_at: base + 3.seconds)
  b2 = FactoryBot.create(:ci_build, :failed, project: project, pipeline: pipeline, ci_stage: stage,
    name: 'rspec', started_at: base, finished_at: base + 5.seconds)
  path = project.project_namespace.traversal_path(with_organization: true)
  puts "project=#{project.full_path} builds=rspec(success 3s), rspec(failed 5s)"

  section '2. Seed siphon_p_ci_* ClickHouse tables'
  now = Time.current
  insert('siphon_p_ci_builds', [b1, b2].map do |b|
    { id: b.id, partition_id: b.partition_id || 100, project_id: project.id, commit_id: b.commit_id || pipeline.id,
      status: b.status, name: b.name, stage_id: b.stage_id, type: 'Ci::Build',
      started_at: b.started_at, finished_at: b.finished_at, created_at: b.created_at,
      updated_at: b.created_at, traversal_path: path, _siphon_replicated_at: now, _siphon_deleted: false }
  end)
  insert('siphon_p_ci_stages', [{ id: stage.id, partition_id: 100, project_id: project.id,
    pipeline_id: pipeline.id, name: stage.name, status: 0, position: stage.position,
    created_at: stage.created_at, updated_at: stage.created_at, traversal_path: path,
    _siphon_replicated_at: now, _siphon_deleted: false }])
  puts 'seeded 2 builds, 1 stage'

  opts = { select_fields: [:name, :stage_name], aggregations: [:mean_duration], from_time: base - 1.hour }

  section '3. Flag OFF -> legacy FinishedBuildsFinder'
  Feature.disable(:job_analytics_siphon)
  qb_off = Ci::JobAnalytics::QueryBuilder.new(project: project, current_user: user, options: opts)
  finder_off = qb_off.send(:build_finder)
  puts "finder=#{finder_off.class.name.split('::').last}"
  puts "EXPECT: FinishedBuildsFinder"

  section '4. Flag ON -> SiphonFinishedBuildsFinder'
  Feature.enable(:job_analytics_siphon, project)
  qb_on = Ci::JobAnalytics::QueryBuilder.new(project: project, current_user: user, options: opts)
  finder_on = qb_on.send(:build_finder)
  puts "finder=#{finder_on.class.name.split('::').last}"
  puts "EXPECT: SiphonFinishedBuildsFinder"

  section '5. Run the built query and inspect the emitted SQL + results'
  query = finder_on.final_query
  puts "SQL:\n#{query.to_sql}\n"
  rows = ClickHouse::Client.select(query, :main)
  rows.each { |r| puts "  name=#{r['name'].inspect} stage_name=#{r['stage_name'].inspect} mean_duration=#{r['mean_duration']}s" }
  puts "EXPECT: rspec, stage_name=test, mean_duration=4.0s; SQL hits siphon_p_ci_builds + stages join"
ensure
  section 'Cleanup'
  if project
    p = project.project_namespace.traversal_path(with_organization: true)
    %w[siphon_p_ci_builds siphon_p_ci_stages].each do |t|
      ClickHouse::Client.execute("ALTER TABLE #{t} DELETE WHERE traversal_path = #{quote(p)} SETTINGS mutations_sync = 1", :main)
    end
    project.namespace.destroy!
    puts 'cleaned up'
  end
end

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist.

Merge request reports