This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.0.
Source code for kombu.transport.consul
"""Consul Transport module for Kombu.
Features
========
It uses Consul.io's Key/Value store to transport messages in Queues
It uses python-consul for talking to Consul's HTTP API
Features
========
* Type: Native
* Supports Direct: Yes
* Supports Topic: *Unreviewed*
* Supports Fanout: *Unreviewed*
* Supports Priority: *Unreviewed*
* Supports TTL: *Unreviewed*
Connection String
=================
Connection string has the following format:
.. code-block::
consul://CONSUL_ADDRESS[:PORT]
"""
import socket
import uuid
from collections import defaultdict
from contextlib import contextmanager
from queue import Empty
from time import monotonic
from kombu.exceptions import ChannelError
from kombu.log import get_logger
from kombu.utils.json import dumps, loads
from kombu.utils.objects import cached_property
from . import virtual
try:
import consul
except ImportError:
consul = None
logger = get_logger('kombu.transport.consul')
DEFAULT_PORT = 8500
DEFAULT_HOST = 'localhost'
class LockError(Exception):
"""An error occurred while trying to acquire the lock."""
[docs]class Channel(virtual.Channel):
"""Consul Channel class which talks to the Consul Key/Value store."""
prefix = 'kombu'
index = None
timeout = '10s'
session_ttl = 30
def __init__(self, *args, **kwargs):
if consul is None:
raise ImportError('Missing python-consul library')
super().__init__(*args, **kwargs)
port = self.connection.client.port or self.connection.default_port
host = self.connection.client.hostname or DEFAULT_HOST
logger.debug('Host: %s Port: %s Timeout: %s', host, port, self.timeout)
self.queues = defaultdict(dict)
self.client = consul.Consul(host=host, port=int(port))
def _lock_key(self, queue):
return f'{self.prefix}/{queue}.lock'
def _key_prefix(self, queue):
return f'{self.prefix}/{queue}'
def _get_or_create_session(self, queue):
"""Get or create consul session.
Try to renew the session if it exists, otherwise create a new
session in Consul.
This session is used to acquire a lock inside Consul so that we achieve
read-consistency between the nodes.
Arguments:
queue (str): The name of the Queue.
Returns:
str: The ID of the session.
"""
try:
session_id = self.queues[queue]['session_id']
except KeyError:
session_id = None
return (self._renew_existing_session(session_id)
if session_id is not None else self._create_new_session())
def _renew_existing_session(self, session_id):
logger.debug('Trying to renew existing session %s', session_id)
session = self.client.session.renew(session_id=session_id)
return session.get('ID')
def _create_new_session(self):
logger.debug('Creating session %s with TTL %s',
self.lock_name, self.session_ttl)
session_id = self.client.session.create(
name=self.lock_name, ttl=self.session_ttl)
logger.debug('Created session %s with id %s',
self.lock_name, session_id)
return session_id
@contextmanager
def _queue_lock(self, queue, raising=LockError):
"""Try to acquire a lock on the Queue.
It does so by creating a object called 'lock' which is locked by the
current session..
This way other nodes are not able to write to the lock object which
means that they have to wait before the lock is released.
Arguments:
queue (str): The name of the Queue.
raising (Exception): Set custom lock error class.
Raises:
LockError: if the lock cannot be acquired.
Returns:
bool: success?
"""
self._acquire_lock(queue, raising=raising)
try:
yield
finally:
self._release_lock(queue)
def _acquire_lock(self, queue, raising=LockError):
session_id = self._get_or_create_session(queue)
lock_key = self._lock_key(queue)
logger.debug('Trying to create lock object %s with session %s',
lock_key, session_id)
if self.client.kv.put(key=lock_key,
acquire=session_id,
value=self.lock_name):
self.queues[queue]['session_id'] = session_id
return
logger.info('Could not acquire lock on key %s', lock_key)
raise raising()
def _release_lock(self, queue):
"""Try to release a lock.
It does so by simply removing the lock key in Consul.
Arguments:
queue (str): The name of the queue we want to release
the lock from.
"""
logger.debug('Removing lock key %s', self._lock_key(queue))
self.client.kv.delete(key=self._lock_key(queue))
def _destroy_session(self, queue):
"""Destroy a previously created Consul session.
Will release all locks it still might hold.
Arguments:
queue (str): The name of the Queue.
"""
logger.debug('Destroying session %s', self.queues[queue]['session_id'])
self.client.session.destroy(self.queues[queue]['session_id'])
def _new_queue(self, queue, **_):
self.queues[queue] = {'session_id': None}
return self.client.kv.put(key=self._key_prefix(queue), value=None)
def _delete(self, queue, *args, **_):
self._destroy_session(queue)
self.queues.pop(queue, None)
self._purge(queue)
def _put(self, queue, payload, **_):
"""Put `message` onto `queue`.
This simply writes a key to the K/V store of Consul
"""
key = '{}/msg/{}_{}'.format(
self._key_prefix(queue),
int(round(monotonic() * 1000)),
uuid.uuid4(),
)
if not self.client.kv.put(key=key, value=dumps(payload), cas=0):
raise ChannelError(f'Cannot add key {key!r} to consul')
def _get(self, queue, timeout=None):
"""Get the first available message from the queue.
Before it does so it acquires a lock on the Key/Value store so
only one node reads at the same time. This is for read consistency
"""
with self._queue_lock(queue, raising=Empty):
key = f'{self._key_prefix(queue)}/msg/'
logger.debug('Fetching key %s with index %s', key, self.index)
self.index, data = self.client.kv.get(
key=key, recurse=True,
index=self.index, wait=self.timeout,
)
try:
if data is None:
raise Empty()
logger.debug('Removing key %s with modifyindex %s',
data[0]['Key'], data[0]['ModifyIndex'])
self.client.kv.delete(key=data[0]['Key'],
cas=data[0]['ModifyIndex'])
return loads(data[0]['Value'])
except TypeError:
pass
raise Empty()
def _purge(self, queue):
self._destroy_session(queue)
return self.client.kv.delete(
key=f'{self._key_prefix(queue)}/msg/',
recurse=True,
)
def _size(self, queue):
size = 0
try:
key = f'{self._key_prefix(queue)}/msg/'
logger.debug('Fetching key recursively %s with index %s',
key, self.index)
self.index, data = self.client.kv.get(
key=key, recurse=True,
index=self.index, wait=self.timeout,
)
size = len(data)
except TypeError:
pass
logger.debug('Found %s keys under %s with index %s',
size, key, self.index)
return size
@cached_property
def lock_name(self):
return f'{socket.gethostname()}'
[docs]class Transport(virtual.Transport):
"""Consul K/V storage Transport for Kombu."""
Channel = Channel
default_port = DEFAULT_PORT
driver_type = 'consul'
driver_name = 'consul'
if consul:
connection_errors = (
virtual.Transport.connection_errors + (
consul.ConsulException, consul.base.ConsulException
)
)
channel_errors = (
virtual.Transport.channel_errors + (
consul.ConsulException, consul.base.ConsulException
)
)
def __init__(self, *args, **kwargs):
if consul is None:
raise ImportError('Missing python-consul library')
super().__init__(*args, **kwargs)
[docs] def verify_connection(self, connection):
port = connection.client.port or self.default_port
host = connection.client.hostname or DEFAULT_HOST
logger.debug('Verify Consul connection to %s:%s', host, port)
try:
client = consul.Consul(host=host, port=int(port))
client.agent.self()
return True
except ValueError:
pass
return False
