consumers validation

.github/workflows/ci.yml

@@ -0,0 +1,25 @@
+name: Lifeguard RabbitMQ CI
+
+on: [push]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.7]
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        make deps
+    - name: Test with nose2
+      run: |
+        make test

.github/workflows/publish.yml

@@ -0,0 +1,32 @@
+# This workflows will upload a Python Package using Twine when a release is created
+# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
+
+name: Lifeguard RabbitMQ Publish
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.x'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        make deps
+        pip install setuptools wheel twine
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        python setup.py sdist bdist_wheel
+        twine upload dist/*

lifeguard_rabbitmq/__init__.py

@@ -4,43 +4,14 @@
 
 from lifeguard.validations import validation
 
-from lifeguard_rabbitmq.validations import consumers_validation
+from lifeguard_rabbitmq.context import RABBITMQ_PLUGIN_CONTEXT
+from lifeguard_rabbitmq.validations import consumers_running_validation
 
 
-class RabbitMQContext:
-    """
-    RabbitMQ Context
-    """
-
-    def __init__(self):
-        self._consumers_validation_options = {
-            "actions": [],
-            "schedule": {"every": {"minutes": 1}},
-            "settings": {},
-        }
-
-    @property
-    def consumers_validation_options(self):
-        """
-        Getter for consumers validation options
-        """
-        return self._consumers_validation_options
-
-    @consumers_validation_options.setter
-    def consumers_validation_options(self, value):
-        """
-        Setter for consumers validation options
-        """
-        self._consumers_validation_options = value
-
-
-RABBITMQ_PLUGIN_CONTEXT = RabbitMQContext()
-
-
-def init(lifeguard_context):
+def init(_lifeguard_context):
     validation(
         "RabbitMQ Consumers Validation",
         RABBITMQ_PLUGIN_CONTEXT.consumers_validation_options["actions"],
         RABBITMQ_PLUGIN_CONTEXT.consumers_validation_options["schedule"],
         RABBITMQ_PLUGIN_CONTEXT.consumers_validation_options["settings"],
-    )(consumers_validation)
+    )(consumers_running_validation)

lifeguard_rabbitmq/context.py

@@ -0,0 +1,45 @@
+"""
+RabbitMQ Plugin Context
+"""
+
+
+class RabbitMQPluginContext:
+    """
+    RabbitMQ Context
+    """
+
+    def __init__(self):
+        self._consumers_validation_options = {
+            "actions": [],
+            "schedule": {"every": {"minutes": 1}},
+            "settings": {},
+            "queues": {},
+        }
+
+    @property
+    def consumers_validation_options(self):
+        """
+        Getter for consumers validation options
+        """
+        return self._consumers_validation_options
+
+    @consumers_validation_options.setter
+    def consumers_validation_options(self, value):
+        """
+        Setter for consumers validation options
+
+        Example:
+
+        {
+            "actions": [],
+            "schedule": {"every": {"minutes": 1}},
+            "settings": {},
+            "queues": {
+                "rabbitmq_admin_instance": [{"name": "queue_name", "min_number_of_consumers": 1}]
+            }
+        }
+        """
+        self._consumers_validation_options = value
+
+
+RABBITMQ_PLUGIN_CONTEXT = RabbitMQPluginContext()

lifeguard_rabbitmq/rabbitmq/admin.py

@@ -0,0 +1,40 @@
+"""
+RabbitMQ admin resources
+"""
+import json
+
+from lifeguard.http_client import get
+
+from lifeguard_rabbitmq.settings import get_rabbitmq_admin_instances
+
+BASE_URL = "{}"
+QUEUE = "/api/queues/{}/{}"
+
+
+def count_consumers(instance_name, queue):
+    """
+    Get consumers for a queue
+    """
+    instance_attributes = get_rabbitmq_admin_instances()[instance_name]
+    url = __url(QUEUE, instance_attributes["base_url"]).format(
+        __vhost(instance_attributes["vhost"]), queue
+    )
+    response = __get(url, instance_attributes["user"], instance_attributes["passwd"])
+
+    print("\n\n\n")
+    print(response)
+    print("\n\n\n")
+
+    return len(response["consumer_details"])
+
+
+def __get(url, user, password):
+    return json.loads(get(url, auth=(user, password)).content)
+
+
+def __url(api, admin):
+    return BASE_URL.format(admin) + api
+
+
+def __vhost(vhost):
+    return vhost.replace(vhost, "%2f")

lifeguard_rabbitmq/settings.py

@@ -5,13 +5,51 @@
 
 SETTINGS_MANAGER = SettingsManager(
     {
-        "LIFEGUARD_RABBITMQ_ADMIN_BASE_URL": {
+        "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_BASE_URL": {
             "default": "http://localhost:15672",
-            "description": "RabbitMQ admin base url",
-        }
+            "description": "RabbitMQ admin base url of default instance",
+        },
+        "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_USER": {
+            "default": "guest",
+            "description": "RabbitMQ admin user of default instance",
+        },
+        "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_PASSWD": {
+            "default": "guest",
+            "description": "RabbitMQ admin password of default instance",
+        },
+        "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_VHOST": {
+            "default": "/",
+            "description": "RabbitMQ admin virtual host of default instance",
+        },
     }
 )
 
-LIFEGUARD_RABBITMQ_ADMIN_BASE_URL = SETTINGS_MANAGER.read_value(
-    "LIFEGUARD_RABBITMQ_ADMIN_BASE_URL"
+LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_BASE_URL = SETTINGS_MANAGER.read_value(
+    "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_BASE_URL"
+)
+
+LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_USER = SETTINGS_MANAGER.read_value(
+    "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_USER"
 )
+
+LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_PASSWD = SETTINGS_MANAGER.read_value(
+    "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_PASSWD"
+)
+
+LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_VHOST = SETTINGS_MANAGER.read_value(
+    "LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_VHOST"
+)
+
+
+def get_rabbitmq_admin_instances():
+    """
+    Recover attributes of each RabbitMQ Admin instances
+    """
+    return {
+        "default": {
+            "base_url": LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_BASE_URL,
+            "user": LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_USER,
+            "passwd": LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_PASSWD,
+            "vhost": LIFEGUARD_RABBITMQ_DEFAULT_ADMIN_VHOST,
+        }
+    }

lifeguard_rabbitmq/validations.py

@@ -1,2 +1,59 @@
-def consumers_validation():
-    pass
+"""
+RabbitMQ Common Validations
+"""
+import traceback
+
+from lifeguard.logger import lifeguard_logger as logger
+from lifeguard import NORMAL, PROBLEM, change_status
+from lifeguard.validations import ValidationResponse
+
+from lifeguard_rabbitmq.context import RABBITMQ_PLUGIN_CONTEXT
+from lifeguard_rabbitmq.rabbitmq.admin import count_consumers
+
+
+def __check_consumers(rabbitmq_admin_instance, queues, details):
+    details[rabbitmq_admin_instance] = []
+    status = NORMAL
+
+    for queue in queues:
+        queue_status = {
+            "queue": queue["name"],
+            "status": NORMAL,
+        }
+        try:
+            queue_status["number_of_consumers"] = count_consumers(
+                rabbitmq_admin_instance, queue["name"]
+            )
+            if queue_status["number_of_consumers"] < queue["min_number_of_consumers"]:
+                queue_status["status"] = PROBLEM
+        except Exception as exception:
+            logger.error(
+                "error on recover queue infos %s",
+                str(exception),
+                extra={"traceback": traceback.format_exc()},
+            )
+            queue_status["status"] = PROBLEM
+            queue_status["error"] = "error on recover queue infos"
+
+        details[rabbitmq_admin_instance].append(queue_status)
+
+        status = change_status(status, queue_status["status"])
+
+    return status
+
+
+def consumers_running_validation():
+    """
+    Validates number of consumers for a queue
+    """
+    options = RABBITMQ_PLUGIN_CONTEXT.consumers_validation_options
+    status = NORMAL
+    details = {}
+
+    for rabbitmq_admin_instance in options["queues"]:
+        queues = options["queues"][rabbitmq_admin_instance]
+
+        status = change_status(
+            status, __check_consumers(rabbitmq_admin_instance, queues, details)
+        )
+    return ValidationResponse("consumers_running_validation", status, details)

requirements.txt

@@ -1 +1 @@
-lifeguard==0.0.15
+lifeguard==0.0.16

setup.py

@@ -10,7 +10,7 @@
     scripts=[],
     include_package_data=True,
     description="Lifeguard integration with RabbitMQ",
-    install_requires=["lifeguard==0.0.15"],
+    install_requires=["lifeguard"],
     classifiers=["Development Status :: 3 - Alpha"],
     packages=find_packages(),
 )

tests/rabbitmq/test_admin.py

@@ -0,0 +1,22 @@
+import unittest
+
+from unittest.mock import patch, MagicMock
+
+from lifeguard_rabbitmq.rabbitmq.admin import count_consumers
+
+
+class RabbitMQAdminTest(unittest.TestCase):
+    @patch("lifeguard_rabbitmq.rabbitmq.admin.get")
+    def test_count_consumers(self, mock_get):
+
+        mock_response = MagicMock(name="response")
+        mock_response.content = '{"consumer_details": []}'
+
+        mock_get.return_value = mock_response
+
+        self.assertEqual(count_consumers("default", "queue"), 0)
+
+        mock_get.assert_called_with(
+            "http://localhost:15672/api/queues/%2f/queue",
+            auth=("guest", "guest"),
+        )