MAIN-2798 - Opsgenie 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,109 @@
+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):
+    """
+        Add a button to the alert - clicking it will ask chat gpt to help find a solution.
+    """
+    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,18 @@ 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
+        for action in slack_actions_message.actions:
+            action.value.slack_username = slack_actions_message.user.username
+            action.value.slack_message = json.loads(message)
+        return slack_actions_message
+
     def on_message(self, ws: websocket.WebSocketApp, message: str) -> None:
         """Callback for incoming websocket message from relay.
 

src/robusta/integrations/slack/sender.py

@@ -15,8 +15,8 @@
 from robusta.core.model.base_params import AIInvestigateParams, ResourceInfo
 from robusta.core.model.env_vars import (
     ADDITIONAL_CERTIFICATE,
-    SLACK_REQUEST_TIMEOUT,
     HOLMES_ENABLED,
+    SLACK_REQUEST_TIMEOUT,
     SLACK_TABLE_COLUMNS_LIMIT,
 )
 from robusta.core.playbooks.internal.ai_integration import ask_holmes
@@ -695,3 +695,33 @@ def send_or_update_summary_message(
             return resp["ts"]
         except Exception as e:
             logging.exception(f"error sending message to slack\n{e}\nchannel={channel}\n")
+
+    def update_slack_message(self, channel: str, ts: str, blocks: list, text: str = ""):
+        """
+        Update a Slack message with new blocks and optional text.
+
+        Args:
+            channel (str): Slack channel ID.
+            ts (str): Timestamp of the message to update.
+            blocks (list): List of Slack Block Kit blocks for the updated message.
+            text (str, optional): Plain text summary for accessibility. Defaults to "".
+        """
+        try:
+            # Ensure channel ID exists in the mapping
+            if channel not in self.channel_name_to_id.values():
+                logging.error(f"Channel ID for {channel} could not be determined. Update aborted.")
+                return
+
+            # Call Slack's chat_update method
+            resp = self.slack_client.chat_update(
+                channel=channel,
+                ts=ts,
+                text=text,
+                blocks=blocks
+            )
+            logging.debug(f"Message updated successfully: {resp['ts']}")
+            return resp["ts"]
+
+        except Exception as e:
+            logging.exception(f"Error updating Slack message: {e}")
+            return None