Add remapper plugin

plugins/remapper/README.md

@@ -0,0 +1,58 @@
+Remapper Plugin
+===============
+
+This plugin will remap alert fields according to the mapping rules that you
+define via plugin configuration.
+
+Installation
+------------
+
+Clone the GitHub repo and run:
+
+    $ python setup.py install
+
+Or, to install remotely from GitHub run:
+
+    $ pip install git+https://github.com/alerta/alerta-contrib.git#subdirectory=plugins/remapper
+
+Note: If Alerta is installed in a python virtual environment then plugins
+need to be installed into the same environment for Alerta to dynamically
+discover them.
+
+Configuration
+-------------
+
+Add `remapper` to the list of enabled `PLUGINS` in `alertad.conf` server
+configuration file and set plugin-specific variables either in the
+server configuration file or as environment variables.
+
+```python
+PLUGINS = ['remapper']
+```
+
+Then configure the mapping rules as shown in the examples section below.
+
+**Example**
+
+```python
+PLUGINS = ['enhance', 'remapper']
+REMAPPER_MAPPING_RULES = "{ \"text\": \"service\", \"resource\": \"attributes.namespace\" }"
+```
+
+In the example above you can see the `REMAPPER_MAPPING_RULES` environment
+variable which is used to defines a dictionary in JSON string format. Such
+dictionary defines the following key-value pairs:
+
+* "text": "service"
+* "resource": "attributes.namespace"
+
+Each "key" will be target alert field, and each "value" will be the source
+alert field. The plugin will try to get the value of the source alert field
+in order to store it in the source alert field.
+So, in the example above, after the plugin has processed the alert, you
+should see the following outcome:
+
+* `alert.text` should store the value of `alert.service`
+* `alert.resource` should store the value of `alert.attributes['namespace']`
+
+Copyright (c) 2024 OJ. Available under the MIT License.

plugins/remapper/alerta_remapper.py

@@ -0,0 +1,56 @@
+import logging
+import os
+import json
+
+from alerta.plugins import PluginBase
+
+try:
+    from alerta.plugins import app  # alerta >= 5.0
+except ImportError:
+    from alerta.app import app  # alerta < 5.0
+
+LOG = logging.getLogger('alerta.plugins.remapper')
+
+# Try loading the mapping rules via environment variables first
+try:
+    # The environment variable must be a string formatted as follows:
+    #   REMAPPER_MAPPING_RULES="{ \"text\": \"service\", \"resource\": \"attributes.namespace\" }"
+    LOG.debug('Trying to load remapper config in JSON format')
+    REMAPPER_MAPPING_RULES = json.loads(os.environ.get('REMAPPER_MAPPING_RULES'))
+except Exception:
+    # Otherwise, we'll try grabbing the mapping rules from the application config
+    # which must be passed in dictionary format, for instance:
+    #   REMAPPER_MAPPING_RULES = {
+    #       "text": "service",
+    #       "resource": "attributes.namespace"
+    #   }
+    LOG.debug('Attempting to load remapper config via app config')
+    REMAPPER_MAPPING_RULES = app.config.get('REMAPPER_MAPPING_RULES', dict())
+
+
+class RemapAlert(PluginBase):
+
+    def pre_receive(self, alert):
+        LOG.debug('(pre_receive) Remapping alert fields...')
+
+        # Go through each mapping rule
+        for target, source in REMAPPER_MAPPING_RULES.items():
+            # Check if we're dealing with a nested dictionary property (only one level deep is supported)
+            if '.' in source:
+                dict_name, dict_key = source.split('.', 1)
+                if hasattr(alert, dict_name):
+                    nested_dict = getattr(alert, dict_name)
+                    if isinstance(nested_dict, dict) and dict_key in nested_dict:
+                        setattr(alert, target, nested_dict[dict_key])
+            else:
+                # Regular attribute remapping
+                if hasattr(alert, source):
+                    setattr(alert, target, getattr(alert, source))
+
+        return alert
+
+    def post_receive(self, alert):
+        return
+
+    def status_change(self, alert, status, text):
+        return

plugins/remapper/setup.py

@@ -0,0 +1,22 @@
+from setuptools import find_packages, setup
+
+version = '1.0.0'
+
+setup(
+    name='alerta-remapper',
+    version=version,
+    description='Alerta plugin for alert fields remapping',
+    url='https://github.com/alerta/alerta-contrib',
+    license='MIT',
+    author='OJ',
+    author_email='[email protected]',
+    packages=find_packages(),
+    py_modules=['alerta_remapper'],
+    include_package_data=True,
+    zip_safe=True,
+    entry_points={
+        'alerta.plugins': [
+            'remapper = alerta_remapper:RemapAlert'
+        ]
+    }
+)

plugins/remapper/test_remapper.py

@@ -0,0 +1,53 @@
+import unittest
+from unittest.mock import patch
+import os
+
+from alerta.models.alert import Alert
+from alerta_remapper import RemapAlert
+
+class RemapperPluginTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.alert = Alert(
+            resource='My Resource',
+            event='My Event',
+            environment='Development',
+            customer='My Customer',
+            severity='critical',
+            service='My Service',
+            group='My Group',
+            value='My Value',
+            text='My Text',
+            attributes={
+                'namespace': 'My Client'
+            },
+            timeout=3600
+        )
+
+    # This test case relies on the default mapping rules, so it expects no mapping to take place.
+    def test_empty_rules(self):
+        remapper = RemapAlert()
+        result = remapper.pre_receive(self.alert)
+        self.assertEqual(result.resource, self.alert.resource)
+        self.assertEqual(result.event, self.alert.event)
+        self.assertEqual(result.environment, self.alert.environment)
+
+    # IMPORTANT: we are actually patching the module variable, which skips the actual init logic
+    # that it's done for it. As a result, this approach works for testing the method that relies
+    # on said module variable, but fails to test the init logic associated to it.
+    @patch('alerta_remapper.REMAPPER_MAPPING_RULES', new={ "text": "customer", "service": "group" })
+    def test_simple_properties_mapping(self):
+        remapper = RemapAlert()
+        result = remapper.pre_receive(self.alert)
+        self.assertEqual(result.text, self.alert.customer)
+        self.assertEqual(result.service, self.alert.group)
+
+    # IMPORTANT: same comment as in the test case above.
+    @patch('alerta_remapper.REMAPPER_MAPPING_RULES', new={ "resource": "attributes.namespace" })
+    def test_nested_properties_mapping(self):
+        remapper = RemapAlert()
+        result = remapper.pre_receive(self.alert)
+        self.assertEqual(result.resource, self.alert.attributes['namespace'])
+
+if __name__ == "__main__":
+    unittest.main()