hook up broker connection failure

Author: crflynn

.gitignore

@@ -3,6 +3,7 @@
 
 # celery
 celerybeat-schedule
+celerybeat_schedule.db
 
 # pytest
 .cache

celery_slack/attachments.py

@@ -302,6 +302,30 @@ def get_beat_init_attachment(**kwargs):
     return attachment
 
 
+def get_broker_disconnection_attachment(**kwargs):
+    """Create the slack message attachment for broker disconnection."""
+    if kwargs["show_celery_hostname"]:
+        message = "*Celery could not connect to broker on {}.*".format(
+            socket.gethostname()
+        )
+    else:
+        message = "*Celery could not connect to broker.*"
+
+    attachment = {
+        "attachments": [
+            {
+                "fallback": message,
+                "color": kwargs["slack_broker_disconnect_color"],
+                "text": message,
+                "mrkdwn_in": ["text"]
+            }
+        ],
+        "text": ''
+    }
+
+    return attachment
+
+
 def schedule_to_string(schedule):
     """Transform a crontab, solar, or timedelta to a string."""
     if isinstance(schedule, crontab):

celery_slack/callbacks.py

@@ -1,7 +1,9 @@
 """Celery state and task callbacks."""
 from functools import wraps
+import time
 
 from .attachments import get_beat_init_attachment
+from .attachments import get_broker_disconnection_attachment
 from .attachments import get_celery_shutdown_attachment
 from .attachments import get_celery_startup_attachment
 from .attachments import get_task_failure_attachment
@@ -119,3 +121,48 @@ def slack_beat_init_callback(**kwargs):
         post_to_slack(cbkwargs["webhook"], ' ', attachment)
 
     return slack_beat_init_callback
+
+
+# Prevent spam
+BROKER_FAILURE_COOLDOWN = 60
+BROKER_FAILURE_TIME = time.time() - BROKER_FAILURE_COOLDOWN
+
+
+def slack_broker_connection_failure(**cbkwargs):
+    """Wrap the kombu.connection.retry_over_time callback callable."""
+    def slack_callback():
+        global BROKER_FAILURE_TIME
+        global BROKER_FAILURE_COOLDOWN
+
+        if time.time() - BROKER_FAILURE_TIME > BROKER_FAILURE_COOLDOWN:
+            BROKER_FAILURE_TIME = time.time()
+            attachment = get_broker_disconnection_attachment(**cbkwargs)
+            post_to_slack(cbkwargs["webhook"], ' ', attachment)
+
+    def wrapper(func):
+
+        @wraps(func)
+        def wrapped_func(fun, catch, args=[], kwargs={}, errback=None,
+                    max_retries=None, interval_start=2, interval_step=2,
+                    interval_max=30, callback=None):
+
+            def callback_wrapper(cb_func):
+                @wraps(cb_func)
+                def wrapped_cb_func(*args, **kwargs):
+                    slack_callback()
+                    return cb_func(*args, **kwargs)
+                return wrapped_cb_func
+
+            if callback is not None:
+                callback = callback_wrapper(callback)
+            else:
+                callback = slack_callback
+
+            func(fun=fun, catch=catch, args=args, kwargs=kwargs,
+                    errback=errback, max_retries=max_retries,
+                    interval_start=interval_start, interval_step=interval_step,
+                    interval_max=interval_max, callback=callback)
+
+        return wrapped_func
+
+    return wrapper

celery_slack/slackify.py

@@ -5,8 +5,10 @@
 from celery.signals import celeryd_init
 from celery.signals import task_prerun
 from celery.signals import worker_shutdown
+import kombu
 
 from .callbacks import slack_beat_init
+from .callbacks import slack_broker_connection_failure
 from .callbacks import slack_celery_shutdown
 from .callbacks import slack_celery_startup
 from .callbacks import slack_task_failure
@@ -24,6 +26,7 @@
     "slack_task_prerun_color": "#D3D3D3",
     "slack_task_success_color": "#36A64F",
     "slack_task_failure_color": "#D00001",
+    "slack_broker_disconnect_color": "#D00001",
     "flower_base_url": None,
     "show_task_id": True,
     "show_task_execution_time": True,
@@ -36,6 +39,7 @@
     "show_startup": True,
     "show_shutdown": True,
     "show_beat": True,
+    "show_broker": True,
     "use_fixed_width": True,
     "include_tasks": None,
     "exclude_tasks": None,
@@ -81,6 +85,7 @@ def __init__(self, app, webhook=None, beat_schedule=None, **options):
 
         self._connect_signals()
         self._decorate_task_methods()
+        self._decorate_kombu_retry()
 
     def _connect_signals(self):
         """Connect callbacks to celery signals.
@@ -119,3 +124,11 @@ def _decorate_task_methods(self):
                 slack_task_success(**self.options)(self.app.Task.on_success)
         self.app.Task.on_failure = \
             slack_task_failure(**self.options)(self.app.Task.on_failure)
+
+    def _decorate_kombu_retry(self):
+        """Decorate the kombu.connection.retry_over_time function."""
+        if self.options["show_broker"]:
+            kombu.connection.retry_over_time = \
+                slack_broker_connection_failure(**self.options)(
+                    kombu.connection.retry_over_time
+                )

tests/celeryapp/celeryapp.py

@@ -25,11 +25,12 @@
     "flower_base_url": "https://flower.example.com",
     "webhook": slack_webhook,
     "beat_schedule": schedule,
-    "show_task_prerun": True,
+    # "show_task_prerun": True,
     # "failures_only": True,
-    # "show_startup": False,
-    # "show_shutdown": False,
-    # "show_beat": False,
+    # "show_celery_hostname": True,
+    "show_startup": False,
+    "show_shutdown": False,
+    "show_beat": False,
 }
 
 # logging.info('Creating celery-slack object.')