Add support for DAG

This implements the support for `needs:` keyword
as part of GitLab CI. That makes some of the jobs
to be run out of order.

9 files changed, 93 insertions, 11 deletions

diff --git a/app/models/ci/build.rb b/app/models/ci/build.rb
index 07813e03f3a..dd2bfc42af9 100644
--- a/app/models/ci/build.rb
+++ b/app/models/ci/build.rb
@@ -38,6 +38,7 @@ module Ci
     has_one :deployment, as: :deployable, class_name: 'Deployment'
     has_many :trace_sections, class_name: 'Ci::BuildTraceSection'
     has_many :trace_chunks, class_name: 'Ci::BuildTraceChunk', foreign_key: :build_id
+    has_many :needs, class_name: 'Ci::BuildNeed', foreign_key: :build_id, inverse_of: :build
 
     has_many :job_artifacts, class_name: 'Ci::JobArtifact', foreign_key: :job_id, dependent: :destroy, inverse_of: :job # rubocop:disable Cop/ActiveRecordDependent
     has_many :job_variables, class_name: 'Ci::JobVariable', foreign_key: :job_id
@@ -50,6 +51,7 @@ module Ci
 
     accepts_nested_attributes_for :runner_session
     accepts_nested_attributes_for :job_variables
+    accepts_nested_attributes_for :needs
 
     delegate :url, to: :runner_session, prefix: true, allow_nil: true
     delegate :terminal_specification, to: :runner_session, allow_nil: true
@@ -713,11 +715,21 @@ module Ci
 
       depended_jobs = depends_on_builds
 
-      return depended_jobs unless options[:dependencies].present?
+      # find all jobs that are dependent on
+      if options[:dependencies].present?
+        depended_jobs = depended_jobs.select do |job|
+          options[:dependencies].include?(job.name)
+        end
+      end
 
-      depended_jobs.select do |job|
-        options[:dependencies].include?(job.name)
+      # find all jobs that are needed by this one
+      if options[:needs].present?
+        depended_jobs = depended_jobs.select do |job|
+          options[:needs].include?(job.name)
+        end
       end
+
+      depended_jobs
     end
 
     def empty_dependencies?
diff --git a/app/models/ci/build_need.rb b/app/models/ci/build_need.rb
new file mode 100644
index 00000000000..6531dfd332f
--- /dev/null
+++ b/app/models/ci/build_need.rb
@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+module Ci
+  class BuildNeed < ApplicationRecord
+    extend Gitlab::Ci::Model
+
+    belongs_to :build, class_name: "Ci::Build", foreign_key: :build_id, inverse_of: :needs
+
+    validates :build, presence: true
+    validates :name, presence: true, length: { maximum: 128 }
+
+    scope :scoped_build, -> { where('ci_builds.id=ci_build_needs.build_id') }
+  end
+end
diff --git a/app/models/ci/pipeline.rb b/app/models/ci/pipeline.rb
index 1c76f401690..3515f0b83ee 100644
--- a/app/models/ci/pipeline.rb
+++ b/app/models/ci/pipeline.rb
@@ -611,8 +611,8 @@ module Ci
     end
 
     # rubocop: disable CodeReuse/ServiceClass
-    def process!
-      Ci::ProcessPipelineService.new(project, user).execute(self)
+    def process!(trigger_build_name = nil)
+      Ci::ProcessPipelineService.new(project, user).execute(self, trigger_build_name)
     end
     # rubocop: enable CodeReuse/ServiceClass
 
diff --git a/app/models/commit_status.rb b/app/models/commit_status.rb
index be6f3e9c5b0..d7eb78db5b8 100644
--- a/app/models/commit_status.rb
+++ b/app/models/commit_status.rb
@@ -43,6 +43,12 @@ class CommitStatus < ApplicationRecord
   scope :after_stage, -> (index) { where('stage_idx > ?', index) }
   scope :processables, -> { where(type: %w[Ci::Build Ci::Bridge]) }
 
+  scope :with_needs, -> (names = nil) do
+    needs = Ci::BuildNeed.scoped_build.select(1)
+    needs = needs.where(name: names) if names
+    where('EXISTS (?)', needs).preload(:needs)
+  end
+
   # We use `CommitStatusEnums.failure_reasons` here so that EE can more easily
   # extend this `Hash` with new values.
   enum_with_nil failure_reason: ::CommitStatusEnums.failure_reasons
@@ -116,7 +122,7 @@ class CommitStatus < ApplicationRecord
       commit_status.run_after_commit do
         if pipeline_id
           if complete? || manual?
-            PipelineProcessWorker.perform_async(pipeline_id)
+            BuildProcessWorker.perform_async(id)
           else
             PipelineUpdateWorker.perform_async(pipeline_id)
           end
diff --git a/app/models/concerns/ci/metadatable.rb b/app/models/concerns/ci/metadatable.rb
index 9eed9492b37..304cc71e9dc 100644
--- a/app/models/concerns/ci/metadatable.rb
+++ b/app/models/concerns/ci/metadatable.rb
@@ -29,6 +29,7 @@ module Ci
     def degenerate!
       self.class.transaction do
         self.update!(options: nil, yaml_variables: nil)
+        self.needs.all.delete_all
         self.metadata&.destroy
       end
     end
diff --git a/app/services/ci/process_pipeline_service.rb b/app/services/ci/process_pipeline_service.rb
index 207cc5017d0..e46615bcf75 100644
--- a/app/services/ci/process_pipeline_service.rb
+++ b/app/services/ci/process_pipeline_service.rb
@@ -4,19 +4,23 @@ module Ci
   class ProcessPipelineService < BaseService
     attr_reader :pipeline
 
-    def execute(pipeline)
+    def execute(pipeline, trigger_build_name = nil)
       @pipeline = pipeline
 
       update_retried
 
-      new_builds =
+      success =
         stage_indexes_of_created_processables.flat_map do |index|
           process_stage(index)
-        end
+        end.any?
+
+      # we evaluate dependent needs,
+      # only when the another job has finished
+      success = process_builds_with_needs(trigger_build_name) || success
 
       @pipeline.update_status
 
-      new_builds.any?
+      success
     end
 
     private
@@ -36,6 +40,28 @@ module Ci
       end
     end
 
+    def process_builds_with_needs(trigger_build_name)
+      return false unless trigger_build_name
+      return false unless Feature.enabled?(:ci_dag_support, project)
+
+      created_processables
+        .with_needs(trigger_build_name)
+        .find_each
+        .map(&method(:process_build_with_needs))
+        .any?
+    end
+
+    def process_build_with_needs(build)
+      current_status = status_for_build_needs(build.needs.map(&:name))
+
+      return unless HasStatus::COMPLETED_STATUSES.include?(current_status)
+
+      Gitlab::OptimisticLocking.retry_lock(build) do |subject|
+        Ci::ProcessBuildService.new(project, @user)
+          .execute(subject, current_status)
+      end
+    end
+
     # rubocop: disable CodeReuse/ActiveRecord
     def status_for_prior_stages(index)
       pipeline.builds.where('stage_idx < ?', index).latest.status || 'success'
@@ -43,6 +69,12 @@ module Ci
     # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
+    def status_for_build_needs(needs)
+      pipeline.builds.where(name: needs).latest.status || 'success'
+    end
+    # rubocop: enable CodeReuse/ActiveRecord
+
+    # rubocop: disable CodeReuse/ActiveRecord
     def stage_indexes_of_created_processables
       created_processables.order(:stage_idx).pluck(Arel.sql('DISTINCT stage_idx'))
     end
diff --git a/app/services/ci/retry_build_service.rb b/app/services/ci/retry_build_service.rb
index fab8a179843..338495ba030 100644
--- a/app/services/ci/retry_build_service.rb
+++ b/app/services/ci/retry_build_service.rb
@@ -5,7 +5,7 @@ module Ci
     CLONE_ACCESSORS = %i[pipeline project ref tag options name
                          allow_failure stage stage_id stage_idx trigger_request
                          yaml_variables when environment coverage_regex
-                         description tag_list protected].freeze
+                         description tag_list protected needs].freeze
 
     def execute(build)
       reprocess!(build).tap do |new_build|
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 991a177018e..400becdd023 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -88,6 +88,7 @@
 - pipeline_processing:ci_build_prepare
 - pipeline_processing:build_queue
 - pipeline_processing:build_success
+- pipeline_processing:build_process
 - pipeline_processing:pipeline_process
 - pipeline_processing:pipeline_success
 - pipeline_processing:pipeline_update
diff --git a/app/workers/build_process_worker.rb b/app/workers/build_process_worker.rb
new file mode 100644
index 00000000000..19e590ee1d7
--- /dev/null
+++ b/app/workers/build_process_worker.rb
@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+class BuildProcessWorker
+  include ApplicationWorker
+  include PipelineQueue
+
+  queue_namespace :pipeline_processing
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform(build_id)
+    CommitStatus.find_by(id: build_id).try do |build|
+      build.pipeline.process!(build.name)
+    end
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+end