pre-testing

Author: Avi-Robusta

docs/configuration/sinks/Opsgenie.rst

@@ -14,7 +14,9 @@ We can add Prometheus alert labels into Opsgenie alert extra details by setting
 Configuring the OpsGenie sink
 ------------------------------------------------
 
-.. admonition:: Add this to your generated_values.yaml
+The OpsGenie sink supports static and dynamic team routing, optional fallback teams, tagging, and enrichment with Kubernetes labels.
+
+.. admonition:: Add this to your `generated_values.yaml`
 
     .. code-block:: yaml
 
@@ -23,13 +25,22 @@ Configuring the OpsGenie sink
             name: ops_genie_sink
             api_key: OpsGenie integration API key  # configured from OpsGenie team integration
             teams:
-            - "noc"
-            - "sre"
+            - "noc"                  # Static team
+            - "$labels.team"         # Dynamic team routing based on alert label
+            default_team: "oncall"   # Optional fallback team for Dynamic team routing
             tags:
             - "prod a"
             extra_details_labels: false # optional, default is false
 
-Save the file and run
+In this example:
+
+- Alerts will be routed to the **"noc"** team by default.
+- If the alert includes a **"team" label**, it will also be routed to the value of that label.
+- If the **"team" label is missing**, the alert will be routed to the **"oncall"** team as a fallback.
+- The tag **"prod a"** will be included with every alert.
+- Kubernetes labels **will not be added** to alert details, as `extra_details_labels` is set to `false`.
+
+Save the file and apply the configuration:
 
 .. code-block:: bash
    :name: cb-add-opsgenie-sink

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

@@ -4,6 +4,7 @@
 import opsgenie_sdk
 
 from robusta.core.reporting.base import Enrichment, Finding, FindingSeverity
+from robusta.core.sinks.common.channel_transformer import ChannelTransformer
 from robusta.core.sinks.opsgenie.opsgenie_sink_params import OpsGenieSinkConfigWrapper
 from robusta.core.sinks.sink_base import SinkBase
 from robusta.core.sinks.transformer import Transformer
@@ -22,9 +23,18 @@ def __init__(self, sink_config: OpsGenieSinkConfigWrapper, registry):
 
         self.api_key = sink_config.opsgenie_sink.api_key
         self.teams = sink_config.opsgenie_sink.teams
+        self.default_team = sink_config.opsgenie_sink.default_team
         self.tags = sink_config.opsgenie_sink.tags
         self.extra_details_labels = sink_config.opsgenie_sink.extra_details_labels
 
+        # Check for dangerous configuration
+        has_templates = any("$" in team for team in self.teams)
+        if has_templates and not self.default_team:
+            logging.warning(
+                "OpsGenie sink is configured with templated team names but no default_team specified. "
+                "Alerts may fail to route if the required label or annotation is missing."
+            )
+
         opsgenie_sdk.configuration.Configuration.set_default(None)
         self.conf = opsgenie_sdk.configuration.Configuration()
         self.conf.api_key["Authorization"] = self.api_key
@@ -73,17 +83,48 @@ def __ack_alert(self, fingerprint: str, user: str, note: str):
         except opsgenie_sdk.ApiException as err:
             logging.error(f"Error acking opsGenie alert {fingerprint} {err}", exc_info=True)
 
+    def __get_teams(self, finding: Finding) -> List[str]:
+        teams = []
+        for team_template in self.teams:
+            try:
+                if "$" in team_template:
+                    # Only process templates that contain variables
+                    team = ChannelTransformer.template(
+                        team_template,
+                        self.default_team,  # Use default_team as fallback
+                        self.cluster_name,
+                        finding.subject.labels,
+                        finding.subject.annotations,
+                    )
+                    if team:  # Only add non-null teams
+                        teams.append(team)
+                else:
+                    # Use static team name directly
+                    teams.append(team_template)
+            except Exception as e:
+                logging.warning(
+                    f"Failed to process team template {team_template} for alert subject {finding.service_key}: {e}"
+                )
+                continue
+        return teams
+
     def __open_alert(self, finding: Finding, platform_enabled: bool):
         description = self.__to_description(finding, platform_enabled)
         details = self.__to_details(finding)
         tags = self.tags.copy()
         tags.insert(0, self.cluster_name)
+
+        # Get teams based on templates
+        teams = self.__get_teams(finding)
+        if not teams and self.teams:  # dynamic routing failed and no default team configured
+            logging.warning(f"No valid teams resolved for finding {finding.title}. Alert may not be routed properly.")
+
         body = opsgenie_sdk.CreateAlertPayload(
             source="Robusta",
             message=finding.title,
             description=description,
             alias=finding.fingerprint,
-            responders=[{"name": team, "type": "team"} for team in self.teams],
+            responders=[{"name": team, "type": "team"} for team in teams],
             details=details,
             tags=tags,
             entity=finding.service_key,

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

@@ -7,6 +7,7 @@
 class OpsGenieSinkParams(SinkBaseParams):
     api_key: str
     teams: List[str] = []
+    default_team: Optional[str] = None
     tags: List[str] = []
     host: Optional[str] = None  # NOTE: If None, the default value will be used from opsgenie_sdk
     extra_details_labels: Optional[bool] = False