Investigate no comms in docker

Dockerfile

@@ -29,4 +29,4 @@ USER flower
 
 VOLUME $FLOWER_DATA_DIR
 
-ENTRYPOINT ["flower"]
+CMD ["celery flower"]

docker-compose.yml

@@ -31,7 +31,7 @@ services:
       - redis
   flower:
     build: ./
-    command: python -m flower -A tasks
+    command: celery -A tasks flower
     volumes:
       - ./examples:/data
     working_dir: /data
@@ -40,3 +40,6 @@ services:
     environment:
       CELERY_BROKER_URL: redis://redis
       CELERY_RESULT_BACKEND: redis://redis
+    depends_on:
+      - worker
+      - redis

docs/prometheus-integration.rst

@@ -142,7 +142,7 @@ Start Flower Monitoring
 
 In your Celery application folder run this command (Flower needs to be installed)::
 
-    celery flower -A tasks --broker=redis://localhost:6379/0
+    celery -A tasks --broker=redis://localhost:6379/0 flower
 
 Configure and Start Prometheus
 ------------------------------

flower/command.py

@@ -9,6 +9,8 @@
 from logging import NullHandler
 
 import click
+from celery.utils.time import humanize_seconds
+from kombu.exceptions import OperationalError
 from tornado.options import options
 from tornado.options import parse_command_line, parse_config_file
 from tornado.log import enable_pretty_logging
@@ -48,7 +50,12 @@ def sigterm_handler(signal, frame):
         sys.exit(0)
 
     signal.signal(signal.SIGTERM, sigterm_handler)
+
+    if not is_broker_connected(celery_app=app):
+        return
+
     print_banner(app, 'ssl_options' in settings)
+
     try:
         flower.start()
     except (KeyboardInterrupt, SystemExit):
@@ -103,6 +110,33 @@ def warn_about_celery_args_used_in_flower_command(ctx, flower_args):
         )
 
 
+def is_broker_connected(celery_app):
+    is_connected = False
+    max_retries = celery_app.conf.broker_connection_max_retries
+
+    if not celery_app.conf.broker_connection_retry:
+        max_retries = 0
+
+    with celery_app.connection() as conn:
+        broker_url = conn.as_uri()
+
+        def _error_handler(exc, interval):
+            next_step = f"Trying again {humanize_seconds(interval, 'in', ' ')}... ({int(interval / 2)}/{max_retries})"
+            logger.error(f'Cannot connect to broker: {broker_url}. Error: {exc}. {next_step}')
+
+        try:
+            conn.ensure_connection(errback=_error_handler, max_retries=max_retries)
+            logger.info(f'Established connection to broker: {broker_url}. Starting Flower...')
+            is_connected = True
+        except OperationalError as e:
+            logger.error(
+                f'Unable to establish connection to broker: : {broker_url}. Error: {e}. '
+                f'Please make sure the broker is running when using Flower. Aborting Flower...'
+            )
+
+    return is_connected
+
+
 def setup_logging():
     if options.debug and options.logging == 'info':
         options.logging = 'debug'

tests/unit/test_command.py

@@ -3,11 +3,13 @@
 import tempfile
 import unittest
 import subprocess
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, create_autospec, MagicMock
 
 import mock
+from celery import Celery
+from kombu.exceptions import OperationalError
 
-from flower.command import apply_options, warn_about_celery_args_used_in_flower_command
+from flower.command import apply_options, warn_about_celery_args_used_in_flower_command, is_broker_connected
 from tornado.options import options
 from tests.unit import AsyncHTTPTestCase
 
@@ -77,6 +79,80 @@ class FakeContext:
         )
 
 
+class TestIsBrokerConnected(AsyncHTTPTestCase):
+    @patch('flower.command.logger.info')
+    def test_returns_true_and_logs_if_connection_to_broker_established(self, mock_info):
+        broker_url = 'broker_url'
+        broker_connection_max_retries = 2
+
+        mock_conf = Mock(broker_connection_retry=True, broker_connection_max_retries=broker_connection_max_retries)
+
+        mock_connection = MagicMock(name='mock connection')
+        mock_connection.as_uri.return_value = broker_url
+        mock_connection.__enter__.return_value = mock_connection
+
+        mock_celery_app = create_autospec(Celery, conf=mock_conf)
+        mock_celery_app.connection.return_value = mock_connection
+
+        assert is_broker_connected(celery_app=mock_celery_app)
+
+        mock_connection.ensure_connection.assert_called_once()
+        ensure_connection_kwargs = mock_connection.ensure_connection.call_args_list[0][1]
+        assert '_error_handler' in str(ensure_connection_kwargs['errback'])
+        assert ensure_connection_kwargs['max_retries'] == broker_connection_max_retries
+
+        mock_info.assert_called_once_with(f'Established connection to broker: {broker_url}. Starting Flower...')
+
+    @patch('flower.command.logger.error')
+    def test_returns_false_and_logs_error_if_connection_to_broker_cannot_be_established(self, mock_error):
+        broker_url = 'broker_url'
+        broker_connection_max_retries = 2
+
+        mock_conf = Mock(broker_connection_retry=True, broker_connection_max_retries=broker_connection_max_retries)
+
+        mock_connection = MagicMock(name='mock connection')
+        mock_connection.as_uri.return_value = broker_url
+        error = OperationalError('test error')
+        mock_connection.ensure_connection.side_effect = error
+        mock_connection.__enter__.return_value = mock_connection
+
+        mock_celery_app = create_autospec(Celery, conf=mock_conf)
+        mock_celery_app.connection.return_value = mock_connection
+
+        assert not is_broker_connected(celery_app=mock_celery_app)
+
+        mock_connection.ensure_connection.assert_called_once()
+        ensure_connection_kwargs = mock_connection.ensure_connection.call_args_list[0][1]
+        assert '_error_handler' in str(ensure_connection_kwargs['errback'])
+        assert ensure_connection_kwargs['max_retries'] == broker_connection_max_retries
+
+        mock_error.assert_called_once_with(
+            f'Unable to establish connection to broker: : {broker_url}. Error: {error}. '
+            f'Please make sure the broker is running when using Flower. Aborting Flower...'
+        )
+
+    @patch('flower.command.logger.error')
+    def test_disabled_broker_connection_retry_sets_max_retries_to_zero(self, mock_error):
+        broker_url = 'broker_url'
+        broker_connection_max_retries = 2
+
+        mock_conf = Mock(broker_connection_retry=False, broker_connection_max_retries=broker_connection_max_retries)
+
+        mock_connection = MagicMock(name='mock connection')
+        mock_connection.as_uri.return_value = broker_url
+        mock_connection.__enter__.return_value = mock_connection
+
+        mock_celery_app = create_autospec(Celery, conf=mock_conf)
+        mock_celery_app.connection.return_value = mock_connection
+
+        assert is_broker_connected(celery_app=mock_celery_app)
+
+        mock_connection.ensure_connection.assert_called_once()
+        ensure_connection_kwargs = mock_connection.ensure_connection.call_args_list[0][1]
+        assert '_error_handler' in str(ensure_connection_kwargs['errback'])
+        assert ensure_connection_kwargs['max_retries'] == 0
+
+
 class TestConfOption(AsyncHTTPTestCase):
     def test_error_conf(self):
         with self.mock_option('conf', None):