This document describes the current stable version of Kombu (5.3). For development docs, go here.

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]

"""

from __future__ import annotations

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
[docs] def driver_version(self): return consul.__version__