feat: Add workflow trigger for incident called 'alert_association_changed'

keep-ui/entities/workflows/model/schema.ts

@@ -48,7 +48,7 @@ export const V2StepAlertTriggerSchema = TriggerSchemaBase.extend({
     .optional(),
 });
 
-export const IncidentEventEnum = z.enum(["created", "updated", "deleted"]);
+export const IncidentEventEnum = z.enum(["created", "updated", "deleted", "alert_association_changed"]);
 
 const IncidentTriggerValueSchema = z.object({
   events: z.array(IncidentEventEnum),

keep-ui/features/workflows/builder/ui/Editor/TriggerEditor.tsx

@@ -199,7 +199,7 @@ export function TriggerEditor() {
         return (
           <>
             <Subtitle className="mt-2.5">Incident events</Subtitle>
-            {Array("created", "updated", "deleted").map((event) => (
+            {Array("created", "updated", "deleted", "alert_association_changed").map((event) => (
               <div key={`incident-${event}`} className="flex">
                 <Switch
                   id={event}

keep/api/bl/incidents_bl.py

@@ -159,6 +159,7 @@ async def add_alerts_to_incident(
                 "alert_fingerprints": alert_fingerprints,
             },
         )
+
         self.__postprocess_alerts_change(incident, alert_fingerprints)
         await self.__generate_summary(incident_id, incident)
         self.logger.info(
@@ -268,6 +269,7 @@ def delete_alerts_from_incident(
         remove_alerts_to_incident_by_incident_id(
             self.tenant_id, incident_id, alert_fingerprints
         )
+
         self.__postprocess_alerts_change(incident, alert_fingerprints)
 
     def delete_incident(self, incident_id: UUID) -> None:
@@ -343,6 +345,7 @@ def __postprocess_alerts_change(self, incident, alert_fingerprints):
                 "alert_fingerprints": alert_fingerprints,
             },
         )
+        self.send_workflow_event(incident_dto, "alert_association_changed")
 
     def update_severity(
         self,

keep/rulesengine/rulesengine.py

@@ -208,6 +208,10 @@ def _run_cel_rules(
                                     self.tenant_id, session, incident_dto, "updated"
                                 )
 
+                            RulesEngine.send_workflow_event(
+                                self.tenant_id, session, incident_dto, "alert_association_changed"
+                            )
+
                             incidents_dto[incident.id] = incident_dto
 
                 else:

keep/topologies/topology_processor.py

@@ -377,3 +377,7 @@ def _create_application_based_incident(
             # Trigger the workflow event
             RulesEngine.send_workflow_event(tenant_id, session, incident_dto, "created")
             self.logger.info(f"Created new incident for application {application.name}")
+
+            RulesEngine.send_workflow_event(
+                tenant_id, session, incident_dto, "alert_association_changed"
+            )

keep/workflowmanager/workflowmanager.py

@@ -162,6 +162,34 @@ def insert_incident(self, tenant_id: str, incident: IncidentDto, trigger: str):
                 for k, v in incident_enrichment.enrichments.items():
                     setattr(incident, k, v)
 
+            if trigger == "alert_association_changed":
+                try:
+                    alerts = incident.alerts
+
+                    processed_alerts = []
+                    # Iterate over the alerts and process them as needed
+                    for alert in alerts:
+                        # Handle multiline description
+                        alert_description = alert.description.split("\n")[0]
+
+                        processed_alert = f"{alert.status.capitalize()} {alert.lastReceived} [{alert.severity}] {alert_description}"
+                        processed_alerts.append(processed_alert)
+
+                    # Add the linked alerts to the incident object for use in the workflow
+                    setattr(incident, 'linked_alerts', processed_alerts)
+
+                except Exception as e:
+                    self.logger.error(
+                        f"Failed to fetch alerts linked to incident {incident.id}",
+                        extra={
+                            "incident_id": incident.id,
+                            "tenant_id": tenant_id,
+                            "exception": str(e)
+                        }
+                    )
+                    # Set empty list if fetch fails
+                    setattr(incident, 'linked_alerts', [])
+
             self.logger.info("Adding workflow to run")
             with self.scheduler.lock:
                 self.scheduler.workflows_to_run.append(

tests/test_incidents.py

@@ -1396,11 +1396,11 @@ async def test_incident_bl_add_alert_to_incident(db_session, create_alert):
             assert data["incident_id"] == str(incident_dto.id)
 
             # Check workflow manager
-            assert len(workflow_manager.events) == 2  # Created, update
+            assert len(workflow_manager.events) == 3  # Created, update, alert_association_changed
             wf_tenant_id, wf_incident_dto, wf_action = workflow_manager.events[-1]
             assert wf_tenant_id == SINGLE_TENANT_UUID
             assert wf_incident_dto.id == incident_dto.id
-            assert wf_action == "updated"
+            assert wf_action == "alert_association_changed"
 
             # Check elastic
             assert len(elastic_client.alerts) == 1
@@ -1490,12 +1490,12 @@ async def test_incident_bl_delete_alerts_from_incident(db_session, create_alert)
             assert data["incident_id"] == str(incident_dto.id)
 
             # Check workflow manager
-            # Created, updated (added event), updated(deleted event)
-            assert len(workflow_manager.events) == 3
+            # Created, updated (added event), alert_association_change(added event), updated(deleted event), alert_association_changed(deleted event)
+            assert len(workflow_manager.events) == 5
             wf_tenant_id, wf_incident_dto, wf_action = workflow_manager.events[-1]
             assert wf_tenant_id == SINGLE_TENANT_UUID
             assert wf_incident_dto.id == incident_dto.id
-            assert wf_action == "updated"
+            assert wf_action == "alert_association_changed"
 
             # Check elastic
             assert len(elastic_client.alerts) == 2

tests/test_workflow_execution.py

@@ -766,6 +766,26 @@ def test_workflow_execution_with_disabled_workflow(
         "deleted incident: {{ incident.name }}"
 """
 
+workflow_definition_6 = """workflow:
+id: incident-triggers-test-alert-association-changed
+description: test incident alert association change
+triggers:
+- type: incident
+  events:
+  - alert_association_changed
+name: alert_association_changed
+owners: []
+services: []
+steps: []
+actions:
+- name: mock-action
+  provider:
+    type: console
+    with:
+      message: |
+        "linked_alerts: keep.len({{ incident.linked_alerts }})"
+"""
+
 
 @pytest.mark.timeout(15)
 @pytest.mark.parametrize(
@@ -792,6 +812,15 @@ def test_workflow_incident_triggers(
     db_session.add(workflow_created)
     db_session.commit()
 
+    alert_1 = AlertDto(
+        id="alert-1",
+        name="Test Alert 1",
+        status=AlertStatus.FIRING,
+        severity=AlertSeverity.HIGH,
+        lastReceived="2025-01-30T10:00:00Z",
+        description="Test alert 1 description\nThis is a multiline description\nWith multiple lines of content\nTo test the split behavior"
+    )
+
     # Create the current alert
     incident = IncidentDto(
         id="ba9ddbb9-3a83-40fc-9ace-1e026e08ca2b",
@@ -802,6 +831,7 @@ def test_workflow_incident_triggers(
         severity="critical",
         is_predicted=False,
         is_candidate=False,
+        alerts=[alert_1]
     )
 
     # Insert the current alert into the workflow manager
@@ -861,6 +891,31 @@ def test_workflow_incident_triggers(
         '"deleted incident: incident"\n'
     ]
 
+    workflow_deleted = Workflow(
+        id="incident-triggers-test-alert-association-changed",
+        name="incident-triggers-test-alert-association-changed",
+        tenant_id=SINGLE_TENANT_UUID,
+        description="Check that incident triggers works",
+        created_by="[email protected]",
+        interval=0,
+        workflow_raw=workflow_definition_6,
+    )
+    db_session.add(workflow_deleted)
+    db_session.commit()
+
+    workflow_manager.insert_incident(SINGLE_TENANT_UUID, incident, "alert_association_changed")
+    assert len(workflow_manager.scheduler.workflows_to_run) == 1
+
+    workflow_execution_alert_association_changed = wait_for_workflow_execution(
+        SINGLE_TENANT_UUID, "incident-triggers-test-alert-association-changed"
+    )
+    assert workflow_execution_alert_association_changed is not None
+    assert workflow_execution_alert_association_changed.status == "success"
+    assert workflow_execution_alert_association_changed.results["mock-action"] == [
+        '"linked_alerts: 1"\n'
+    ]
+    assert len(workflow_manager.scheduler.workflows_to_run) == 0
+
 
 # @pytest.mark.parametrize(
 #     "test_app, test_case, alert_statuses, expected_tier, db_session",

tests/test_workflowmanager.py

@@ -1,11 +1,16 @@
 from pathlib import Path
+import uuid
+from datetime import datetime
 from unittest.mock import Mock, patch
 
 import pytest
 from fastapi import HTTPException
 
 from keep.api.routes.workflows import get_event_from_body
 from keep.parser.parser import Parser
+from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus
+from keep.api.models.incident import IncidentDto
+from keep.api.models.db.incident import IncidentSeverity, IncidentStatus
 
 # Assuming WorkflowParser is the class containing the get_workflow_from_dict method
 from keep.workflowmanager.workflow import Workflow
@@ -182,3 +187,102 @@ def test_handle_manual_event_workflow_test_run():
         )
         assert workflow_scheduler.workflows_to_run[0]["test_run"] == True
         assert workflow_scheduler.workflows_to_run[0]["workflow"] == mock_workflow
+
+def test_insert_incident_alert_association_changed_adds_linked_alerts():
+    """Test that linked_alerts key is present when workflow trigger is alert_association_changed."""
+
+    # Create mock alerts that would be associated with the incident
+    mock_alert_1 = AlertDto(
+        id="alert-1",
+        name="Test Alert 1",
+        status=AlertStatus.FIRING,
+        severity=AlertSeverity.HIGH,
+        lastReceived="2025-01-30T10:00:00Z",
+        description="Test alert 1 description\nThis is a multiline description\nWith multiple lines of content\nTo test the split behavior"
+    )
+
+    mock_alert_2 = AlertDto(
+        id="alert-2",
+        name="Test Alert 2",
+        status=AlertStatus.RESOLVED,
+        severity=AlertSeverity.CRITICAL,
+        lastReceived="2025-01-30T11:00:00Z",
+        description="Test alert 2 description"
+    )
+
+    # Create incident DTO with mock alerts
+    incident_dto = IncidentDto(
+        id=uuid.uuid4(),
+        user_generated_name="Test Incident",
+        alerts_count=2,
+        alert_sources=["prometheus", "grafana"],
+        services=["web-service"],
+        severity=IncidentSeverity.HIGH,
+        status=IncidentStatus.FIRING,
+        is_predicted=False,
+        is_candidate=False,
+        creation_time=datetime.utcnow()
+    )
+
+    # Mock the alerts property to return our test alerts
+    incident_dto._alerts = [mock_alert_1, mock_alert_2]
+
+    # Create a mock workflow with alert_association_changed trigger
+    mock_workflow = Mock(spec=Workflow)
+    mock_workflow.workflow_triggers = [
+        {
+            "type": "incident",
+            "events": ["alert_association_changed"]
+        }
+    ]
+
+    # Create mock workflow model
+    mock_workflow_model = Mock()
+    mock_workflow_model.id = "test-workflow-id"
+    mock_workflow_model.name = "test-workflow"
+    mock_workflow_model.tenant_id = "test-tenant"
+    mock_workflow_model.is_disabled = False
+
+    # Create WorkflowManager and mock dependencies
+    workflow_manager = WorkflowManager()
+
+    with patch.object(workflow_manager.workflow_store, 'get_all_workflows') as mock_get_workflows, \
+         patch.object(workflow_manager, '_get_workflow_from_store') as mock_get_workflow, \
+         patch('keep.workflowmanager.workflowmanager.get_enrichment') as mock_get_enrichment:
+
+        # Set up mocks
+        mock_get_workflows.return_value = [mock_workflow_model]
+        mock_get_workflow.return_value = mock_workflow
+        mock_get_enrichment.return_value = None
+
+        # Mock the scheduler
+        workflow_manager.scheduler = Mock()
+        workflow_manager.scheduler.lock = Mock()
+        workflow_manager.scheduler.lock.__enter__ = Mock(return_value=None)
+        workflow_manager.scheduler.lock.__exit__ = Mock(return_value=None)
+        workflow_manager.scheduler.workflows_to_run = []
+
+        # Call insert_incident with alert_association_changed trigger
+        workflow_manager.insert_incident("test-tenant", incident_dto, "alert_association_changed")
+
+        # Verify workflow was added to run
+        assert len(workflow_manager.scheduler.workflows_to_run) == 1
+
+        # Get the workflow execution event
+        workflow_execution = workflow_manager.scheduler.workflows_to_run[0]
+        executed_incident = workflow_execution["event"]
+
+        # Verify that linked_alerts attribute was added to the incident
+        assert hasattr(executed_incident, 'linked_alerts'), "linked_alerts attribute should be present"
+
+        # Verify the content of linked_alerts
+        linked_alerts = executed_incident.linked_alerts
+        assert isinstance(linked_alerts, list), "linked_alerts should be a list"
+        assert len(linked_alerts) == 2, "Should have 2 linked alerts"
+
+        # Verify the format of linked alerts entries
+        expected_alert_1 = "Firing 2025-01-30T10:00:00.000Z [high] Test alert 1 description"
+        expected_alert_2 = "Resolved 2025-01-30T11:00:00.000Z [critical] Test alert 2 description"
+
+        assert expected_alert_1 in linked_alerts, f"Expected '{expected_alert_1}' in linked_alerts"
+        assert expected_alert_2 in linked_alerts, f"Expected '{expected_alert_2}' in linked_alerts"