Add cronjob and schedule check workers

Add specs and config

config/initializers/1_settings.rb

@@ -645,6 +645,9 @@ Gitlab.ee do
   Settings.cron_jobs['incident_management_persist_oncall_rotation_worker'] ||= Settingslogic.new({})
   Settings.cron_jobs['incident_management_persist_oncall_rotation_worker']['cron'] ||= '*/5 * * * *'
   Settings.cron_jobs['incident_management_persist_oncall_rotation_worker']['job_class'] = 'IncidentManagement::OncallRotations::PersistAllRotationsShiftsJob'
+  Settings.cron_jobs['incident_management_schedule_escalation_check_worker'] ||= Settingslogic.new({})
+  Settings.cron_jobs['incident_management_schedule_escalation_check_worker']['cron'] ||= '*/1 * * * *'
+  Settings.cron_jobs['incident_management_schedule_escalation_check_worker']['job_class'] = 'IncidentManagement::Escalations::ScheduleEscalationCheckCronWorker'
   Settings.cron_jobs['import_software_licenses_worker'] ||= Settingslogic.new({})
   Settings.cron_jobs['import_software_licenses_worker']['cron'] ||= '0 3 * * 0'
   Settings.cron_jobs['import_software_licenses_worker']['job_class'] = 'ImportSoftwareLicensesWorker'

config/sidekiq_queues.yml

@@ -180,6 +180,8 @@
   - 2
 - - incident_management_apply_incident_sla_exceeded_label
   - 1
+- - incident_management_escalations_pending_alert_escalation_check
+  - 1
 - - incident_management_oncall_rotations_persist_shifts_job
   - 1
 - - invalid_gpg_signature_update

ee/app/models/incident_management/pending_escalations/alert.rb

@@ -4,11 +4,15 @@ module IncidentManagement
   module PendingEscalations
     class Alert < ApplicationRecord
       include PartitionedTable
+      include EachBatch
+
       alias_attribute :target, :alert
 
       self.primary_key = :id
       self.table_name = 'incident_management_pending_alert_escalations'
 
+      ESCALATION_BUFFER = 1.month.freeze
+
       partitioned_by :process_at, strategy: :monthly
 
       belongs_to :oncall_schedule, class_name: 'OncallSchedule', foreign_key: 'schedule_id'

ee/app/services/incident_management/pending_escalations/create_service.rb

@@ -28,11 +28,13 @@ module IncidentManagement
       end
 
       def create_escalations(rules)
-        rules.each do |rule|
+        escalation_ids = rules.map do |rule|
           escalaton = create_escalation(rule)
-          process_escalation(escalaton) if rule.elapsed_time_seconds == 0
+          escalaton.id
         end
 
+        process_escalations(escalation_ids)
+
       rescue StandardError => e
         Gitlab::ErrorTracking.track_exception(e, target_type: target.class.to_s, target_id: target.id)
       end
@@ -47,8 +49,10 @@ module IncidentManagement
         )
       end
 
-      def process_escalation(escalation)
-        ::IncidentManagement::PendingEscalations::ProcessService.new(escalation).execute
+      def process_escalations(escalation_ids)
+        args = escalation_ids.map { |id| [id] }
+
+        ::IncidentManagement::Escalations::PendingAlertEscalationCheckWorker.bulk_perform_async(args) # rubocop:disable Scalability/BulkPerformWithContext
       end
     end
   end

ee/app/workers/all_queues.yml

@@ -294,6 +294,15 @@
   :weight: 1
   :idempotent:
   :tags: []
+- :name: cronjob:incident_management_escalations_schedule_escalation_check_cron
+  :worker_name: IncidentManagement::Escalations::ScheduleEscalationCheckCronWorker
+  :feature_category: :incident_management
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: cronjob:incident_management_incident_sla_exceeded_check
   :worker_name: IncidentManagement::IncidentSlaExceededCheckWorker
   :feature_category: :incident_management
@@ -1044,6 +1053,15 @@
   :idempotent: true
   :tags:
   - :exclude_from_kubernetes
+- :name: incident_management_escalations_pending_alert_escalation_check
+  :worker_name: IncidentManagement::Escalations::PendingAlertEscalationCheckWorker
+  :feature_category: :incident_management
+  :has_external_dependencies:
+  :urgency: :high
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: incident_management_oncall_rotations_persist_shifts_job
   :worker_name: IncidentManagement::OncallRotations::PersistShiftsJob
   :feature_category: :incident_management

ee/app/workers/incident_management/escalations/pending_alert_escalation_check_worker.rb

+# frozen_string_literal: true
+
+module IncidentManagement
+  module Escalations
+    class PendingAlertEscalationCheckWorker
+      include ApplicationWorker
+
+      urgency :high
+
+      idempotent!
+      feature_category :incident_management
+
+      def initialize(escalation_id)
+        @escalation_id = escalation_id
+      end
+
+      def perform
+        escalation = IncidentManagement::PendingEscalations::Alert.find(escalation_id)
+
+        IncidentManagement::PendingEscalations::ProcessService.new(escalation).execute
+      end
+
+      private
+
+      attr_reader :escalation_id
+    end
+  end
+end

ee/app/workers/incident_management/escalations/schedule_escalation_check_cron_worker.rb

+# frozen_string_literal: true
+
+module IncidentManagement
+  module Escalations
+    class ScheduleEscalationCheckCronWorker
+      include ApplicationWorker
+      # This worker does not perform work scoped to a context
+      include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+      idempotent!
+      feature_category :incident_management
+
+      def perform
+        ::IncidentManagement::PendingEscalations::Alert.processable.each_batch do |relation|
+          args = relation.pluck(:id).map { |id| [id] } # rubocop:disable  CodeReuse/ActiveRecord
+          ::IncidentManagement::Escalations::PendingAlertEscalationCheckWorker.bulk_perform_async(args)  # rubocop:disable Scalability/BulkPerformWithContext
+        end
+      end
+    end
+  end
+end

ee/spec/services/incident_management/pending_escalations/create_service_spec.rb

@@ -53,13 +53,11 @@ RSpec.describe IncidentManagement::PendingEscalations::CreateService do
       end
     end
 
-    it 'processes the escalation' do
-      expect(IncidentManagement::PendingEscalations::ProcessService)
-        .to receive(:new)
-        .with(having_attributes(rule_id: first_escalation_rule.id))
-        .and_return(process_service_spy)
+    it 'creates the escalations and queues the escalation process check' do
+      expect(IncidentManagement::Escalations::PendingAlertEscalationCheckWorker)
+        .to receive(:bulk_perform_async)
+        .with([[a_kind_of(Integer)], [a_kind_of(Integer)]])
 
-      expect(process_service_spy).to receive(:execute)
       expect { execute }.to change { IncidentManagement::PendingEscalations::Alert.count }.by(rule_count)
     end
 

ee/spec/support/shared_examples/services/alert_management/alert_processing/escalation_policies_shared_examples.rb

@@ -8,6 +8,13 @@ RSpec.shared_examples 'creates an escalation' do |count|
       .exactly(count).times
       .and_call_original
 
+    expected_args = []
+    count.times { expected_args << [a_kind_of(Integer)]}
+
+    expect(IncidentManagement::Escalations::PendingAlertEscalationCheckWorker)
+      .to receive(:bulk_perform_async)
+      .with(expected_args)
+
     subject
   end
 end

ee/spec/workers/incident_management/escalations/pending_alert_escalation_check_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe IncidentManagement::Escalations::PendingAlertEscalationCheckWorker do
+  let(:worker) { described_class.new(escalation.id) }
+
+  let_it_be(:escalation) { create(:incident_management_pending_alert_escalation) }
+
+  describe '#perform' do
+    subject { worker.perform }
+
+    it 'processes the escalation' do
+      process_service = spy(IncidentManagement::PendingEscalations::ProcessService)
+
+      expect(IncidentManagement::PendingEscalations::ProcessService).to receive(:new).with(escalation).and_return(process_service)
+      subject
+      expect(process_service).to have_received(:execute)
+    end
+  end
+end

ee/spec/workers/incident_management/escalations/schedule_escalation_check_cron_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe IncidentManagement::Escalations::ScheduleEscalationCheckCronWorker do
+  let(:worker) { described_class.new }
+
+  let_it_be(:escalation_1) { create(:incident_management_pending_alert_escalation, process_at: 5.minutes.ago) }
+  let_it_be(:escalation_2) { create(:incident_management_pending_alert_escalation, process_at: 2.days.ago) }
+  let_it_be(:escalation_not_ready_to_process) { create(:incident_management_pending_alert_escalation) }
+
+  describe '#perform' do
+    subject { worker.perform }
+
+    it 'schedules a job for each processable escalation' do
+      expect(IncidentManagement::Escalations::PendingAlertEscalationCheckWorker).to receive(:bulk_perform_async)
+        .with(array_including([escalation_2.id], [escalation_1.id]))
+
+      subject
+    end
+  end
+end