MAIN-2798 - Opsgenie slack (#1673)

* Working opsgenie link and ack

* working version

* pre pr changes

* small changes

* pr changes

* loading json once

* added documentation and comment

---------

Co-authored-by: moshemorad <moshemorad12340@gmail.com>

docs/configuration/sinks/Opsgenie.rst

@@ -43,3 +43,27 @@ Save the file and run
     .. image:: /images/deployment-babysitter-opsgenie.png
       :width: 1000
       :align: center
+
+
+Action to connect Slack to OpsGenie
+------------------------------------------------
+
+The `opsgenie_slack_enricher` action enriches Slack alerts with OpsGenie integration. It performs the following:
+
+- Adds a button in Slack to acknowledge the OpsGenie alert directly.
+- Includes a link in Slack messages that redirects to the alert in OpsGenie for easy access.
+
+To use this action, ensure it is included in your playbook configuration.
+
+**Example Configuration:**
+
+.. code-block:: yaml
+
+   customPlaybooks:
+   - actions:
+     - opsgenie_slack_enricher:
+         url_base: team-name.app.eu.opsgenie.com
+     triggers:
+     - on_prometheus_alert: {}
+
+With this integration, teams can efficiently manage OpsGenie alerts directly from Slack.

docs/playbook-reference/actions/miscellaneous.rst

@@ -8,6 +8,11 @@ ArgoCD
 
 .. robusta-action:: playbooks.robusta_playbooks.argo_cd.argo_app_sync
 
+Slack-OpsGenie sync
+^^^^^^^^^^^^^^
+
+.. robusta-action:: playbooks.robusta_playbooks.sink_enrichments.opsgenie_slack_enricher
+
 Kubernetes Optimization
 -----------------------
 

playbooks/robusta_playbooks/sink_enrichments.py

@@ -0,0 +1,117 @@
+import logging
+from typing import Any, Optional
+from urllib.parse import urlparse
+
+from robusta.api import (
+    ActionParams,
+    CallbackBlock,
+    CallbackChoice,
+    ExecutionBaseEvent,
+    PrometheusKubernetesAlert,
+    action,
+)
+from robusta.core.reporting.base import Link, LinkType
+
+
+class SlackCallbackParams(ActionParams):
+    """
+    :var slack_username: The username that clicked the slack callback. - Auto-populated by slack
+    :var slack_message: The message from the slack callback. - Auto-populated by slack
+    """
+
+    slack_username: Optional[str]
+    slack_message: Optional[Any]
+
+
+class OpsGenieAckParams(SlackCallbackParams):
+    """
+    :var alertmanager_url: Alternative Alert Manager url to send requests.
+    """
+
+    alert_fingerprint: str
+
+
+@action
+def ack_opsgenie_alert_from_slack(event: ExecutionBaseEvent, params: OpsGenieAckParams):
+    """
+    Sends an ack to opsgenie alert
+    """
+    event.emit_event(
+        "opsgenie_ack",
+        fingerprint=params.alert_fingerprint,
+        user=params.slack_username,
+        note=f"This alert was ack-ed from a Robusta Slack message by {params.slack_username}",
+    )
+
+    if not params.slack_message:
+        logging.warning("No action Slack found, unable to update slack message.")
+        return
+
+    # slack action block
+    actions = params.slack_message.get("actions", [])
+    if not actions:
+        logging.warning("No actions found in the Slack message.")
+        return
+
+    block_id = actions[0].get("block_id")
+    if not block_id:
+        logging.warning("Block ID is missing in the first action of the Slack message.")
+        return
+
+    event.emit_event(
+        "replace_callback_with_string",
+        slack_message=params.slack_message,
+        block_id=block_id,
+        message_string=f"✅ *OpsGenie Ack by @{params.slack_username}*",
+    )
+
+
+class OpsGenieLinkParams(ActionParams):
+    """
+    :var url_base: The base url for your opsgenie account for example: "robusta-test-url.app.eu.opsgenie.com"
+    """
+
+    url_base: str
+
+
+@action
+def opsgenie_slack_enricher(alert: PrometheusKubernetesAlert, params: OpsGenieLinkParams):
+    """
+    Adds a button in slack to ack an opsGenie alert
+    Adds a Link to slack to the alert in opsgenie
+    """
+    normalized_url_base = normalize_url_base(params.url_base)
+    alert.add_link(
+        Link(
+            url=f"https://{normalized_url_base}/alert/list?query=alias:{alert.alert.fingerprint}",
+            name="OpsGenie Alert",
+            type=LinkType.OPSGENIE_LIST_ALERT_BY_ALIAS,
+        )
+    )
+
+    alert.add_enrichment(
+        [
+            CallbackBlock(
+                {
+                    f"Ack Opsgenie Alert": CallbackChoice(
+                        action=ack_opsgenie_alert_from_slack,
+                        action_params=OpsGenieAckParams(
+                            alert_fingerprint=alert.alert.fingerprint,
+                        ),
+                    )
+                },
+            )
+        ]
+    )
+
+
+def normalize_url_base(url_base: str) -> str:
+    """
+    Normalize the url_base to remove 'https://' or 'http://' and any trailing slashes.
+    """
+    # Remove the scheme (http/https) if present
+    parsed_url = urlparse(url_base)
+    url_base = parsed_url.netloc if parsed_url.netloc else parsed_url.path
+
+    # Remove trailing slash if present
+    return url_base.rstrip("/")

src/robusta/core/reporting/base.py

@@ -6,11 +6,11 @@
 from abc import ABC, abstractmethod
 from datetime import datetime
 from enum import Enum
-from strenum import StrEnum
 from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlencode
 
 from pydantic.main import BaseModel
+from strenum import StrEnum
 
 from robusta.core.discovery.top_service_resolver import TopServiceResolver
 from robusta.core.model.env_vars import ROBUSTA_UI_DOMAIN
@@ -94,6 +94,7 @@ def to_emoji(self) -> str:
 class LinkType(StrEnum):
     VIDEO = "video"
     PROMETHEUS_GENERATOR_URL = "prometheus_generator_url"
+    OPSGENIE_LIST_ALERT_BY_ALIAS = "opsgenie_list_alert_by_alias"
 
 
 class Link(BaseModel):

src/robusta/core/sinks/opsgenie/opsgenie_sink.py

@@ -33,9 +33,17 @@ def __init__(self, sink_config: OpsGenieSinkConfigWrapper, registry):
         if sink_config.opsgenie_sink.host is not None:
             self.conf.host = sink_config.opsgenie_sink.host
 
+        self.registry.subscribe("opsgenie_ack", self)
+
         self.api_client = opsgenie_sdk.api_client.ApiClient(configuration=self.conf)
         self.alert_api = opsgenie_sdk.AlertApi(api_client=self.api_client)
 
+    def handle_event(self, event_name: str, **kwargs):
+        if event_name == "opsgenie_ack":
+            self.__ack_alert(**kwargs)
+        else:
+            logging.warning(f"OpsGenieSink subscriber called with unknown event {event_name}")
+
     def __close_alert(self, finding: Finding):
         body = opsgenie_sdk.CloseAlertPayload(
             user="Robusta",
@@ -51,6 +59,21 @@ def __close_alert(self, finding: Finding):
         except opsgenie_sdk.ApiException as err:
             logging.error(f"Error closing opsGenie alert {finding} {err}", exc_info=True)
 
+    def __ack_alert(self, fingerprint: str, user: str, note: str):
+        body = opsgenie_sdk.AcknowledgeAlertPayload(
+            user=user,
+            note=note,
+            source="Robusta",
+        )
+        try:
+            self.alert_api.acknowledge_alert(
+                identifier=fingerprint,
+                acknowledge_alert_payload=body,
+                identifier_type="alias",
+            )
+        except opsgenie_sdk.ApiException as err:
+            logging.error(f"Error acking opsGenie alert {fingerprint} {err}", exc_info=True)
+
     def __open_alert(self, finding: Finding, platform_enabled: bool):
         description = self.__to_description(finding, platform_enabled)
         details = self.__to_details(finding)

src/robusta/core/sinks/slack/slack_sink.py

@@ -1,3 +1,5 @@
+import logging
+
 from robusta.core.model.env_vars import ROBUSTA_UI_DOMAIN
 from robusta.core.reporting.base import Finding, FindingStatus
 from robusta.core.sinks.sink_base import NotificationGroup, NotificationSummary, SinkBase
@@ -15,6 +17,13 @@ def __init__(self, sink_config: SlackSinkConfigWrapper, registry):
         self.slack_sender = slack_module.SlackSender(
             self.api_key, self.account_id, self.cluster_name, self.signing_key, self.slack_channel
         )
+        self.registry.subscribe("replace_callback_with_string", self)
+
+    def handle_event(self, event_name: str, **kwargs):
+        if event_name == "replace_callback_with_string":
+            self.__replace_callback_with_string(**kwargs)
+        else:
+            logging.warning("SlackSink subscriber called with unknown event")
 
     def write_finding(self, finding: Finding, platform_enabled: bool) -> None:
         if self.grouping_enabled:
@@ -75,6 +84,48 @@ def handle_notification_grouping(self, finding: Finding, platform_enabled: bool)
                     finding, self.params, platform_enabled, thread_ts=slack_thread_ts
                 )
 
-
     def get_timeline_uri(self, account_id: str, cluster_name: str) -> str:
         return f"{ROBUSTA_UI_DOMAIN}/graphs?account_id={account_id}&cluster={cluster_name}"
+
+    def __replace_callback_with_string(self, slack_message, block_id, message_string):
+        """
+        Replace a specific block in a Slack message with a given string while preserving other blocks.
+
+        Args:
+            slack_message (dict): The payload received from Slack.
+            block_id (str): The ID of the block to replace.
+            message_string (str): The text to replace the block content with.
+        """
+        try:
+            # Extract required fields
+            channel_id = slack_message.get("channel", {}).get("id")
+            message_ts = slack_message.get("container", {}).get("message_ts")
+            blocks = slack_message.get("message", {}).get("blocks", [])
+
+            # Validate required fields
+            if not channel_id or not message_ts or not blocks:
+                raise ValueError("Missing required fields: channel_id, message_ts, or blocks.")
+
+            # Update the specific block
+            for i, block in enumerate(blocks):
+                if block.get("block_id") == block_id:
+                    blocks[i] = {
+                        "type": "section",
+                        "block_id": block_id,
+                        "text": {
+                            "type": "mrkdwn",
+                            "text": message_string
+                        }
+                    }
+                    break
+
+            # Call the shorter update function
+            return self.slack_sender.update_slack_message(
+                channel=channel_id,
+                ts=message_ts,
+                blocks=blocks,
+                text=message_string
+            )
+
+        except Exception as e:
+            logging.exception(f"Error updating Slack message: {e}")

src/robusta/integrations/receiver.py

@@ -5,15 +5,14 @@
 import logging
 import os
 import time
-from threading import Thread
-from typing import Dict, Optional, List, Union
-from uuid import UUID
-
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import nullcontext
+from threading import Thread
+from typing import Any, Dict, List, Optional, Union
+from uuid import UUID
 
-import websocket
 import sentry_sdk
+import websocket
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
@@ -22,9 +21,9 @@
 from robusta.core.model.env_vars import (
     INCOMING_REQUEST_TIME_WINDOW_SECONDS,
     RUNNER_VERSION,
+    SENTRY_ENABLED,
     WEBSOCKET_PING_INTERVAL,
     WEBSOCKET_PING_TIMEOUT,
-    SENTRY_ENABLED,
 )
 from robusta.core.playbooks.playbook_utils import to_safe_str
 from robusta.core.playbooks.playbooks_event_handler import PlaybooksEventHandler
@@ -50,17 +49,30 @@ class ValidationResponse(BaseModel):
     error_msg: Optional[str] = None
 
 
+class SlackExternalActionRequest(ExternalActionRequest):
+    # Optional Slack Params
+    slack_username: Optional[str] = None
+    slack_message: Optional[Any] = None
+
+
 class SlackActionRequest(BaseModel):
-    value: ExternalActionRequest
+    value: SlackExternalActionRequest
 
     @validator("value", pre=True, always=True)
     def validate_value(cls, v: str) -> dict:
         # Slack value is sent as a stringified json, so we need to parse it before validation
         return json.loads(v)
 
 
+class SlackUserID(BaseModel):
+    username: str
+    name: str
+    team_id: str
+
+
 class SlackActionsMessage(BaseModel):
     actions: List[SlackActionRequest]
+    user: Optional[SlackUserID]
 
 
 class ActionRequestReceiver:
@@ -144,6 +156,13 @@ def __exec_external_request(self, action_request: ExternalActionRequest, validat
                 )
             return
 
+        # add global slack values to callback
+        if hasattr(action_request, 'slack_username'):
+            action_request.body.action_params["slack_username"] = action_request.slack_username
+
+        if hasattr(action_request, 'slack_message'):
+            action_request.body.action_params["slack_message"] = action_request.slack_message
+
         response = self.event_handler.run_external_action(
             action_request.body.action_name,
             action_request.body.action_params,
@@ -182,10 +201,19 @@ def _parse_websocket_message(
         message: Union[str, bytes, bytearray]
     ) -> Union[SlackActionsMessage, ExternalActionRequest]:
         try:
-            return SlackActionsMessage.parse_raw(message)  # this is slack callback format
+            return ActionRequestReceiver._parse_slack_message(message)  # this is slack callback format
         except ValidationError:
             return ExternalActionRequest.parse_raw(message)
 
+    @staticmethod
+    def _parse_slack_message(message: Union[str, bytes, bytearray]) -> SlackActionsMessage:
+        slack_actions_message = SlackActionsMessage.parse_raw(message)  # this is slack callback format
+        json_slack_message = json.loads(message)
+        for action in slack_actions_message.actions:
+            action.value.slack_username = slack_actions_message.user.username
+            action.value.slack_message = json_slack_message
+        return slack_actions_message
+
     def on_message(self, ws: websocket.WebSocketApp, message: str) -> None:
         """Callback for incoming websocket message from relay.