Merge branch 'mc_rocha-add-work-create-pipeline-sep' into 'master'

Add a new worker to process the pipeline creation for sep

See merge request https://gitlab.com/gitlab-org/gitlab/-/merge_requests/147691



Merged-by: Sashi Kumar Kumaresan <skumar@gitlab.com>
Approved-by: Dominic Bauer <dbauer@gitlab.com>
Approved-by: Sashi Kumar Kumaresan <skumar@gitlab.com>
Reviewed-by: Sashi Kumar Kumaresan <skumar@gitlab.com>
Co-authored-by: mc_rocha <mrocha@gitlab.com>

config/sidekiq_queues.yml

@@ -717,6 +717,8 @@
   - 1
 - - security_refresh_project_policies
   - 1
+- - security_scan_execution_policies_create_pipeline
+  - 1
 - - security_scan_execution_policies_rule_schedule
   - 1
 - - security_scan_result_policies_add_approvers_to_rules

ee/app/services/security/security_orchestration_policies/rule_schedule_service.rb

@@ -14,7 +14,15 @@ def execute(schedule)
 
         branches = branches_for(rule)
         actions = actions_for(schedule)
-        schedule_errors = schedule_scan(actions, branches).select { |service_result| service_result[:status] == :error }
+
+        if Feature.enabled?(:scan_execution_pipeline_worker, project)
+          schedule_scans_using_a_worker(branches, schedule) unless actions.blank?
+          schedule_errors = []
+        else
+          schedule_errors = schedule_scan(actions, branches).select do |service_result|
+            service_result[:status] == :error
+          end
+        end
 
         return ServiceResponse.success if schedule_errors.blank?
 
@@ -62,6 +70,15 @@ def schedule_scan(actions, branches)
             .execute
         end
       end
+
+      def schedule_scans_using_a_worker(branches, schedule)
+        branches.map do |branch|
+          ::Security::ScanExecutionPolicies::CreatePipelineWorker.perform_async(project.id,
+            current_user.id,
+            schedule.id,
+            branch)
+        end
+      end
     end
   end
 end

ee/app/workers/all_queues.yml

@@ -2082,6 +2082,15 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: security_scan_execution_policies_create_pipeline
+  :worker_name: Security::ScanExecutionPolicies::CreatePipelineWorker
+  :feature_category: :security_policy_management
+  :has_external_dependencies: false
+  :urgency: :throttled
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: false
+  :tags: []
 - :name: security_scan_execution_policies_rule_schedule
   :worker_name: Security::ScanExecutionPolicies::RuleScheduleWorker
   :feature_category: :security_policy_management

ee/app/workers/security/scan_execution_policies/create_pipeline_worker.rb

+# frozen_string_literal: true
+
+module Security
+  module ScanExecutionPolicies
+    class CreatePipelineWorker # rubocop:disable Scalability/IdempotentWorker -- The worker should not run multiple times to avoid creating multiple pipelines
+      include ApplicationWorker
+
+      feature_category :security_policy_management
+      deduplicate :until_executing
+      urgency :throttled
+      data_consistency :delayed
+
+      concurrency_limit -> { 50 if Feature.enabled?(:scan_execution_pipeline_worker) }
+
+      def perform(project_id, current_user_id, schedule_id, branch)
+        project = Project.find_by_id(project_id)
+        return unless project
+
+        current_user = User.find_by_id(current_user_id)
+        return unless current_user
+
+        schedule = Security::OrchestrationPolicyRuleSchedule.find_by_id(schedule_id)
+        return unless schedule
+
+        actions = actions_for(schedule)
+
+        service_result = ::Security::SecurityOrchestrationPolicies::CreatePipelineService
+          .new(project: project, current_user: current_user, params: { actions: actions, branch: branch })
+          .execute
+
+        return unless service_result[:status] == :error
+
+        log_error(current_user, schedule, service_result[:message])
+      end
+
+      private
+
+      def actions_for(schedule)
+        policy = schedule.policy
+        return [] if policy.blank?
+
+        policy[:actions]
+      end
+
+      def log_error(current_user, schedule, message)
+        ::Gitlab::AppJsonLogger.warn(
+          build_structured_payload(
+            security_orchestration_policy_configuration_id: schedule&.security_orchestration_policy_configuration&.id,
+            user_id: current_user.id,
+            message: message
+          )
+        )
+      end
+    end
+  end
+end

ee/config/feature_flags/gitlab_com_derisk/scan_execution_pipeline_worker.yml

+---
+name: scan_execution_pipeline_worker
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/147691
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/451890
+milestone: '16.11'
+type: gitlab_com_derisk
+group: group::security policies
+default_enabled: false

ee/spec/services/security/security_orchestration_policies/rule_schedule_service_spec.rb

@@ -15,6 +15,7 @@
     let(:rule) { { type: 'schedule', branches: branches, cadence: '*/20 * * * *' } }
     let(:other_schedule_rule) { { type: 'schedule', branches: ['main'], cadence: '0 10 * * *' } }
     let(:branches) { %w[master production non-existing-branch] }
+    let(:existing_branches) { %w[master production] }
 
     subject(:service) { described_class.new(project: project, current_user: current_user) }
 
@@ -25,6 +26,7 @@
     end
 
     before do
+      stub_feature_flags(scan_execution_pipeline_worker: false)
       stub_licensed_features(security_on_demand_scans: true)
 
       project.repository.create_branch('production', project.default_branch)
@@ -41,6 +43,39 @@
       expect(service_result.success?).to be(true)
     end
 
+    shared_examples 'with scan' do |scan_type|
+      context "when scan is #{scan_type}" do
+        context 'when the feature flag scan_execution_pipeline_worker is enabled' do
+          before do
+            stub_feature_flags(scan_execution_pipeline_worker: true)
+          end
+
+          it 'enqueues Security::SyncScanPoliciesWorker for each branch' do
+            existing_branches.each do |branch|
+              expect(::Security::ScanExecutionPolicies::CreatePipelineWorker).to(
+                receive(:perform_async)
+                  .with(project.id, current_user.id, schedule.id, branch)
+                  .and_call_original
+              )
+            end
+
+            service.execute(schedule)
+          end
+
+          it 'does not invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+            existing_branches.each do |branch|
+              expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).not_to(
+                receive(:new)
+                  .with(project: project, current_user: current_user,
+                    params: { actions: [{ scan: 'scan_type' }], branch: branch }))
+            end
+
+            service.execute(schedule)
+          end
+        end
+      end
+    end
+
     context 'when scan type is dast' do
       before do
         policy[:actions] = [{ scan: 'dast' }]
@@ -59,6 +94,8 @@
 
         service.execute(schedule)
       end
+
+      it_behaves_like 'with scan', 'dast'
     end
 
     context 'when scan type is secret_detection' do
@@ -79,6 +116,8 @@
 
         service.execute(schedule)
       end
+
+      it_behaves_like 'with scan', 'secret_detection'
     end
 
     context 'when scan type is container_scanning' do
@@ -100,6 +139,8 @@
 
           service.execute(schedule)
         end
+
+        it_behaves_like 'with scan', 'container_scanning'
       end
 
       context 'when agents are defined in the rule' do
@@ -131,6 +172,8 @@
 
         service.execute(schedule)
       end
+
+      it_behaves_like 'with scan', 'sast'
     end
 
     context 'when policy actions exists and there are multiple matching branches' do

ee/spec/workers/security/scan_execution_policies/create_pipeline_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::ScanExecutionPolicies::CreatePipelineWorker, feature_category: :security_policy_management do
+  let_it_be(:project) { create(:project) }
+  let_it_be(:current_user) { create(:user) }
+  let_it_be(:security_orchestration_policy_configuration) do
+    create(:security_orchestration_policy_configuration, project: project)
+  end
+
+  let_it_be(:schedule) do
+    create(:security_orchestration_policy_rule_schedule,
+      security_orchestration_policy_configuration: security_orchestration_policy_configuration)
+  end
+
+  let(:project_id) { project.id }
+  let(:current_user_id) { current_user.id }
+  let(:branch) { 'production' }
+  let(:actions) { [{ scan: 'dast' }] }
+  let(:params) { { actions: actions, branch: branch } }
+  let(:schedule_id) { schedule.id }
+  let(:policy) { build(:scan_execution_policy, enabled: true, actions: [{ scan: 'dast' }]) }
+
+  shared_examples_for 'does not call RuleScheduleService' do
+    it do
+      expect(Security::SecurityOrchestrationPolicies::RuleScheduleService).not_to receive(:new)
+
+      run_worker
+    end
+  end
+
+  describe '#perform' do
+    before do
+      allow_next_found_instance_of(Security::OrchestrationPolicyConfiguration) do |instance|
+        allow(instance).to receive(:active_scan_execution_policies).and_return([policy])
+      end
+    end
+
+    subject(:run_worker) { described_class.new.perform(project_id, current_user_id, schedule_id, branch) }
+
+    context 'when project is not found' do
+      let(:project_id) { non_existing_record_id }
+
+      it_behaves_like 'does not call RuleScheduleService'
+    end
+
+    context 'when user is not found' do
+      let(:current_user_id) { non_existing_record_id }
+
+      it_behaves_like 'does not call RuleScheduleService'
+    end
+
+    context 'when the user and project exists' do
+      it 'delegates the pipeline creation to Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+        expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+          receive(:new)
+            .with(project: project, current_user: current_user, params: params)
+            .and_call_original)
+
+        run_worker
+      end
+
+      context 'when create pipeline service returns errors' do
+        before do
+          allow_next_instance_of(::Security::SecurityOrchestrationPolicies::CreatePipelineService) do |service|
+            allow(service).to receive(:execute).and_return(ServiceResponse.error(message: 'message'))
+          end
+        end
+
+        it 'logs the error' do
+          expect(::Gitlab::AppJsonLogger).to receive(:warn).with({
+            'class' => 'Security::ScanExecutionPolicies::CreatePipelineWorker',
+            'security_orchestration_policy_configuration_id' => security_orchestration_policy_configuration.id,
+            'user_id' => current_user.id,
+            'message' => 'message'
+          })
+          run_worker
+        end
+      end
+    end
+  end
+end