Introduce Ci::PipelineCreation::StartPipelineService class

Introduce Ci::PipelineCreation::StartPipelineService to drop not
runnable builds.

app/models/ci/pipeline.rb

@@ -1246,6 +1246,10 @@ module Ci
       end
     end
 
+    def build_matchers
+      self.builds.build_matchers(project)
+    end
+
     private
 
     def add_message(severity, content)

app/models/ci/runner.rb

@@ -433,13 +433,7 @@ module Ci
     end
 
     def matches_build?(build)
-      return false if self.ref_protected? && !build.protected?
-
-      accepting_tags?(build)
-    end
-
-    def accepting_tags?(build)
-      (run_untagged? || build.has_tags?) && (build.tag_list - tag_list).empty?
+      runner_matcher.matches?(build.build_matcher)
     end
   end
 end

app/models/commit_status.rb

@@ -174,8 +174,11 @@ class CommitStatus < ApplicationRecord
       next if commit_status.processed?
       next unless commit_status.project
 
+      last_arg = transition.args.last
+      transition_options = last_arg.is_a?(Hash) && last_arg.extractable_options? ? last_arg : {}
+
       commit_status.run_after_commit do
-        PipelineProcessWorker.perform_async(pipeline_id)
+        PipelineProcessWorker.perform_async(pipeline_id) unless transition_options[:skip_pipeline_processing]
         ExpireJobCacheWorker.perform_async(id)
       end
     end

app/models/concerns/enums/ci/commit_status.rb

@@ -24,6 +24,7 @@ module Enums
           project_deleted: 15,
           ci_quota_exceeded: 16,
           pipeline_loop_detected: 17,
+          no_matching_runner: 18,
           insufficient_bridge_permissions: 1_001,
           downstream_bridge_project_not_found: 1_002,
           invalid_bridge_trigger: 1_003,

app/presenters/commit_status_presenter.rb

@@ -25,7 +25,8 @@ class CommitStatusPresenter < Gitlab::View::Presenter::Delegated
     reached_max_descendant_pipelines_depth: 'You reached the maximum depth of child pipelines',
     project_deleted: 'The job belongs to a deleted project',
     user_blocked: 'The user who created this job is blocked',
-    ci_quota_exceeded: 'No more CI minutes available'
+    ci_quota_exceeded: 'No more CI minutes available',
+    no_matching_runner: 'No matching runner available'
   }.freeze
 
   private_constant :CALLOUT_FAILURE_MESSAGES

app/services/ci/pipeline_creation/drop_not_runnable_builds_service.rb

+# frozen_string_literal: true
+
+module Ci
+  module PipelineCreation
+    class DropNotRunnableBuildsService
+      include Gitlab::Utils::StrongMemoize
+
+      def initialize(pipeline)
+        @pipeline = pipeline
+      end
+
+      ##
+      # We want to run this service exactly once,
+      # before the first pipeline processing call
+      #
+      def execute
+        return unless ::Feature.enabled?(:ci_drop_new_builds_when_ci_quota_exceeded, project, default_enabled: :yaml)
+        return unless pipeline.created?
+
+        load_runners
+        validate_build_matchers
+      end
+
+      private
+
+      attr_reader :pipeline
+      attr_reader :instance_runners, :private_runners
+      delegate :project, to: :pipeline
+
+      def load_runners
+        @instance_runners, @private_runners = project
+          .all_runners
+          .runner_matchers
+          .partition(&:instance_type?)
+      end
+
+      def validate_build_matchers
+        pipeline.build_matchers.each do |build_matcher|
+          failure_reason = validate_build_matcher(build_matcher)
+          next unless failure_reason
+
+          drop_all_builds(build_matcher.build_ids, failure_reason)
+        end
+      end
+
+      def validate_build_matcher(build_matcher)
+        return if matching_private_runners?(build_matcher)
+        return if matching_instance_runners_available?(build_matcher)
+
+        matching_failure_reason(build_matcher)
+      end
+
+      ##
+      # We skip pipeline processing until we drop all required builds. Otherwise
+      # as we drop the first build, the remaining builds to be dropped could
+      # transition to other states by `PipelineProcessWorker` running async.
+      #
+      def drop_all_builds(build_ids, failure_reason)
+        pipeline.builds.id_in(build_ids).each do |build|
+          build.drop(failure_reason, skip_pipeline_processing: true)
+        end
+      end
+
+      def matching_private_runners?(build_matcher)
+        private_runners
+          .find { |matcher| matcher.matches?(build_matcher) }
+          .present?
+      end
+
+      # Overridden in EE to include more conditions
+      def matching_instance_runners_available?(build_matcher)
+        matching_instance_runners?(build_matcher)
+      end
+
+      def matching_instance_runners?(build_matcher)
+        instance_runners
+          .find { |matcher| matcher.matches?(build_matcher) }
+          .present?
+      end
+
+      # Overridden in EE
+      def matching_failure_reason(build_matcher)
+        :no_matching_runner
+      end
+    end
+  end
+end
+
+Ci::PipelineCreation::DropNotRunnableBuildsService.prepend_mod_with('Ci::PipelineCreation::DropNotRunnableBuildsService')

app/services/ci/pipeline_creation/start_pipeline_service.rb

+# frozen_string_literal: true
+
+module Ci
+  module PipelineCreation
+    class StartPipelineService
+      attr_reader :pipeline
+
+      def initialize(pipeline)
+        @pipeline = pipeline
+      end
+
+      def execute
+        DropNotRunnableBuildsService.new(pipeline).execute
+        Ci::ProcessPipelineService.new(pipeline).execute
+      end
+    end
+  end
+end

app/workers/ci/initial_pipeline_process_worker.rb

@@ -15,7 +15,7 @@ module Ci
 
     def perform(pipeline_id)
       Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline|
-        Ci::ProcessPipelineService
+        Ci::PipelineCreation::StartPipelineService
           .new(pipeline)
           .execute
       end

config/feature_flags/development/ci_drop_new_builds_when_ci_quota_exceeded.yml

+---
+name: ci_drop_new_builds_when_ci_quota_exceeded
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/61166
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/326709
+milestone: '14.0'
+type: development
+group: group::continuous integration
+default_enabled: false

ee/app/services/ee/ci/pipeline_creation/drop_not_runnable_builds_service.rb

+# frozen_string_literal: true
+
+module EE
+  module Ci
+    module PipelineCreation
+      module DropNotRunnableBuildsService
+        extend ::Gitlab::Utils::Override
+
+        private
+
+        override :matching_instance_runners_available?
+        def matching_instance_runners_available?(build_matcher)
+          instance_runners
+            .find { |matcher| matcher.matches?(build_matcher) && matcher.matches_quota?(build_matcher) }
+            .present?
+        end
+
+        override :matching_failure_reason
+        def matching_failure_reason(build_matcher)
+          if matching_instance_runners?(build_matcher)
+            :ci_quota_exceeded
+          else
+            :no_matching_runner
+          end
+        end
+      end
+    end
+  end
+end

ee/spec/features/merge_trains/two_merge_requests_on_train_spec.rb

@@ -8,6 +8,7 @@ RSpec.describe 'Two merge requests on a merge train' do
   let(:project) { create(:project, :repository) }
   let_it_be(:maintainer_1) { create(:user) }
   let_it_be(:maintainer_2) { create(:user) }
+  let_it_be(:runner) { create(:ci_runner) }
 
   let(:merge_request_1) do
     create(:merge_request,

ee/spec/features/merge_trains/user_adds_merge_request_to_merge_train_spec.rb

@@ -64,6 +64,8 @@ RSpec.describe 'User adds a merge request to a merge train', :js do
     end
 
     context 'when pipeline for merge train succeeds', :sidekiq_might_not_need_inline do
+      let_it_be(:runner) { create(:ci_runner) }
+
       before do
         visit project_merge_request_path(project, merge_request)
         merge_request.merge_train.pipeline.builds.map(&:success!)

ee/spec/services/ci/create_pipeline_service/needs_spec.rb

@@ -75,6 +75,7 @@ RSpec.describe Ci::CreatePipelineService do
     end
 
     it 'creates a pipeline with regular_job and bridge_dag_job pending' do
+      create(:ci_runner)
       pipeline = create_pipeline!
       processables = pipeline.processables
       Ci::InitialPipelineProcessWorker.new.perform(pipeline.id)

ee/spec/services/ci/pipeline_creation/drop_not_runnable_builds_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PipelineCreation::DropNotRunnableBuildsService do
+  let_it_be_with_reload(:pipeline) do
+    create(:ci_pipeline, status: :created)
+  end
+
+  let_it_be_with_reload(:job) do
+    create(:ci_build, project: pipeline.project, pipeline: pipeline)
+  end
+
+  let_it_be(:instance_runner) do
+    create(:ci_runner,
+      runner_type: :instance_type,
+      public_projects_minutes_cost_factor: 0,
+      private_projects_minutes_cost_factor: 1)
+  end
+
+  describe '#execute' do
+    subject(:execute) { described_class.new(pipeline).execute }
+
+    context 'with public projects' do
+      before do
+        pipeline.project.update!(visibility_level: ::Gitlab::VisibilityLevel::PUBLIC)
+      end
+
+      context 'with available CI quota' do
+        it 'does not drop the jobs' do
+          expect { execute }.not_to change { job.reload.status }
+        end
+      end
+
+      context 'when the Ci quota is exceeded' do
+        before do
+          allow(pipeline.project).to receive(:ci_minutes_quota)
+            .and_return(double('quota', minutes_used_up?: true))
+        end
+
+        it 'does not drop the jobs' do
+          expect { execute }.not_to change { job.reload.status }
+        end
+      end
+    end
+
+    context 'with internal projects' do
+      before do
+        pipeline.project.update!(visibility_level: ::Gitlab::VisibilityLevel::INTERNAL)
+      end
+
+      context 'with available CI quota' do
+        it 'does not drop the jobs' do
+          expect { execute }.not_to change { job.reload.status }
+        end
+      end
+
+      context 'when the Ci quota is exceeded' do
+        before do
+          expect(pipeline.project).to receive(:ci_minutes_quota)
+            .and_return(double('quota', minutes_used_up?: true))
+        end
+
+        it 'drops the job' do
+          execute
+          job.reload
+
+          expect(job).to be_failed
+          expect(job.failure_reason).to eq('ci_quota_exceeded')
+        end
+      end
+    end
+
+    context 'with private projects' do
+      before do
+        pipeline.project.update!(visibility_level: ::Gitlab::VisibilityLevel::PRIVATE)
+      end
+
+      context 'with available CI quota' do
+        it 'does not drop the jobs' do
+          expect { execute }.not_to change { job.reload.status }
+        end
+      end
+
+      context 'when the Ci quota is exceeded' do
+        before do
+          expect(pipeline.project).to receive(:ci_minutes_quota)
+            .and_return(double('quota', minutes_used_up?: true))
+        end
+
+        it 'drops the job' do
+          execute
+          job.reload
+
+          expect(job).to be_failed
+          expect(job.failure_reason).to eq('ci_quota_exceeded')
+        end
+      end
+    end
+  end
+end

ee/spec/services/ci/process_pipeline_service_spec.rb

@@ -29,22 +29,48 @@ RSpec.describe Ci::ProcessPipelineService, '#execute' do
       stub_ci_pipeline_to_return_yaml_file
     end
 
-    it 'creates a downstream cross-project pipeline' do
-      service.execute
-      Sidekiq::Worker.drain_all
+    context 'when there is a runner available' do
+      let_it_be(:runner) do
+        create(:ci_runner, tag_list: %w[ruby postgres mysql])
+      end
 
-      expect_statuses(%w[test pending], %w[cross created], %w[deploy created])
+      it 'creates a downstream cross-project pipeline' do
+        service.execute
+        Sidekiq::Worker.drain_all
 
-      update_build_status(:test, :success)
-      Sidekiq::Worker.drain_all
+        expect_statuses(%w[test pending], %w[cross created], %w[deploy created])
 
-      expect_statuses(%w[test success], %w[cross success], %w[deploy pending])
+        update_build_status(:test, :success)
+        Sidekiq::Worker.drain_all
 
-      expect(downstream.ci_pipelines).to be_one
-      expect(downstream.ci_pipelines.first).to be_pending
-      expect(downstream.builds).not_to be_empty
-      expect(downstream.builds.first.variables)
-        .to include(key: 'BRIDGE', value: 'cross', public: false, masked: false)
+        expect_statuses(%w[test success], %w[cross success], %w[deploy pending])
+
+        expect(downstream.ci_pipelines).to be_one
+        expect(downstream.ci_pipelines.first).to be_pending
+        expect(downstream.builds).not_to be_empty
+        expect(downstream.builds.first.variables)
+          .to include(key: 'BRIDGE', value: 'cross', public: false, masked: false)
+      end
+    end
+
+    context 'with no runners' do
+      it 'creates a downstream cross-project pipeline' do
+        service.execute
+        Sidekiq::Worker.drain_all
+
+        expect_statuses(%w[test pending], %w[cross created], %w[deploy created])
+
+        update_build_status(:test, :success)
+        Sidekiq::Worker.drain_all
+
+        expect_statuses(%w[test success], %w[cross success], %w[deploy pending])
+
+        expect(downstream.ci_pipelines).to be_one
+        expect(downstream.ci_pipelines.first).to be_failed
+        expect(downstream.builds).not_to be_empty
+        expect(downstream.builds).to all be_failed
+        expect(downstream.builds.map(&:failure_reason)).to all eq('no_matching_runner')
+      end
     end
   end
 

lib/gitlab/ci/status/build/failed.rb

@@ -30,7 +30,8 @@ module Gitlab
             reached_max_descendant_pipelines_depth: 'reached maximum depth of child pipelines',
             project_deleted: 'pipeline project was deleted',
             user_blocked: 'pipeline user was blocked',
-            ci_quota_exceeded: 'no more CI minutes available'
+            ci_quota_exceeded: 'no more CI minutes available',
+            no_matching_runner: 'no matching runner available'
           }.freeze
 
           private_constant :REASONS

spec/features/merge_request/user_sees_merge_request_pipelines_spec.rb

@@ -25,6 +25,8 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
     }
   end
 
+  let_it_be(:runner) { create(:ci_runner) }
+
   before do
     stub_application_setting(auto_devops_enabled: false)
     stub_ci_pipeline_yaml_file(YAML.dump(config))

spec/lib/gitlab/ci/build/auto_retry_spec.rb

@@ -8,7 +8,7 @@ RSpec.describe Gitlab::Ci::Build::AutoRetry do
   describe '#allowed?' do
     using RSpec::Parameterized::TableSyntax
 
-    let(:build) { create(:ci_build) }
+    let(:build) { build_stubbed(:ci_build) }
 
     subject { auto_retry.allowed? }
 

spec/models/ci/pipeline_spec.rb

@@ -4512,4 +4512,17 @@ RSpec.describe Ci::Pipeline, :mailer, factory_default: :keep do
         .not_to exceed_query_limit(control_count)
     end
   end
+
+  describe '#build_matchers' do
+    let_it_be(:pipeline) { create(:ci_pipeline) }
+    let_it_be(:builds) { create_list(:ci_build, 2, pipeline: pipeline, project: pipeline.project) }
+
+    subject(:matchers) { pipeline.build_matchers }
+
+    it 'returns build matchers' do
+      expect(matchers.size).to eq(1)
+      expect(matchers).to all be_a(Gitlab::Ci::Matching::BuildMatcher)
+      expect(matchers.first.build_ids).to match_array(builds.map(&:id))
+    end
+  end
 end

spec/services/ci/create_pipeline_service/cross_project_pipeline_spec.rb

@@ -53,6 +53,8 @@ RSpec.describe Ci::CreatePipelineService, '#execute' do
     end
 
     context 'when sidekiq processes the job', :sidekiq_inline do
+      let_it_be(:runner) { create(:ci_runner) }
+
       it 'transitions to pending status and triggers a downstream pipeline' do
         pipeline = create_pipeline!
 

spec/services/ci/create_pipeline_service/needs_spec.rb

@@ -202,21 +202,37 @@ RSpec.describe Ci::CreatePipelineService do
         YAML
       end
 
-      it 'creates a pipeline with build_a and test_b pending; deploy_b manual', :sidekiq_inline do
-        processables = pipeline.processables
+      context 'when there are runners matching the builds' do
+        before do
+          create(:ci_runner)
+        end
+
+        it 'creates a pipeline with build_a and test_b pending; deploy_b manual', :sidekiq_inline do
+          processables = pipeline.processables
+
+          build_a = processables.find { |processable| processable.name == 'build_a' }
+          test_a = processables.find { |processable| processable.name == 'test_a' }
+          test_b = processables.find { |processable| processable.name == 'test_b' }
+          deploy_a = processables.find { |processable| processable.name == 'deploy_a' }
+          deploy_b = processables.find { |processable| processable.name == 'deploy_b' }
+
+          expect(pipeline).to be_created_successfully
+          expect(build_a.status).to eq('pending')
+          expect(test_a.status).to eq('created')
+          expect(test_b.status).to eq('pending')
+          expect(deploy_a.status).to eq('created')
+          expect(deploy_b.status).to eq('manual')
+        end
+      end
 
-        build_a = processables.find { |processable| processable.name == 'build_a' }
-        test_a = processables.find { |processable| processable.name == 'test_a' }
-        test_b = processables.find { |processable| processable.name == 'test_b' }
-        deploy_a = processables.find { |processable| processable.name == 'deploy_a' }
-        deploy_b = processables.find { |processable| processable.name == 'deploy_b' }
+      context 'when there are no runners matching the builds' do
+        it 'creates a pipeline with build_a and test_b pending; deploy_b manual', :sidekiq_inline do
+          processables = pipeline.processables
 
-        expect(pipeline).to be_persisted
-        expect(build_a.status).to eq('pending')
-        expect(test_a.status).to eq('created')
-        expect(test_b.status).to eq('pending')
-        expect(deploy_a.status).to eq('created')
-        expect(deploy_b.status).to eq('manual')
+          expect(pipeline).to be_created_successfully
+          expect(processables).to all be_failed
+          expect(processables.map(&:failure_reason)).to all eq('no_matching_runner')
+        end
       end
     end
 

spec/services/ci/create_pipeline_service_spec.rb

@@ -7,6 +7,7 @@ RSpec.describe Ci::CreatePipelineService do
 
   let_it_be(:project, reload: true) { create(:project, :repository) }
   let_it_be(:user, reload: true) { project.owner }
+  let_it_be(:runner) { create(:ci_runner, tag_list: %w[postgres mysql ruby]) }
 
   let(:ref_name) { 'refs/heads/master' }
 

spec/services/ci/pipeline_creation/drop_not_runnable_builds_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PipelineCreation::DropNotRunnableBuildsService do
+  let_it_be(:group) { create(:group) }
+  let_it_be(:project) { create(:project, group: group) }
+
+  let_it_be_with_reload(:pipeline) do
+    create(:ci_pipeline, project: project, status: :created)
+  end
+
+  let_it_be_with_reload(:job) do
+    create(:ci_build, project: project, pipeline: pipeline)
+  end
+
+  describe '#execute' do
+    subject(:execute) { described_class.new(pipeline).execute }
+
+    context 'when the feature flag is disabled' do
+      before do
+        stub_feature_flags(ci_drop_new_builds_when_ci_quota_exceeded: false)
+      end
+
+      it 'does not drop the jobs' do
+        expect { execute }.not_to change { job.reload.status }
+      end
+    end
+
+    context 'when the pipeline status is not created' do
+      before do
+        pipeline.update!(status: :running)
+      end
+
+      it 'does not drop the jobs' do
+        expect { execute }.not_to change { job.reload.status }
+      end
+    end
+
+    context 'when there are no runners available' do
+      it 'drops the job' do
+        execute
+        job.reload
+
+        expect(job).to be_failed
+        expect(job.failure_reason).to eq('no_matching_runner')
+      end
+    end
+
+    context 'with project runners' do
+      let_it_be(:project_runner) do
+        create(:ci_runner, runner_type: :project_type, projects: [project])
+      end
+
+      it 'does not drop the jobs' do
+        expect { execute }.not_to change { job.reload.status }
+      end
+    end
+
+    context 'with group runners' do
+      let_it_be(:group_runner) do
+        create(:ci_runner, runner_type: :group_type, groups: [group])
+      end
+
+      it 'does not drop the jobs' do
+        expect { execute }.not_to change { job.reload.status }
+      end
+    end
+
+    context 'with instance runners' do
+      let_it_be(:instance_runner) do
+        create(:ci_runner, runner_type: :instance_type)
+      end
+
+      it 'does not drop the jobs' do
+        expect { execute }.not_to change { job.reload.status }
+      end
+    end
+  end
+end

spec/services/ci/pipeline_creation/start_pipeline_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PipelineCreation::StartPipelineService do
+  let(:pipeline) { build(:ci_pipeline) }
+
+  subject(:service) { described_class.new(pipeline) }
+
+  describe '#execute' do
+    it 'calls the pipeline runners matching validation service' do
+      expect(Ci::PipelineCreation::DropNotRunnableBuildsService)
+        .to receive(:new)
+        .with(pipeline)
+        .and_return(double('service', execute: true))
+
+      service.execute
+    end
+
+    it 'calls the pipeline process service' do
+      expect(Ci::ProcessPipelineService)
+        .to receive(:new)
+        .with(pipeline)
+        .and_return(double('service', execute: true))
+
+      service.execute
+    end
+  end
+end

spec/services/ci/pipeline_processing/shared_processing_service.rb

@@ -859,6 +859,8 @@ RSpec.shared_examples 'Pipeline Processing Service' do
   end
 
   context 'when a bridge job has parallel:matrix config', :sidekiq_inline do
+    let_it_be(:runner) { create(:ci_runner) }
+
     let(:parent_config) do
       <<-EOY
       test:

spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb

@@ -3,6 +3,7 @@
 RSpec.shared_context 'Pipeline Processing Service Tests With Yaml' do
   let_it_be(:project) { create(:project, :repository) }
   let_it_be(:user)    { project.owner }
+  let_it_be(:runner)  { create(:ci_runner) }
 
   where(:test_file_path) do
     Dir.glob(Rails.root.join('spec/services/ci/pipeline_processing/test_cases/*.yml'))

spec/workers/ci/initial_pipeline_process_worker_spec.rb

@@ -4,17 +4,33 @@ require 'spec_helper'
 
 RSpec.describe Ci::InitialPipelineProcessWorker do
   describe '#perform' do
-    let_it_be(:pipeline) { create(:ci_pipeline, :with_job, status: :created) }
+    let_it_be_with_reload(:pipeline) do
+      create(:ci_pipeline, :with_job, status: :created)
+    end
 
     include_examples 'an idempotent worker' do
       let(:job_args) { pipeline.id }
 
-      it 'marks the pipeline as pending' do
+      context 'when there are runners available' do
+        before do
+          create(:ci_runner)
+        end
+
+        it 'marks the pipeline as pending' do
+          expect(pipeline).to be_created
+
+          subject
+
+          expect(pipeline.reload).to be_pending
+        end
+      end
+
+      it 'marks the pipeline as failed' do
         expect(pipeline).to be_created
 
         subject
 
-        expect(pipeline.reload).to be_pending
+        expect(pipeline.reload).to be_failed
       end
     end
   end