Feat/add zulip sink (#1363)

* chore: check additional deps and resolve issues

* refactor: make channel_override generally usable

* feat: add new zulip sink/integration

* docs: add zulip sink setup

* docs: update steps to add a bot

* docs: update

* fix: correctly export channel transformer

* test: refactor channel_transformer tests

* chore: rm old slack tests targeted at the channel_transformer

* fix: update channel_transformer imports

* chore: add missing comments

* fix: dont validate channel_override on each template call, rather do it once on sinkparams creation. Also add tets....

* perf: only do work if the PREF is actually set

---------

Co-authored-by: Oscar Guertler <[email protected]>

docs/configuration/sinks/zulip.rst

@@ -0,0 +1,57 @@
+Zulip
+######
+
+Robusta can report issues and events in your Kubernetes cluster to Zulip.
+
+.. image:: /images/zulip_example.png
+    :width: 1000
+    :align: center
+
+To configure the Zulip sink you will need a *bot email*, an *api token* and your *api/zulip url*
+
+Creating a bot account
+-----------------------
+
+1. Open Zulip
+2. Click on the gear icon in the upper right corner
+3. Select **Personal** or **Organization** settings
+4. On the left, click **Bots**
+5. Click **Add a new bot**
+6. Fill out the fields, and click **Add**
+4. Copy email and token
+
+Settings
+------------------
+
+* ``api_url`` : The url of your Zulip instance
+* ``bot_email`` : The email of the bot account
+* ``bot_api_key`` : The api key of your bot account
+* ``stream_name`` : Name of the channel to send the message to
+* ``topic_name`` : Name of the topic of the stream to send messages to
+* ``topic_override`` : Dynamic topic override, same as the channel_override in the slack sink
+* ``log_preview_char_limit`` : [Optional - default: ``500``] The amount of log characters to append to the alert message (zulip doesnt have a builtin text file preview). If set to ``0`` a text file will be sent
+
+Configuring the Zulip sink
+---------------------------
+
+.. admonition:: Add this to your generated_values.yaml
+
+   .. code-block:: yaml
+
+        sinksConfig:
+        - zulip_sink:
+            name: my_zulip_sink
+            api_url: https://my-zulip-instance.com
+            bot_email: [email protected]
+            bot_api_key: very_secret_key
+            stream_name: Monitoring
+            topic_name: Robusta
+
+Save the file and run
+
+.. code-block:: bash
+   :name: cb-add-zulip-sink
+
+   helm upgrade robusta robusta/robusta -f generated_values.yaml
+
+You should now get alerts in Zulip!

run_runner_locally.sh

@@ -27,15 +27,22 @@ export NC='\033[0m' # No Color
 # Check if Python 3.9 or higher is installed
 if ! $PYTHON_BINARY -c "import sys; exit(not (sys.version_info.major == 3 and sys.version_info.minor >= 9))"; then
   echo -e "${RED}Error: Python 3.9 or higher is not installed or was not run\n${NC}"
-  echo "You are using `$PYTHON_BINARY --version` located at `which $PYTHON_BINARY`"
+  echo "You are using $(PYTHON_BINARY --version) located at $(which $PYTHON_BINARY)"
   echo "To change your Python version, edit the PYTHON_BINARY variable at the top of this script"
   exit 1
 fi
 
+# Check if mirrord is installed globally
+if ! command -v mirrord &> /dev/null
+then
+    echo -e "${RED}Mirrord is not installed globally. Follow the guide here to install it: https://github.com/metalbear-co/mirrord?tab=readme-ov-file#cli-tool${NC}"
+    exit 1
+fi
+
 # Check if Poetry is installed globally
 if ! command -v poetry &> /dev/null
 then
-    echo -e "${RED}Poetry is not installed globally. Make sure the `poetry` command works${NC}"
+    echo -e "${RED}Poetry is not installed globally. Make sure the 'poetry' command works${NC}"
     exit 1
 fi
 
@@ -50,8 +57,8 @@ fi
 
 echo "Setting up local runner environment"
 mkdir -p deployment/playbooks/defaults
-ln -fs $(pwd)/playbooks/robusta_playbooks/ ./deployment/playbooks/defaults
-ln -fs $(pwd)/playbooks/pyproject.toml ./deployment/playbooks/defaults
+ln -fs "$(pwd)/playbooks/robusta_playbooks/" ./deployment/playbooks/defaults
+ln -fs "$(pwd)/playbooks/pyproject.toml" ./deployment/playbooks/defaults
 
 echo "Checking if runner can listen on port ${PORT}"
 if lsof -Pi :${PORT} -sTCP:LISTEN -t >/dev/null ; then

src/robusta/core/model/runner_config.py

@@ -25,6 +25,7 @@
 from robusta.core.sinks.webhook.webhook_sink_params import WebhookSinkConfigWrapper
 from robusta.core.sinks.yamessenger.yamessenger_sink_params import YaMessengerSinkConfigWrapper
 from robusta.core.sinks.pushover.pushover_sink_params import PushoverSinkConfigWrapper
+from robusta.core.sinks.zulip.zulip_sink_params import ZulipSinkConfigWrapper
 from robusta.model.alert_relabel_config import AlertRelabel
 from robusta.model.playbook_definition import PlaybookDefinition
 from robusta.utils.base64_utils import is_base64_encoded
@@ -63,6 +64,7 @@ class RunnerConfig(BaseModel):
                 PushoverSinkConfigWrapper,
                 GoogleChatSinkConfigWrapper,
                 ServiceNowSinkConfigWrapper,
+                ZulipSinkConfigWrapper
             ]
         ]
     ]

src/robusta/core/reporting/blocks.py

@@ -427,13 +427,13 @@ def to_markdown(self, max_chars=None, add_table_header: bool = True) -> Markdown
 
         return MarkdownBlock(f"{prefix}{table_contents}{suffix}")
 
-    def to_table_string(self, table_max_width: int = PRINTED_TABLE_MAX_WIDTH) -> str:
+    def to_table_string(self, table_max_width: int = PRINTED_TABLE_MAX_WIDTH, table_fmt: str = "presto") -> str:
         rendered_rows = self.__to_strings_rows(self.render_rows())
         col_max_width = self.__calc_max_width(self.headers, rendered_rows, table_max_width)
         return tabulate(
             rendered_rows,
             headers=self.headers,
-            tablefmt="presto",
+            tablefmt=table_fmt,
             maxcolwidths=col_max_width,
         )
 

src/robusta/core/sinks/common/__init__.py

@@ -0,0 +1 @@
+from robusta.core.sinks.common.channel_transformer import ChannelTransformer

src/robusta/core/sinks/common/channel_transformer.py

@@ -0,0 +1,106 @@
+from collections import defaultdict
+from string import Template
+from typing import Dict, Optional, Union
+
+import regex
+
+CLUSTER_PREF = "cluster_name"
+CLUSTER_PREF_PATTERN = regex.compile("(\$?\{?" + CLUSTER_PREF + "\}?)")  # noqa: W605
+LABELS_PREF = "labels."
+ESCAPED_LABEL_PREF = regex.escape(LABELS_PREF)
+LABEL_PREF_PATTERN = regex.compile("\$?" + ESCAPED_LABEL_PREF + "[\w.]+")  # noqa: W605
+ANNOTATIONS_PREF = "annotations."
+ESCAPED_ANNOTATIONS_PREF = regex.escape(ANNOTATIONS_PREF)
+ANNOTATIONS_PREF_PATTERN = regex.compile("\$?" + ESCAPED_ANNOTATIONS_PREF + "[\w.]+")  # noqa: W605
+BRACKETS_PATTERN = regex.compile(r"\$\{[^\}]+\}")
+COMPOSITE_PATTERN = r".*\$({?labels.[^$]+|{?annotations.[^$]+|{?cluster_name).*"
+ONLY_VALUE_PATTERN = r"^(labels.[^$]+|annotations.[^$]+|cluster_name)$"
+MISSING = "<missing>"
+
+
+class ChannelTransformer:
+    @classmethod
+    def validate_channel_override(cls, v: Union[str, None]):
+        if v:
+            if regex.match(ONLY_VALUE_PATTERN, v):
+                return "$" + v
+            if not regex.match(COMPOSITE_PATTERN, v):
+                err_msg = (
+                    f"channel_override must be '{CLUSTER_PREF}' or '{LABELS_PREF}foo' or '{ANNOTATIONS_PREF}foo' "
+                    f"or contain patters like: '${CLUSTER_PREF}'/'${LABELS_PREF}foo'/"
+                    f"'${ANNOTATIONS_PREF}foo'"
+                )
+                raise ValueError(err_msg)
+        return v
+
+    @classmethod
+    def normalize_key_string(cls, s: str) -> str:
+        return s.replace("/", "_").replace(".", "_").replace("-", "_")
+
+    @classmethod
+    def normalize_dict_keys(cls, metadata: Dict) -> Dict:
+        result = defaultdict(lambda: MISSING)
+        result.update({cls.normalize_key_string(k): v for k, v in metadata.items()})
+        return result
+
+    # if prefix not present, return ""
+    # else, if found, return replacement else return MISSING
+    @classmethod
+    def get_replacement(cls, prefix: str, value: str, normalized_replacements: Dict) -> str:
+        if prefix in value:  # value is in the format of "$prefix" or "prefix"
+            value = cls.normalize_key_string(value.replace(prefix, ""))
+            if "$" in value:
+                return Template(value).safe_substitute(normalized_replacements)
+            else:
+                return normalized_replacements[value]
+        return ""
+
+    @classmethod
+    def replace_token(
+        cls,
+        pattern: regex.Pattern,
+        prefix: str,
+        channel: str,
+        replacements: Dict[str, str],
+    ) -> str:
+        tokens = pattern.findall(channel)
+        for token in tokens:
+            clean_token = token.replace("{", "").replace("}", "")
+            replacement = cls.get_replacement(prefix, clean_token, replacements)
+            if replacement:
+                channel = channel.replace(token, replacement)
+        return channel
+
+    @classmethod
+    def template(
+        cls,
+        channel_override: Optional[str],
+        default_channel: str,
+        cluster_name: str,
+        labels: Dict[str, str],
+        annotations: Dict[str, str],
+    ) -> str:
+        if not channel_override:
+            return default_channel
+
+        channel = channel_override
+        if CLUSTER_PREF in channel:
+            # replace "cluster_name" or "$cluster_name" or ${cluster_name} with the value of the cluster name
+            channel = CLUSTER_PREF_PATTERN.sub(cluster_name, channel)
+
+        if LABELS_PREF in channel:
+            normalized_labels = cls.normalize_dict_keys(labels)
+            channel = cls.replace_token(BRACKETS_PATTERN, LABELS_PREF, channel, normalized_labels)
+            channel = cls.replace_token(LABEL_PREF_PATTERN, LABELS_PREF, channel, normalized_labels)
+
+        if ANNOTATIONS_PREF in channel:
+            normalized_annotations = cls.normalize_dict_keys(annotations)
+            channel = cls.replace_token(BRACKETS_PATTERN, ANNOTATIONS_PREF, channel, normalized_annotations)
+            channel = cls.replace_token(
+                ANNOTATIONS_PREF_PATTERN,
+                ANNOTATIONS_PREF,
+                channel,
+                normalized_annotations,
+            )
+
+        return channel if MISSING not in channel else default_channel

src/robusta/core/sinks/sink_factory.py

@@ -4,8 +4,8 @@
 from robusta.core.sinks.discord import DiscordSink, DiscordSinkConfigWrapper
 from robusta.core.sinks.file.file_sink import FileSink
 from robusta.core.sinks.file.file_sink_params import FileSinkConfigWrapper
-from robusta.core.sinks.google_chat.google_chat_params import GoogleChatSinkConfigWrapper
 from robusta.core.sinks.google_chat.google_chat import GoogleChatSink
+from robusta.core.sinks.google_chat.google_chat_params import GoogleChatSinkConfigWrapper
 from robusta.core.sinks.jira import JiraSink, JiraSinkConfigWrapper
 from robusta.core.sinks.kafka import KafkaSink, KafkaSinkConfigWrapper
 from robusta.core.sinks.mail.mail_sink import MailSink
@@ -28,6 +28,7 @@
 from robusta.core.sinks.webhook import WebhookSink, WebhookSinkConfigWrapper
 from robusta.core.sinks.yamessenger import YaMessengerSink, YaMessengerSinkConfigWrapper
 from robusta.core.sinks.pushover import PushoverSink, PushoverSinkConfigWrapper
+from robusta.core.sinks.zulip import ZulipSink, ZulipSinkConfigWrapper
 
 class SinkFactory:
     __sink_config_mapping: Dict[Type[SinkConfigBase], Type[SinkBase]] = {
@@ -52,6 +53,7 @@ class SinkFactory:
         PushoverSinkConfigWrapper: PushoverSink,
         GoogleChatSinkConfigWrapper: GoogleChatSink,
         ServiceNowSinkConfigWrapper: ServiceNowSink,
+        ZulipSinkConfigWrapper: ZulipSink
     }
 
     @classmethod

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

@@ -1,25 +1,9 @@
-from collections import defaultdict
-from string import Template
-from typing import Dict, Optional
-
-import regex
-from pydantic import validator
-
 from robusta.core.sinks.sink_base_params import SinkBaseParams
 from robusta.core.sinks.sink_config import SinkConfigBase
+from robusta.core.sinks.common import ChannelTransformer
 
-CLUSTER_PREF = "cluster_name"
-CLUSTER_PREF_PATTERN = regex.compile("(\$?\{?" + CLUSTER_PREF + "\}?)")  # noqa: W605
-LABELS_PREF = "labels."
-ESCAPED_LABEL_PREF = regex.escape(LABELS_PREF)
-LABEL_PREF_PATTERN = regex.compile("\$?" + ESCAPED_LABEL_PREF + "[\w.]+")  # noqa: W605
-ANNOTATIONS_PREF = "annotations."
-ESCAPED_ANNOTATIONS_PREF = regex.escape(ANNOTATIONS_PREF)
-ANNOTATIONS_PREF_PATTERN = regex.compile("\$?" + ESCAPED_ANNOTATIONS_PREF + "[\w.]+")  # noqa: W605
-BRACKETS_PATTERN = regex.compile(r"\$\{[^\}]+\}")
-COMPOSITE_PATTERN = r".*\$({?labels.[^$]+|{?annotations.[^$]+|{?cluster_name).*"
-ONLY_VALUE_PATTERN = r"^(labels.[^$]+|annotations.[^$]+|cluster_name)$"
-MISSING = "<missing>"
+from typing import Optional
+from pydantic import validator
 
 
 class SlackSinkParams(SinkBaseParams):
@@ -30,83 +14,7 @@ class SlackSinkParams(SinkBaseParams):
 
     @validator("channel_override")
     def validate_channel_override(cls, v: str):
-        if v:
-            if regex.match(ONLY_VALUE_PATTERN, v):
-                return "$" + v
-            if not regex.match(COMPOSITE_PATTERN, v):
-                err_msg = (
-                    f"channel_override must be '{CLUSTER_PREF}' or '{LABELS_PREF}foo' or '{ANNOTATIONS_PREF}foo' "
-                    f"or contain patters like: '${CLUSTER_PREF}'/'${LABELS_PREF}foo'/"
-                    f"'${ANNOTATIONS_PREF}foo'"
-                )
-                raise ValueError(err_msg)
-        return v
-
-    def normalize_key_string(cls, s: str) -> str:
-        return s.replace("/", "_").replace(".", "_").replace("-", "_")
-
-    def normalize_dict_keys(cls, metadata: Dict) -> Dict:
-        result = defaultdict(lambda: MISSING)
-        result.update({cls.normalize_key_string(k): v for k, v in metadata.items()})
-        return result
-
-    # if prefix not present, return ""
-    # else, if found, return replacement else return MISSING
-    def get_replacement(self, prefix: str, value: str, normalized_replacements: Dict) -> str:
-        if prefix in value:  # value is in the format of "$prefix" or "prefix"
-            value = self.normalize_key_string(value.replace(prefix, ""))
-            if "$" in value:
-                return Template(value).safe_substitute(normalized_replacements)
-            else:
-                return normalized_replacements[value]
-        return ""
-
-    def get_slack_channel(self, cluster_name: str, labels: Dict, annotations: Dict) -> str:
-        if self.channel_override:
-            channel = self.channel_override
-            if CLUSTER_PREF in channel:
-                # replace "cluster_name" or "$cluster_name" or ${cluster_name} with the value of the cluster name
-                channel = CLUSTER_PREF_PATTERN.sub(cluster_name, channel)
-
-            if LABELS_PREF in channel or ANNOTATIONS_PREF in channel:
-                normalized_labels = self.normalize_dict_keys(labels)
-                normalized_annotations = self.normalize_dict_keys(annotations)
-
-                # # replace anything from the format of "${annotations.kubernetes.io/service-name}"
-                curly_brackets_tokens = BRACKETS_PATTERN.findall(channel)
-                for token in curly_brackets_tokens:
-                    clean_token = token.replace("{", "").replace("}", "")
-                    # labels
-                    replacement = self.get_replacement(LABELS_PREF, clean_token, normalized_labels)
-                    if replacement:
-                        channel = channel.replace(token, replacement)
-
-                    # annotations
-                    replacement = self.get_replacement(ANNOTATIONS_PREF, clean_token, normalized_annotations)
-                    if replacement:
-                        channel = channel.replace(token, replacement)
-
-                # labels replace: anything from the format of "labels.xyz" or "$labels.xyz"
-                if LABELS_PREF in channel:
-                    labels_tokens = LABEL_PREF_PATTERN.findall(channel)
-                    for label_token in labels_tokens:
-                        replacement = self.get_replacement(LABELS_PREF, label_token, normalized_labels)
-                        if replacement:
-                            channel = channel.replace(label_token, replacement)
-
-                # annotations replace: anything from the format of "annotations.xyz" or "$annotations.xyz"
-                if ANNOTATIONS_PREF in channel:
-                    annotations_tokens = ANNOTATIONS_PREF_PATTERN.findall(channel)
-                    for annotation_token in annotations_tokens:
-                        replacement = self.get_replacement(ANNOTATIONS_PREF, annotation_token, normalized_annotations)
-                        if replacement:
-                            channel = channel.replace(annotation_token, replacement)
-
-            if MISSING not in channel:
-                return channel
-
-        # Return default slack_channel if no channel_override
-        return self.slack_channel
+        return ChannelTransformer.validate_channel_override(v)
 
 
 class SlackSinkConfigWrapper(SinkConfigBase):

src/robusta/core/sinks/zulip/__init__.py

@@ -0,0 +1,2 @@
+from robusta.core.sinks.zulip.zulip_sink import ZulipSink
+from robusta.core.sinks.zulip.zulip_sink_params import ZulipSinkConfigWrapper, ZulipSinkParams