feat: Enable per-alert Incident Severity

src/robusta/core/sinks/incidentio/incidentio_sink.py

@@ -23,6 +23,10 @@ class IncidentioSink(SinkBase):
     def __init__(self, sink_config: IncidentioSinkConfigWrapper, registry):
         super().__init__(sink_config.incidentio_sink, registry)
         self.source_config_id = sink_config.incidentio_sink.source_config_id
+        self.severity_alert_label_name = sink_config.incidentio_sink.severity_alert_label_name
+        self.severity_default = sink_config.incidentio_sink.severity_default
+        self.dashboard_url_annotation_name = sink_config.incidentio_sink.dashboard_url_annotation_name
+        self.runbook_url_annotation_name = sink_config.incidentio_sink.runbook_url_annotation_name
         self.client = IncidentIoClient(
             base_url=sink_config.incidentio_sink.base_url,
             token=sink_config.incidentio_sink.token
@@ -40,6 +44,10 @@ def __send_event_to_incidentio(self, finding: Finding, platform_enabled: bool) -
         metadata: Dict[str, Any] = {}
         links: List[Dict[str, str]] = []
 
+        # Add additional link for a dashboard (ie Grafana)
+        dashboard_url = finding.subject.annotations.get(self.dashboard_url_annotation_name)
+        source_url = dashboard_url or ""
+
         # Add Robusta links if platform is enabled
         if platform_enabled:
             links.append(
@@ -49,15 +57,33 @@ def __send_event_to_incidentio(self, finding: Finding, platform_enabled: bool) -
                 }
             )
 
+            source_url = finding.get_source_uri(self.account_id, self.cluster_name)
+
         # Collect metadata
         metadata["resource"] = finding.subject.name
         metadata["namespace"] = finding.subject.namespace
         metadata["cluster"] = self.cluster_name
-        metadata["severity"] = finding.severity.name
         metadata["description"] = finding.description or ""
         metadata["source"] = finding.source.name
         metadata["fingerprint_id"] = finding.fingerprint
 
+        # Define Incident Severity based on Alert Label
+        severity_from_label = finding.subject.labels.get(self.severity_alert_label_name)
+        metadata["severity"] = severity_from_label or self.severity_default
+
+        # Add additional link if a runbook is provided
+        runbook_url = finding.subject.annotations.get(self.runbook_url_annotation_name)
+        metadata["runbook_url"] = runbook_url or ""
+
+        # Log Debug Information
+        logging.debug(
+            f"--Incident.io Sink Information--\n"
+            f"finding:{finding}\n"
+            f"labels:{finding.subject.labels}\n"
+            f"annotations:{finding.subject.annotations}\n"
+            f"metadata:{metadata}\n"
+        )
+
         # Convert blocks to metadata
         for enrichment in finding.enrichments:
             for block in enrichment.blocks:
@@ -71,7 +97,7 @@ def __send_event_to_incidentio(self, finding: Finding, platform_enabled: bool) -
             "description": finding.description or "No description provided.",
             "status": self.__to_incidentio_status_type(finding.title),
             "metadata": metadata,
-            "source_url": finding.get_investigate_uri(self.account_id, self.cluster_name),
+            "source_url": source_url,
             "links": links,
         }
 

src/robusta/core/sinks/incidentio/incidentio_sink_params.py

@@ -12,6 +12,10 @@ class IncidentioSinkParams(SinkBaseParams):
     base_url: Optional[str] = "https://api.incident.io/v2/"
     token: str
     source_config_id: str
+    severity_alert_label_name: Optional[str] = "robusta_incidentio_severity"
+    severity_default: str
+    dashboard_url_annotation_name: Optional[str] = "dashboard_url"
+    runbook_url_annotation_name: Optional[str] = "runbook_url"
 
     @classmethod
     def _get_sink_type(cls):