Merge branch 'sy-rule-based-escalations' into 'master'

Route alert notifications only through escalation rules

See merge request gitlab-org/gitlab!65635

db/migrate/20210721174411_add_is_removed_to_escalation_rules.rb

+# frozen_string_literal: true
+
+class AddIsRemovedToEscalationRules < ActiveRecord::Migration[6.1]
+  def change
+    add_column :incident_management_escalation_rules, :is_removed, :boolean, null: false, default: false
+  end
+end

db/migrate/20210721174441_update_escalation_rule_fk_for_pending_alert_escalations.rb

+# frozen_string_literal: true
+
+class UpdateEscalationRuleFkForPendingAlertEscalations < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+  include Gitlab::Database::PartitioningMigrationHelpers
+
+  disable_ddl_transaction!
+
+  OLD_FOREIGN_KEY_CONSTRAINT = 'fk_rails_057c1e3d87'
+
+  # Swap foreign key contrainst from ON DELETE SET NULL to ON DELETE CASCADE
+  def up
+    remove_foreign_key_if_exists :incident_management_pending_alert_escalations, :incident_management_escalation_rules, name: OLD_FOREIGN_KEY_CONSTRAINT
+
+    add_concurrent_partitioned_foreign_key :incident_management_pending_alert_escalations,
+      :incident_management_escalation_rules,
+      column: :rule_id
+  end
+
+  def down
+    remove_foreign_key_if_exists :incident_management_pending_alert_escalations, :incident_management_escalation_rules, column: :rule_id
+
+    add_concurrent_partitioned_foreign_key :incident_management_pending_alert_escalations,
+      :incident_management_escalation_rules,
+      column: :rule_id,
+      on_delete: :nullify,
+      name: OLD_FOREIGN_KEY_CONSTRAINT
+  end
+end

db/migrate/20210721174453_remove_schedule_and_status_null_constraints_from_pending_escalations_alert.rb

+# frozen_string_literal: true
+
+class RemoveScheduleAndStatusNullConstraintsFromPendingEscalationsAlert < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  # In preparation of removal of these columns in 14.3.
+  def up
+    with_lock_retries do
+      change_column_null :incident_management_pending_alert_escalations, :status, true
+      change_column_null :incident_management_pending_alert_escalations, :schedule_id, true
+    end
+  end
+
+  def down
+    backfill_from_rules_and_disallow_column_null :status, value: :status
+    backfill_from_rules_and_disallow_column_null :schedule_id, value: :oncall_schedule_id
+  end
+
+  private
+
+  def backfill_from_rules_and_disallow_column_null(column, value:)
+    with_lock_retries do
+      execute <<~SQL
+        UPDATE incident_management_pending_alert_escalations AS escalations
+        SET #{column} = rules.#{value}
+        FROM incident_management_escalation_rules AS rules
+        WHERE rule_id = rules.id
+        AND escalations.#{column} IS NULL
+      SQL
+
+      change_column_null :incident_management_pending_alert_escalations, column, false
+    end
+  end
+end

db/post_migrate/20210721174521_add_non_null_constraint_for_escalation_rule_on_pending_alert_escalations.rb

+# frozen_string_literal: true
+
+class AddNonNullConstraintForEscalationRuleOnPendingAlertEscalations < ActiveRecord::Migration[6.1]
+  ELAPSED_WHOLE_MINUTES_IN_SECONDS = <<~SQL
+    ABS(ROUND(
+      EXTRACT(EPOCH FROM (escalations.process_at - escalations.created_at))/60*60
+    ))
+  SQL
+
+  INSERT_RULES_FOR_ESCALATIONS_WITHOUT_RULES = <<~SQL
+    INSERT INTO incident_management_escalation_rules (policy_id, oncall_schedule_id, status, elapsed_time_seconds, is_removed)
+    SELECT
+      policies.id,
+      schedule_id,
+      status,
+      #{ELAPSED_WHOLE_MINUTES_IN_SECONDS} AS elapsed_time_seconds,
+      TRUE
+    FROM incident_management_pending_alert_escalations AS escalations
+    INNER JOIN incident_management_oncall_schedules AS schedules ON schedules.id = schedule_id
+    INNER JOIN incident_management_escalation_policies AS policies ON policies.project_id = schedules.project_id
+    WHERE rule_id IS NULL
+    GROUP BY policies.id, schedule_id, status, elapsed_time_seconds
+    ON CONFLICT DO NOTHING;
+  SQL
+
+  UPDATE_EMPTY_RULE_IDS = <<~SQL
+    UPDATE incident_management_pending_alert_escalations AS escalations
+    SET rule_id = rules.id
+    FROM incident_management_pending_alert_escalations AS through_escalations
+    INNER JOIN incident_management_oncall_schedules AS schedules ON schedules.id = through_escalations.schedule_id
+    INNER JOIN incident_management_escalation_policies AS policies ON policies.project_id = schedules.project_id
+    INNER JOIN incident_management_escalation_rules AS rules ON rules.policy_id = policies.id
+    WHERE escalations.rule_id IS NULL
+    AND rules.status = escalations.status
+    AND rules.oncall_schedule_id = escalations.schedule_id
+    AND rules.elapsed_time_seconds = #{ELAPSED_WHOLE_MINUTES_IN_SECONDS};
+  SQL
+
+  DELETE_LEFTOVER_ESCALATIONS_WITHOUT_RULES = 'DELETE FROM incident_management_pending_alert_escalations WHERE rule_id IS NULL;'
+
+  # For each alert which has a pending escalation without a corresponding rule,
+  # create a rule with the expected attributes for the project's policy.
+  #
+  # Deletes all escalations without rules/policy & adds non-null constraint for rule_id.
+  def up
+    exec_query INSERT_RULES_FOR_ESCALATIONS_WITHOUT_RULES
+    exec_query UPDATE_EMPTY_RULE_IDS
+    exec_query DELETE_LEFTOVER_ESCALATIONS_WITHOUT_RULES
+
+    change_column_null :incident_management_pending_alert_escalations, :rule_id, false
+  end
+
+  def down
+    change_column_null :incident_management_pending_alert_escalations, :rule_id, true
+  end
+end

db/schema_migrations/20210721174411

+ac95292b2ab05f17ed13cb8e95ace0660e6dc82e33d6ef1cccd02890abf6c739
\ No newline at end of file

db/schema_migrations/20210721174441

+9f3a39b11f250f64e4e6b8623279604c1dba14330f45c26840f6e0b46f7d48a7
\ No newline at end of file

db/schema_migrations/20210721174453

+7b20c623b58982ba5d228902c6da6d10245edf3874ece9b02d58e8560d2d5d96
\ No newline at end of file

db/schema_migrations/20210721174521

+f16b563bbfa15b97143e82d2a1e78e9d9805d13e02e3a0845369d4ce3204b3cc
\ No newline at end of file

db/structure.sql

@@ -257,13 +257,13 @@ PARTITION BY RANGE (created_at);
 
 CREATE TABLE incident_management_pending_alert_escalations (
     id bigint NOT NULL,
-    rule_id bigint,
+    rule_id bigint NOT NULL,
     alert_id bigint NOT NULL,
-    schedule_id bigint NOT NULL,
+    schedule_id bigint,
     process_at timestamp with time zone NOT NULL,
     created_at timestamp with time zone NOT NULL,
     updated_at timestamp with time zone NOT NULL,
-    status smallint NOT NULL
+    status smallint
 )
 PARTITION BY RANGE (process_at);
 
@@ -13975,7 +13975,8 @@ CREATE TABLE incident_management_escalation_rules (
     policy_id bigint NOT NULL,
     oncall_schedule_id bigint NOT NULL,
     status smallint NOT NULL,
-    elapsed_time_seconds integer NOT NULL
+    elapsed_time_seconds integer NOT NULL,
+    is_removed boolean DEFAULT false NOT NULL
 );
 
 CREATE SEQUENCE incident_management_escalation_rules_id_seq
@@ -26731,9 +26732,6 @@ ALTER TABLE ONLY terraform_state_versions
 ALTER TABLE ONLY ci_build_report_results
     ADD CONSTRAINT fk_rails_056d298d48 FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE;
 
-ALTER TABLE incident_management_pending_alert_escalations
-    ADD CONSTRAINT fk_rails_057c1e3d87 FOREIGN KEY (rule_id) REFERENCES incident_management_escalation_rules(id) ON DELETE SET NULL;
-
 ALTER TABLE ONLY ci_daily_build_group_report_results
     ADD CONSTRAINT fk_rails_0667f7608c FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE;
 
@@ -28210,6 +28208,9 @@ ALTER TABLE ONLY approval_project_rules_users
 ALTER TABLE ONLY insights
     ADD CONSTRAINT fk_rails_f36fda3932 FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE;
 
+ALTER TABLE incident_management_pending_alert_escalations
+    ADD CONSTRAINT fk_rails_f3d17bc8af FOREIGN KEY (rule_id) REFERENCES incident_management_escalation_rules(id) ON DELETE CASCADE;
+
 ALTER TABLE ONLY board_group_recent_visits
     ADD CONSTRAINT fk_rails_f410736518 FOREIGN KEY (group_id) REFERENCES namespaces(id) ON DELETE CASCADE;
 

ee/app/graphql/resolvers/incident_management/escalation_policies_resolver.rb

@@ -25,7 +25,7 @@ module Resolvers
 
       def preloads
         {
-          rules: [:ordered_rules]
+          rules: [:active_rules]
         }
       end
     end

ee/app/graphql/types/incident_management/escalation_policy_type.rb

@@ -23,7 +23,7 @@ module Types
       field :rules, [Types::IncidentManagement::EscalationRuleType],
             null: true,
             description: 'Steps of the escalation policy.',
-            method: :ordered_rules
+            method: :active_rules
     end
   end
 end

ee/app/models/incident_management/escalation_policy.rb

@@ -6,7 +6,7 @@ module IncidentManagement
 
     belongs_to :project
     has_many :rules, class_name: 'EscalationRule', inverse_of: :policy, foreign_key: 'policy_id', index_errors: true
-    has_many :ordered_rules, -> { order(:elapsed_time_seconds, :status) }, class_name: 'EscalationRule', inverse_of: :policy, foreign_key: 'policy_id'
+    has_many :active_rules, -> { not_removed.order(:elapsed_time_seconds, :status) }, class_name: 'EscalationRule', inverse_of: :policy, foreign_key: 'policy_id'
 
     validates :project_id, uniqueness: { message: _('can only have one escalation policy') }, on: :create
     validates :name, presence: true, uniqueness: { scope: [:project_id] }, length: { maximum: 72 }
@@ -14,7 +14,5 @@ module IncidentManagement
     validates :rules, presence: true
 
     accepts_nested_attributes_for :rules
-
-    scope :with_rules, -> { includes(:rules) }
   end
 end

ee/app/models/incident_management/escalation_rule.rb

@@ -16,5 +16,8 @@ module IncidentManagement
               numericality: { only_integer: true, greater_than_or_equal_to: 0, less_than_or_equal_to: 24.hours }
 
     validates :policy_id, uniqueness: { scope: [:oncall_schedule_id, :status, :elapsed_time_seconds], message: _('must have a unique schedule, status, and elapsed time') }
+
+    scope :not_removed, -> { where(is_removed: false) }
+    scope :removed, -> { where(is_removed: true) }
   end
 end

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

@@ -5,6 +5,9 @@ module IncidentManagement
     class Alert < ApplicationRecord
       include PartitionedTable
       include EachBatch
+      include IgnorableColumns
+
+      ignore_columns :schedule_id, :status, remove_with: '14.4', remove_after: '2021-09-22'
 
       alias_attribute :target, :alert
 
@@ -15,20 +18,15 @@ module IncidentManagement
 
       partitioned_by :process_at, strategy: :monthly
 
-      belongs_to :oncall_schedule, class_name: 'OncallSchedule', foreign_key: 'schedule_id'
       belongs_to :alert, class_name: 'AlertManagement::Alert', foreign_key: 'alert_id', inverse_of: :pending_escalations
-      belongs_to :rule, class_name: 'EscalationRule', foreign_key: 'rule_id', optional: true
+      belongs_to :rule, class_name: 'EscalationRule', foreign_key: 'rule_id'
 
       scope :processable, -> { where(process_at: ESCALATION_BUFFER.ago..Time.current) }
 
-      enum status: AlertManagement::Alert::STATUSES.slice(:acknowledged, :resolved)
-
       validates :process_at, presence: true
-      validates :status, presence: true
       validates :rule_id, presence: true, uniqueness: { scope: [:alert_id] }
 
       delegate :project, to: :alert
-      delegate :policy, to: :rule, allow_nil: true
     end
   end
 end

ee/app/services/ee/system_note_service.rb

@@ -112,8 +112,8 @@ module EE
       issuables_service(noteable, project, author).publish_issue_to_status_page
     end
 
-    def notify_via_escalation(noteable, project, recipients, escalation_policy, oncall_schedule)
-      escalations_service(noteable, project).notify_via_escalation(recipients, escalation_policy: escalation_policy, oncall_schedule: oncall_schedule)
+    def notify_via_escalation(noteable, project, recipients, escalation_policy)
+      escalations_service(noteable, project).notify_via_escalation(recipients, escalation_policy: escalation_policy)
     end
 
     private

ee/app/services/incident_management/escalation_policies/update_service.rb

@@ -46,36 +46,45 @@ module IncidentManagement
         params[:rules_attributes] && params[:rules_attributes].empty?
       end
 
-      # Limits rules_attributes to only new records & prepares
-      # to delete existing rules which are no longer needed
-      # when the policy is saved.
-      #
-      # Context: Rules are managed via `accepts_nested_attributes_for`
-      # on the IncidentManagement::EscalationPolicy.
-      # `accepts_nested_attributes_for` requires explicit
-      # removal of records, so we'll limit `rules_attributes`
-      # to new records, then rely on `autosave` to actually
-      # destroy the unwanted rules after marking them for
-      # deletion.
+      # Replaces rules params with records for existing rules,
+      # creates records for new rules, and marks appropriate
+      # rule records for removal. Records are not actually
+      # deleted as there may be pending escalations for the rule.
       def reconcile_rules!
-        return unless rules_attributes = params.delete(:rules_attributes)
+        return unless expected_rules_by_uniq_id.present?
 
-        params[:rules_attributes] = remove_obsolete_rules(rules_attributes).to_a
+        update_existing_rules!
+
+        params[:rules] = expected_rules_by_uniq_id.merge(existing_rules_by_uniq_id).values
+      end
+
+      # Prepares existing rules to be removed or un-removed
+      # based on whether they're included in the input params
+      def update_existing_rules!
+        existing_rules_by_uniq_id.each do |uniq_id, rule|
+          rule.is_removed = !expected_rules_by_uniq_id.key?(uniq_id)
+        end
       end
 
-      def remove_obsolete_rules(rules_attrs)
-        expected_rules = rules_attrs.to_set { |attrs| normalize(::IncidentManagement::EscalationRule.new(**attrs)) }
+      # @return [Hash<Array, IncidentManagement::EscalationRule>]
+      def existing_rules_by_uniq_id
+        strong_memoize(:existing_rules_by_uniq_id) do
+          escalation_policy.rules.index_by { |rule| unique_id(rule) }
+        end
+      end
 
-        escalation_policy.rules.each_with_object(expected_rules) do |existing_rule, new_rules|
-          # Exclude an expected rule which already corresponds to a persisted record - it's a no-op.
-          next if new_rules.delete?(normalize(existing_rule))
+      # @return [Hash<Array, IncidentManagement::EscalationRule>]
+      def expected_rules_by_uniq_id
+        strong_memoize(:expected_rules_by_uniq_id) do
+          params.delete(:rules_attributes).to_h do |attrs|
+            rule = ::IncidentManagement::EscalationRule.new(**attrs)
 
-          # Destroy a persisted record, since we don't expect this rule to be on the policy anymore.
-          existing_rule.mark_for_destruction
+            [unique_id(rule), rule]
+          end
         end
       end
 
-      def normalize(rule)
+      def unique_id(rule)
         rule.slice(:oncall_schedule_id, :elapsed_time_seconds, :status)
       end
     end

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

@@ -12,20 +12,16 @@ module IncidentManagement
       def execute
         return unless ::Gitlab::IncidentManagement.escalation_policies_available?(project) && !target.resolved?
 
-        policy = escalation_policies.first
+        policy = project.incident_management_escalation_policies.first
 
         return unless policy
 
-        create_escalations(policy.rules)
+        create_escalations(policy.active_rules)
       end
 
       private
 
-      attr_reader :target, :project, :escalation, :process_time
-
-      def escalation_policies
-        project.incident_management_escalation_policies.with_rules
-      end
+      attr_reader :target, :project, :process_time
 
       def create_escalations(rules)
         escalation_ids = rules.map do |rule|
@@ -40,8 +36,6 @@ module IncidentManagement
         IncidentManagement::PendingEscalations::Alert.create!(
           target: target,
           rule: rule,
-          schedule_id: rule.oncall_schedule_id,
-          status: rule.status,
           process_at: rule.elapsed_time_seconds.seconds.after(process_time)
         )
       end

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

@@ -8,7 +8,7 @@ module IncidentManagement
       def initialize(escalation)
         @escalation = escalation
         @project = escalation.project
-        @oncall_schedule = escalation.oncall_schedule
+        @rule = escalation.rule
         @target = escalation.target
       end
 
@@ -16,6 +16,7 @@ module IncidentManagement
         return unless ::Gitlab::IncidentManagement.escalation_policies_available?(project)
         return if too_early_to_process?
         return if target_already_resolved?
+        return unless rule # Remove in %14.3; Rule might be unavailable after deploy, but before post-migrations complete.
         return if target_status_exceeded_rule?
 
         notify_recipients
@@ -25,7 +26,7 @@ module IncidentManagement
 
       private
 
-      attr_reader :escalation, :project, :target, :oncall_schedule
+      attr_reader :escalation, :project, :target, :rule
 
       def target_already_resolved?
         return false unless target.resolved?
@@ -34,7 +35,7 @@ module IncidentManagement
       end
 
       def target_status_exceeded_rule?
-        target.status >= escalation.status_before_type_cast
+        target.status >= rule.status_before_type_cast
       end
 
       def too_early_to_process?
@@ -49,12 +50,12 @@ module IncidentManagement
       end
 
       def create_system_notes
-        SystemNoteService.notify_via_escalation(target, project, oncall_notification_recipients, escalation.policy, oncall_schedule)
+        SystemNoteService.notify_via_escalation(target, project, oncall_notification_recipients, rule.policy)
       end
 
       def oncall_notification_recipients
         strong_memoize(:oncall_notification_recipients) do
-          ::IncidentManagement::OncallUsersFinder.new(project, schedule: oncall_schedule).execute.to_a
+          ::IncidentManagement::OncallUsersFinder.new(project, schedule: rule.oncall_schedule).execute.to_a
         end
       end
 

ee/app/services/system_notes/escalations_service.rb

@@ -8,12 +8,8 @@ module SystemNotes
       @author = User.alert_bot
     end
 
-    def notify_via_escalation(recipients, escalation_policy: nil, oncall_schedule: nil)
-      body = if escalation_policy
-               "notified #{recipients.map(&:to_reference).to_sentence} of this alert via escalation policy **#{escalation_policy.name}**"
-             else
-               "notified #{recipients.map(&:to_reference).to_sentence} of this alert via schedule **#{oncall_schedule.name}**, per an escalation rule which no longer exists"
-             end
+    def notify_via_escalation(recipients, escalation_policy:)
+      body = "notified #{recipients.map(&:to_reference).to_sentence} of this alert via escalation policy **#{escalation_policy.name}**"
 
       create_note(NoteSummary.new(noteable, project, author, body, action: 'new_alert_added'))
     end

ee/spec/factories/incident_management/escalation_rules.rb

@@ -2,13 +2,18 @@
 
 FactoryBot.define do
   factory :incident_management_escalation_rule, class: 'IncidentManagement::EscalationRule' do
-    association :policy, factory: :incident_management_escalation_policy
+    policy { association :incident_management_escalation_policy }
     oncall_schedule { association :incident_management_oncall_schedule, project: policy.project }
     status { IncidentManagement::EscalationRule.statuses[:acknowledged] }
     elapsed_time_seconds { 5.minutes }
+    is_removed { false }
 
     trait :resolved do
       status { IncidentManagement::EscalationRule.statuses[:resolved] }
     end
+
+    trait :removed do
+      is_removed { true }
+    end
   end
 end

ee/spec/factories/incident_management/pending_alert_escalations.rb

@@ -3,14 +3,12 @@
 FactoryBot.define do
   factory :incident_management_pending_alert_escalation, class: 'IncidentManagement::PendingEscalations::Alert' do
     transient do
-      project { create(:project) } # rubocop:disable FactoryBot/InlineAssociation
-      policy { create(:incident_management_escalation_policy, project: project) } # rubocop:disable FactoryBot/InlineAssociation
+      project { association :project }
+      policy { association :incident_management_escalation_policy, project: project }
     end
 
     rule { association :incident_management_escalation_rule, policy: policy }
-    oncall_schedule { association :incident_management_oncall_schedule, project: project }
-    alert { association :alert_management_alert, project: project }
-    status { IncidentManagement::EscalationRule.statuses[:acknowledged] }
+    alert { association :alert_management_alert, project: rule.policy.project }
     process_at { 5.minutes.from_now }
   end
 end

ee/spec/graphql/mutations/incident_management/escalation_policy/update_spec.rb

@@ -47,7 +47,7 @@ RSpec.describe Mutations::IncidentManagement::EscalationPolicy::Update do
 
         expect(resolve[:escalation_policy]).to have_attributes(escalation_policy.reload.attributes)
         expect(escalation_policy).to have_attributes(args.slice(:name, :description))
-        expect(escalation_policy.rules).to match_array(expected_rules)
+        expect(escalation_policy.active_rules).to match_array(expected_rules)
       end
     end
 

ee/spec/migrations/add_non_null_constraint_for_escalation_rule_on_pending_alert_escalations_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe AddNonNullConstraintForEscalationRuleOnPendingAlertEscalations, :migration, schema: 20210721174453 do
+  let(:namespaces) { table(:namespaces) }
+  let(:projects) { table(:projects) }
+  let(:schedules) { table(:incident_management_oncall_schedules) }
+  let(:policies) { table(:incident_management_escalation_policies) }
+  let(:rules) { table(:incident_management_escalation_rules) }
+  let(:alerts) { table(:alert_management_alerts) }
+  let(:escalations) { partitioned_table(:incident_management_pending_alert_escalations, by: :process_at) }
+
+  let!(:namespace) { namespaces.create!(name: 'namespace', path: 'namespace') }
+  let!(:project) { projects.create!(name: 'project', path: 'project', namespace_id: namespace.id) }
+  let!(:schedule) { schedules.create!(name: 'Schedule', iid: 1, project_id: project.id) }
+  let!(:policy) { policies.create!(name: 'Policy', project_id: project.id) }
+  let!(:rule) { rules.create!(oncall_schedule_id: schedule.id, policy_id: policy.id, status: 2, elapsed_time_seconds: 5.minutes) }
+  let!(:alert) { alerts.create!(iid: 1, project_id: project.id, title: 'Alert 1', started_at: Time.current) }
+  let!(:other_alert) { alerts.create!(iid: 2, project_id: project.id, title: 'Alert 2', started_at: Time.current) }
+  let!(:other_schedule) { schedules.create!(name: 'Other Schedule', iid: 2, project_id: project.id) }
+
+  let!(:other_project) { projects.create!(name: 'project2', path: 'project2', namespace_id: namespace.id) }
+  let!(:other_project_policy) { policies.create!(name: 'Other Policy', project_id: other_project.id) }
+  let!(:other_project_schedule) { schedules.create!(name: 'Other Schedule', iid: 1, project_id: other_project.id) }
+  let!(:other_project_alert) { alerts.create!(iid: 1, project_id: other_project.id, title: 'Other Project Alert', started_at: Time.current) }
+
+  let!(:policyless_project) { projects.create!(name: 'project2', path: 'project2', namespace_id: namespace.id) }
+  let!(:policyless_project_schedule) { schedules.create!(name: 'Schedule for no policy', iid: 1, project_id: policyless_project.id) }
+  let!(:policyless_project_alert) { alerts.create!(iid: 1, project_id: policyless_project.id, title: 'Alert with no policy', started_at: Time.current) }
+
+  it 'creates rules for each escalation and removes escalations without policies', :aggregate_failures do
+    # Should all point to the same new rule
+    escalation = create_escalation
+    matching_escalation = create_escalation
+    other_alert_escalation = create_escalation(alert_id: other_alert.id)
+    escalations_with_same_new_rule = [escalation, matching_escalation, other_alert_escalation]
+
+    # Should each point to a different new rule
+    other_status_escalation = create_escalation(status: 2)
+    other_elapsed_time_escalation = create_escalation(process_at: 1.minute.from_now)
+    other_schedule_escalation = create_escalation(schedule_id: other_schedule.id)
+    other_project_escalation = create_escalation(schedule_id: other_project_schedule.id, alert_id: other_project_alert.id)
+    escalations_with_unique_new_rules = [other_status_escalation, other_elapsed_time_escalation, other_schedule_escalation, other_project_escalation]
+
+    # Should be deleted
+    policyless_escalation = create_escalation(schedule_id: policyless_project_schedule.id, alert_id: policyless_project_alert.id)
+
+    # Should all point to the existing rule
+    existing_rule_escalation = create_escalation(rule_id: rule.id)
+    existing_rule_without_schedule_escalation = create_escalation(rule_id: rule.id, status: nil, schedule_id: nil)
+    existing_rule_escalation_without_rule_id = create_escalation(status: rule.status, process_at: rule.elapsed_time_seconds.seconds.after(Time.current))
+    escalations_with_existing_rule = [existing_rule_escalation, existing_rule_without_schedule_escalation, existing_rule_escalation_without_rule_id]
+
+    # Assert all escalations were updated with a rule id or deleted
+    expect { migrate! }
+      .to change(rules, :count).by(5)
+      .and change(escalations, :count).by(-1)
+
+    expect(escalations.pluck(:rule_id)).to all( be_present )
+
+    escalations_with_same_new_rule.each(&:reload)
+    escalations_with_unique_new_rules.each(&:reload)
+    escalations_with_existing_rule.each(&:reload)
+
+    expect { policyless_escalation.reload }.to raise_error(ActiveRecord::RecordNotFound)
+
+    # Assert rules are associated with the correct escalations
+    expect(escalations_with_existing_rule.map(&:rule_id)).to all( eq(rule.id) )
+    expect(rules.where(is_removed: false)).to contain_exactly(rule)
+
+    expect(escalations_with_same_new_rule.map(&:rule_id).uniq.length).to eq(1)
+
+    rule_ids_for_escalations_with_unique_new_rules = escalations_with_unique_new_rules.map(&:rule_id)
+    expect(rule_ids_for_escalations_with_unique_new_rules.uniq.length).to eq(rule_ids_for_escalations_with_unique_new_rules.length)
+    expect(rule_ids_for_escalations_with_unique_new_rules).not_to include(escalation.rule_id, rule.id)
+
+    # Assert new rules have the correct attributes
+    escalations.where.not(rule_id: rule.id).each do |esc|
+      rule = rules.find(esc.rule_id)
+
+      expect(esc).to have_attributes(
+        status: rule.status,
+        schedule_id: rule.oncall_schedule_id,
+        process_at: be_within(5.seconds).of(rule.elapsed_time_seconds.seconds.after(esc.created_at))
+      )
+    end
+
+    # Assert all rules are associated with the correct projects
+    project_rule_ids = rules.where(policy_id: policy.id).ids
+    expected_rule_ids_for_project = escalations.where.not(id: other_project_escalation.id).pluck(:rule_id)
+    expect(project_rule_ids - expected_rule_ids_for_project).to be_empty
+
+    other_project_rule_ids = rules.where(policy_id: other_project_policy.id).ids
+    expect(other_project_rule_ids).to contain_exactly(other_project_escalation.rule_id)
+
+    # Assert non-null constraint was effectively applied
+    expect(escalations.where(rule_id: nil)).to be_empty
+    expect { create_escalation(rule_id: nil) }.to raise_error(ActiveRecord::NotNullViolation)
+  end
+
+  context 'without escalations' do
+    it 'adds a non-null constraint for rule_id' do
+      migrate!
+
+      expect(escalations.where(rule_id: nil)).to be_empty
+      expect { create_escalation(rule_id: nil) }.to raise_error(ActiveRecord::NotNullViolation)
+    end
+  end
+
+  private
+
+  def create_escalation(options = {})
+    escalations.create!(
+      schedule_id: schedule.id,
+      status: 1,
+      alert_id: alert.id,
+      process_at: Time.current,
+      **options
+    )
+  end
+end

ee/spec/migrations/remove_schedule_and_status_null_constraints_from_pending_escalations_alert_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe RemoveScheduleAndStatusNullConstraintsFromPendingEscalationsAlert, :migration, schema: 20210721174441 do
+  let(:namespaces) { table(:namespaces) }
+  let(:projects) { table(:projects) }
+  let(:schedules) { table(:incident_management_oncall_schedules) }
+  let(:policies) { table(:incident_management_escalation_policies) }
+  let(:rules) { table(:incident_management_escalation_rules) }
+  let(:alerts) { table(:alert_management_alerts) }
+  let(:escalations) { partitioned_table(:incident_management_pending_alert_escalations, by: :process_at) }
+
+  let!(:namespace) { namespaces.create!(name: 'namespace', path: 'namespace') }
+  let!(:project) { projects.create!(name: 'project', path: 'project', namespace_id: namespace.id) }
+  let!(:schedule) { schedules.create!(name: 'Schedule', iid: 1, project_id: project.id) }
+  let!(:policy) { policies.create!(name: 'Policy', project_id: project.id) }
+  let!(:rule) { rules.create!(oncall_schedule_id: schedule.id, policy_id: policy.id, status: 2, elapsed_time_seconds: 5.minutes) }
+  let!(:alert) { alerts.create!(iid: 1, project_id: project.id, title: 'Alert 1', started_at: Time.current) }
+
+  let!(:other_schedule) { schedules.create!(name: 'Other Schedule', iid: 2, project_id: project.id) }
+  let!(:other_rule) { rules.create!(oncall_schedule_id: other_schedule.id, policy_id: policy.id, status: 2, elapsed_time_seconds: 5.minutes) }
+  let!(:other_alert) { alerts.create!(iid: 2, project_id: project.id, title: 'Alert 2', started_at: Time.current) }
+
+  let(:rule_only_escalation) { create_escalation }
+  let(:duplicate_escalation) { create_escalation }
+  let(:other_rule_escalation) { create_escalation(rule_id: other_rule.id) }
+  let(:other_alert_escalation) { create_escalation(alert_id: other_alert.id) }
+  let(:all_escalations) { [rule_only_escalation, duplicate_escalation, other_rule_escalation, other_alert_escalation] }
+
+  it 'reversibly removes the non-null constraint for schedule_id and status', :aggregate_failures do
+    reversible_migration do |migration|
+      migration.before -> {
+        expect { create_escalation(status: rule.status, schedule_id: rule.oncall_schedule_id) }.not_to raise_error
+        expect { create_escalation }.to raise_error(ActiveRecord::NotNullViolation)
+
+        escalations.all.each do |escalation|
+          rule = rules.find(escalation.rule_id)
+          expect(escalation).to have_attributes(schedule_id: rule.oncall_schedule_id, status: rule.status)
+        end
+      }
+
+      migration.after -> {
+        expect { all_escalations }.not_to raise_error
+      }
+    end
+  end
+
+  context 'without escalations' do
+    it 'removes the non-null constraint for schedule_id and status' do
+      migrate!
+
+      expect { create_escalation(status: 1) }.not_to raise_error
+      expect { create_escalation(schedule_id: schedule.id) }.not_to raise_error
+    end
+  end
+
+  private
+
+  def create_escalation(options = {})
+    escalations.create!(
+      rule_id: rule.id,
+      alert_id: alert.id,
+      process_at: Time.current,
+      **options
+    )
+  end
+end

ee/spec/models/incident_management/escalation_policy_spec.rb

@@ -3,8 +3,6 @@
 require 'spec_helper'
 
 RSpec.describe IncidentManagement::EscalationPolicy do
-  let_it_be(:project) { create(:project) }
-
   subject { build(:incident_management_escalation_policy) }
 
   it { is_expected.to be_valid }
@@ -12,7 +10,17 @@ RSpec.describe IncidentManagement::EscalationPolicy do
   describe 'associations' do
     it { is_expected.to belong_to(:project) }
     it { is_expected.to have_many(:rules) }
-    it { is_expected.to have_many(:ordered_rules).order(elapsed_time_seconds: :asc, status: :asc) }
+    it { is_expected.to have_many(:active_rules).order(elapsed_time_seconds: :asc, status: :asc).class_name('EscalationRule').inverse_of(:policy) }
+
+    describe '.active_rules' do
+      let(:policy) { create(:incident_management_escalation_policy) }
+      let(:rule) { policy.rules.first }
+      let(:removed_rule) { create(:incident_management_escalation_rule, :removed, policy: policy) }
+
+      subject { policy.reload.active_rules }
+
+      it { is_expected.to contain_exactly(rule) }
+    end
   end
 
   describe 'validations' do

ee/spec/models/incident_management/escalation_rule_spec.rb

@@ -4,7 +4,6 @@ require 'spec_helper'
 
 RSpec.describe IncidentManagement::EscalationRule do
   let_it_be(:policy) { create(:incident_management_escalation_policy) }
-  let_it_be(:schedule) { create(:incident_management_oncall_schedule, project: policy.project) }
 
   subject { build(:incident_management_escalation_rule, policy: policy) }
 
@@ -21,4 +20,21 @@ RSpec.describe IncidentManagement::EscalationRule do
     it { is_expected.to validate_numericality_of(:elapsed_time_seconds).is_greater_than_or_equal_to(0).is_less_than_or_equal_to(24.hours) }
     it { is_expected.to validate_uniqueness_of(:policy_id).scoped_to([:oncall_schedule_id, :status, :elapsed_time_seconds] ).with_message('must have a unique schedule, status, and elapsed time') }
   end
+
+  describe 'scopes' do
+    let_it_be(:rule) { policy.rules.first }
+    let_it_be(:removed_rule) { create(:incident_management_escalation_rule, :removed, policy: policy) }
+
+    describe '.not_removed' do
+      subject { described_class.not_removed }
+
+      it { is_expected.to contain_exactly(rule) }
+    end
+
+    describe '.removed' do
+      subject { described_class.removed }
+
+      it { is_expected.to contain_exactly(removed_rule) }
+    end
+  end
 end

ee/spec/models/incident_management/pending_escalations/alert_spec.rb

@@ -9,14 +9,11 @@ RSpec.describe IncidentManagement::PendingEscalations::Alert do
 
   describe 'validations' do
     it { is_expected.to validate_presence_of(:process_at) }
-    it { is_expected.to validate_presence_of(:status) }
     it { is_expected.to delegate_method(:project).to(:alert) }
-    it { is_expected.to delegate_method(:policy).to(:rule).allow_nil }
     it { is_expected.to validate_uniqueness_of(:rule_id).scoped_to([:alert_id]) }
   end
 
   describe 'associations' do
-    it { is_expected.to belong_to(:oncall_schedule) }
     it { is_expected.to belong_to(:alert) }
     it { is_expected.to belong_to(:rule) }
   end

ee/spec/requests/api/graphql/mutations/incident_management/escalation_policy/update_spec.rb

@@ -74,7 +74,7 @@ RSpec.describe 'Updating an escalation policy' do
     expect(escalation_policy.reload).to have_attributes(
       name: variables[:name],
       description: variables[:description],
-      rules: [
+      active_rules: [
         have_attributes(
           oncall_schedule: schedule,
           status: rule_variables[:status].downcase,

ee/spec/services/incident_management/escalation_policies/update_service_spec.rb

@@ -38,6 +38,7 @@ RSpec.describe IncidentManagement::EscalationPolicies::UpdateService do
   end
 
   let(:new_rule) { have_attributes(**new_rule_params.except(:status), status: 'acknowledged') }
+  let(:removed_rules) { [] }
 
   before do
     stub_licensed_features(oncall_schedules: true, escalation_policies: true)
@@ -62,7 +63,8 @@ RSpec.describe IncidentManagement::EscalationPolicies::UpdateService do
 
         expect(execute.payload).to eq(escalation_policy: escalation_policy.reload)
         expect(escalation_policy).to have_attributes(params.slice(:name, :description))
-        expect(escalation_policy.rules).to match_array(expected_rules)
+        expect(escalation_policy.active_rules).to match_array(expected_rules)
+        expect(escalation_policy.rules.removed).to match_array(removed_rules)
       end
     end
 
@@ -97,6 +99,7 @@ RSpec.describe IncidentManagement::EscalationPolicies::UpdateService do
     context 'when all old rules are replaced' do
       let(:rule_params) { [new_rule_params] }
       let(:expected_rules) { [new_rule] }
+      let(:removed_rules) { escalation_rules }
 
       it_behaves_like 'successful update with no errors'
     end
@@ -104,6 +107,15 @@ RSpec.describe IncidentManagement::EscalationPolicies::UpdateService do
     context 'when some rules are preserved, added, and deleted' do
       let(:rule_params) { [existing_rules_params.first, new_rule_params] }
       let(:expected_rules) { [escalation_rules.first, new_rule] }
+      let(:removed_rules) { [escalation_rules.last] }
+
+      it_behaves_like 'successful update with no errors'
+    end
+
+    context 'when rules are only deleted' do
+      let(:rule_params) { [existing_rules_params.first] }
+      let(:expected_rules) { [escalation_rules.first] }
+      let(:removed_rules) { [escalation_rules.last] }
 
       it_behaves_like 'successful update with no errors'
     end

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

@@ -8,6 +8,7 @@ RSpec.describe IncidentManagement::PendingEscalations::CreateService do
   let_it_be(:rule_count) { 2 }
 
   let!(:escalation_policy) { create(:incident_management_escalation_policy, project: project, rule_count: rule_count) }
+  let!(:removed_rule) { create(:incident_management_escalation_rule, :removed, policy: escalation_policy) }
   let(:rules) { escalation_policy.rules }
 
   let(:service) { described_class.new(target) }
@@ -46,6 +47,7 @@ RSpec.describe IncidentManagement::PendingEscalations::CreateService do
 
     context 'when there is no escalation policy for the project' do
       let!(:escalation_policy) { nil }
+      let!(:removed_rule) { nil }
 
       it 'does nothing' do
         expect { execute }.not_to change { IncidentManagement::PendingEscalations::Alert.count }
@@ -64,8 +66,6 @@ RSpec.describe IncidentManagement::PendingEscalations::CreateService do
       expect(escalation).to have_attributes(
         rule_id: rule.id,
         alert_id: target.id,
-        schedule_id: rule.oncall_schedule_id,
-        status: rule.status,
         process_at: be_within(1.minute).of(rule.elapsed_time_seconds.seconds.after(execution_time))
       )
     end

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

@@ -15,7 +15,7 @@ RSpec.describe IncidentManagement::PendingEscalations::ProcessService do
 
   let(:target) { alert }
   let(:process_at) { 5.minutes.ago }
-  let(:escalation) { create(:incident_management_pending_alert_escalation, rule: escalation_rule, oncall_schedule: schedule_1, target: target, status: IncidentManagement::EscalationRule.statuses[:acknowledged], process_at: process_at) }
+  let(:escalation) { create(:incident_management_pending_alert_escalation, rule: escalation_rule, alert: target, process_at: process_at) }
 
   let(:service) { described_class.new(escalation) }
 
@@ -31,7 +31,7 @@ RSpec.describe IncidentManagement::PendingEscalations::ProcessService do
 
       it 'does not delete the escalation' do
         subject
-        expect { escalation.reload }.not_to raise_error(ActiveRecord::RecordNotFound)
+        expect { escalation.reload }.not_to raise_error
       end
     end
 
@@ -50,7 +50,7 @@ RSpec.describe IncidentManagement::PendingEscalations::ProcessService do
 
       it 'creates a system note' do
         expect(SystemNoteService)
-          .to receive(:notify_via_escalation).with(alert, project, [a_kind_of(User)], escalation_policy, schedule_1)
+          .to receive(:notify_via_escalation).with(alert, project, [a_kind_of(User)], escalation_policy)
           .and_call_original
 
         expect { execute }.to change(Note, :count).by(1)

ee/spec/services/system_notes/escalations_service_spec.rb

@@ -9,10 +9,9 @@ RSpec.describe SystemNotes::EscalationsService do
   let_it_be(:author) { User.alert_bot }
 
   describe '#notify_via_escalation' do
-    subject { described_class.new(noteable: noteable, project: project).notify_via_escalation([user, user_2], escalation_policy: escalation_policy, oncall_schedule: oncall_schedule) }
+    subject { described_class.new(noteable: noteable, project: project).notify_via_escalation([user, user_2], escalation_policy: escalation_policy) }
 
     let_it_be(:escalation_policy) { create(:incident_management_escalation_policy, project: project) }
-    let_it_be(:oncall_schedule) { create(:incident_management_oncall_schedule, project: project) }
     let_it_be(:noteable) { create(:alert_management_alert, project: project) }
 
     it_behaves_like 'a system note' do
@@ -22,13 +21,5 @@ RSpec.describe SystemNotes::EscalationsService do
     it 'posts the correct text to the system note' do
       expect(subject.note).to match("notified #{user.to_reference} and #{user_2.to_reference} of this alert via escalation policy **#{escalation_policy.name}**")
     end
-
-    context 'when policy is missing' do
-      let_it_be(:escalation_policy) { nil }
-
-      it 'posts the correct text to the system note' do
-        expect(subject.note).to match("notified #{user.to_reference} and #{user_2.to_reference} of this alert via schedule **#{oncall_schedule.name}**, per an escalation rule which no longer exists")
-      end
-    end
   end
 end

spec/support/helpers/migrations_helpers.rb

@@ -16,6 +16,23 @@ module MigrationsHelpers
     end
   end
 
+  def partitioned_table(name, by: :created_at, strategy: :monthly)
+    klass = Class.new(active_record_base) do
+      include PartitionedTable
+
+      self.table_name = name
+      self.primary_key = :id
+
+      partitioned_by by, strategy: strategy
+
+      def self.name
+        table_name.singularize.camelcase
+      end
+    end
+
+    klass.tap { Gitlab::Database::Partitioning::PartitionManager.new.sync_partitions }
+  end
+
   def migrations_paths
     ActiveRecord::Migrator.migrations_paths
   end