Сельдерей с MongoDB Брокер

Question

Работа над настройкой Celery (в соответствии с базовым руководством) с брокером mongodb в качестве бэкэнда. В соответствии с рекомендациями по настройке, изложенными в официальных документах, мой celeryconfig.py настроен следующим образом:

CELERY_RESULT_BACKEND = "mongodb"

BROKER_BACKEND = "mongodb"
BROKER_URL = "mongodb://user:pass@subdomain.mongolab.com:123456/testdb"

CELERY_MONGODB_BACKEND_SETTINGS = {
    "host":"subdomain.mongolab.com",
    "port":123456,
    "database":"testdb",
    "taskmeta_collection":"taskmeta",
    "user":"user",
    "pass":"pass",
}
CELERY_IMPORTS = ("tasks",)

Запуск сельдерея с --loglevel=INFO возвращает следующее исключение, происходящее из пимонго, но кипящее как через комбу, так и из сельдерея.

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/__init__.py", line 230, in start
    component.start()
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py", line 338, in start
    self.reset_connection()
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py", line 596, in reset_connection
    on_decode_error=self.on_decode_error)
  File "/usr/local/lib/python2.7/dist-packages/celery/app/amqp.py", line 335, in get_task_consumer
    **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/kombu/compat.py", line 187, in __init__
    super(ConsumerSet, self).__init__(self.backend, queues, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/kombu/messaging.py", line 285, in __init__
    self.declare()
  File "/usr/local/lib/python2.7/dist-packages/kombu/messaging.py", line 295, in declare
    queue.declare()
  File "/usr/local/lib/python2.7/dist-packages/kombu/entity.py", line 388, in declare
    self.queue_declare(nowait, passive=False)
  File "/usr/local/lib/python2.7/dist-packages/kombu/entity.py", line 408, in queue_declare
    nowait=nowait)
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/virtual/__init__.py", line 380, in queue_declare
    return queue, self._size(queue), 0
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/mongodb.py", line 74, in _size
    return self.client.messages.find({"queue": queue}).count()
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/mongodb.py", line 171, in client
    self._client = self._open()
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/mongodb.py", line 97, in _open
    mongoconn = Connection(host=conninfo.hostname, port=conninfo.port)
  File "/usr/local/lib/python2.7/dist-packages/pymongo/connection.py", line 325, in __init__
    nodes.update(uri_parser.split_hosts(entity, port))
  File "/usr/local/lib/python2.7/dist-packages/pymongo/uri_parser.py", line 198, in split_hosts
    nodes.append(parse_host(entity, default_port))
  File "/usr/local/lib/python2.7/dist-packages/pymongo/uri_parser.py", line 127, in parse_host
    raise ConfigurationError("Reserved characters such as ':' must be "
ConfigurationError: Reserved characters such as ':' must be escaped according RFC 2396. An IPv6 address literal must be enclosed in '[' and ']' according to RFC 2732.

Что-то в том, как Селери обрабатывает монгури, неправильно кодирует, так как это * URI-синтаксический анализатор в pymongo вызывает эту ошибку. Я попытался экранировать символы : в строке uri, но все, что удалось сделать, это сбросить транспорт обратно к AMQP по умолчанию с помощью строки искаженного соединения.

amqp://guest@localhost:5672/mongodb\http://user\:password@subdomain.mongolab.com\:29217/testdb

Что явно не так.

Я попытался ввести uri в конфигурации в виде необработанной строки, используя r, и ничего не изменилось.

Я знаю, что этот тип конфигурации соединения поддерживается в Celery начиная с версии 2.4 (я использую 2.5.1, pymongo 2.1.1), и все официальные документы ссылаются на нее как на предпочтительный способ подключения к брокеру mongodb.

Может ли это быть ошибкой, возможно, несовместимостью с последней сборкой pymongo? Если этот подход не работает, как бы прикрепить очередь задач к набору реплик, поскольку я предполагаю, что они должны быть переданы в монгури с использованием параметра ?replicaSet.

Я должен отметить, что я бы предпочел не переключаться на использование брокера RabbitMQ, поскольку Mongo уже находится в стеке для рассматриваемого приложения и просто кажется более интуитивно понятным использовать то, что уже есть. Если есть конкретная причина, по которой Mongo будет менее эффективен для этой цели (количество заданий в день будет относительно небольшим), я бы хотел знать! Заранее спасибо.

Answer 1

Я думаю, что это ошибка.Сельдерей передал имя хоста вместо server_uri в kombu, поэтому возникает эта проблема.После отслеживания кода я обнаружил следующий файл conf, чтобы обойти ошибку, прежде чем она исправит ее.

CELERY_RESULT_BACKEND = 'mongodb'
BROKER_HOST = "subdomain.mongolab.com"
BROKER_PORT = 123456
BROKER_TRANSPORT = 'mongodb'
BROKER_VHOST = 'testdb'
CELERY_IMPORTS = ('tasks',)
CELERY_MONGODB_BACKEND_SETTINGS = { 
       'host': 'subdomain.mongolab.com',
       'port': 123456,
       'database': 'testdb',
       'user': user,
       'password': password,
       'taskmeta_collection': 'teskmeta'
       }   

, просто повторяя настройку.

Answer 2

Поможет ли это, если вы удалите слова «пользователь», «пароль», «порт» и «база данных» из запроса CELERY_MONGODB_BACKEND_SETTINGS и выполните:

BROKER_URL = "mongodb://user:pass@subdomain.mongolab.com:123456/testdb"

CELERY_MONGODB_BACKEND_SETTINGS = {
    "host":BROKER_URL,
    "taskmeta_collection":"taskmeta",
}