Managing python application configuration in a key-value store

Question

For a project we're working on, we need a central place to store the configurations for various applications (they're all headless services running in docker containers so local configuration files or command line parameters aren't going to cut it in production)

We've chosen to use Consul as the central key-value store and since most of the modules are written in Python I've created a config wrapper to interact with it. It's making use of the python-consul SDK for that purpose.

There are two main modes of operation:

1. On initialization we load the current configuration values synchronously.
2. After that a background monitoring job gets kicked off which executes a callback whenever a key changes.

For the second part asyncio is used since Consul provides a long-polling API which blocks key queries until there is an update to the value (or a timeout has elapsed). python-consul provides an asyncio adapter which makes use of aiohttp.

Since this is my first time in working with Python and asyncio I'm looking for feedback on best practices around the use of it. But any other feedback is welcome as well.

Implementation (kv_consul.py):

"""
You'll need: pip install python-consul aiohttp

This implements the interface to the Consul key-value store (http://consul.io)
"""
from typing import List, Callable, Coroutine, Iterable, Union, Tuple
from urllib.parse import urlparse
import logging
import asyncio
import threading

import consul
import consul.aio


class BackgroundTask:
    def run(self, coro: Callable[[any], Coroutine], args: Iterable, done_callback: Callable=None):
        loop = asyncio.get_event_loop()
        loop.run_in_executor(None, self._task_runner, coro, args, done_callback)

    def _task_runner(self, coro: Callable[[any], Coroutine], args: Iterable, done_callback: Callable):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        try:
            fut = asyncio.ensure_future(coro(*args))
            if done_callback:
                fut.add_done_callback(done_callback)

            loop.run_until_complete(fut)
        finally:
            loop.close()


class KvStoreConfig:
    CONSUL_DEFAULT_SCHEME = 'http'

    def __init__(self, keys: List[str], kvstore_endpoint: str=None):
        self.config_keys = keys

        args = {}
        try:
            if kvstore_endpoint:
                if '//' not in kvstore_endpoint:
                    kvstore_endpoint = '//' + kvstore_endpoint
                parts = urlparse(kvstore_endpoint, scheme=self.CONSUL_DEFAULT_SCHEME)
                if parts.hostname:
                    args['host'] = parts.hostname
                if parts.port:
                    args['port'] = parts.port
                if parts.scheme:
                    args['scheme'] = parts.scheme
        except:
            logging.exception("Failed to parse Consul endpoint '{}'".format(str(kvstore_endpoint)))
            raise

        self.consul_args = args
        self.consul = consul.Consul(**self.consul_args)

    def create_if_not_present(self, full_key: str, value: Union[str, bytes]) -> bool:
        return self.consul.kv.put(full_key, value, cas=0)

    def get_source(self) -> str:
        return "Consul@"+self.consul.http.base_uri

    @staticmethod
    def _decode_consul_data_value(data):
        if data is None:
            return None

        val = data['Value']
        if type(val) == str:
            return val
        if type(val) == bytes:
            return val.decode()
        return str(val)

    def __getitem__(self, item: str) -> Union[str, None]:
        index, data = self.consul.kv.get(item)
        return self._decode_consul_data_value(data)

    def start_monitoring(self, change_callback: Callable[[str], None]) -> bool:
        monitoring_started_event = threading.Event()
        BackgroundTask().run(self._monitor, [change_callback, monitoring_started_event])
        return monitoring_started_event.wait(5)

    async def _monitor(self, change_callback: Callable[[str], None], monitoring_started_event: threading.Event) -> None:
        loop = asyncio.get_event_loop()
        c = consul.aio.Consul(loop=loop, **self.consul_args)

        # get the current indices for each key
        futures = [asyncio.ensure_future(self._get_single_key_index(c, k), loop=loop) for k in self.config_keys]
        results = await asyncio.gather(*futures)
        index_map = {tup[0]: tup[1] for tup in results}

        # at this point we've captured the current index for each key, so even if the key gets modified before the
        # individual monitoring futures are executed we can deal with it since Consul will return immediately with
        # the updated value
        monitoring_started_event.set()

        # start monitoring all keys based on the last known index
        awaitables = [self._monitor_single_key(c, k, i, change_callback) for k, i in index_map.items()]
        # block forever - ensures that the event loop keeps running
        await asyncio.wait([asyncio.ensure_future(a, loop=loop) for a in awaitables])

    async def _monitor_single_key(self, c: consul.aio.Consul, key: str, index: str, change_callback: Callable) -> None:
        while True:
            old_index = index
            index, data = await c.kv.get(key, index)
            if old_index != index:
                change_callback(key, self._decode_consul_data_value(data))

    async def _get_single_key_index(self, c: consul.aio.Consul, key: str) -> Tuple[str, str]:
        index, data = await c.kv.get(key)
        return key, index

Implementation notes:

• the KvConfigStore's interface is the way it is because it's being used by a more generic config wrapper which also supports loading config settings from command line and config files (for development, testing and debugging purposes). The idea is that it can be swapped for another implementation if needed (in case we decide to not use Consul any longer)
• The BackgroundTask is a bit of a crutch since asyncio needs a thread driving an event loop. Since none of the existing application modules are written around asyncio I couldn't run the event loop on the main thread so had to fork it off to a background thread

Integration tests:

"""
Note: This test fixture requires internet access and a working docker install in order
      to spin up the consul test container.
      requires: pip install python-consul docker
"""
from unittest import TestCase
import socket
import docker
import time
import consul
import threading
import json

from kv_consul import KvStoreConfig


class TestConsulServer:
    def __init__(self):
        docker_client = docker.from_env(version='auto')

        self.api_port = self.find_free_port()

        config = {
            "data_dir": "/consul/data",
            "advertise_addr": "127.0.0.1",
            "ports": {
                "http": self.api_port,
                "dns": self.find_free_port(),
                "rpc": self.find_free_port(),
                "serf_lan": self.find_free_port(),
                "serf_wan": self.find_free_port(),
                "server": self.find_free_port(),
            },
        }

        env = {'CONSUL_LOCAL_CONFIG': json.dumps(config)}

        self.consul_container = \
            docker_client.containers.run('consul', 'agent -server -bootstrap-expect=1', environment=env,
                                         detach=True, name='unittest_kv_consul', network_mode='host')
        start = time.time()
        while not self.is_port_open(self.api_port) and time.time() - start < 5:
            time.sleep(0.1)

        if not self.is_port_open(self.api_port):
            print(self.consul_container.logs())
            self.shutdown()
            raise Exception('Timed out while waiting for Consul to start up')

        while "cluster leadership acquired" not in str(self.consul_container.logs()) and time.time() - start < 15:
            time.sleep(0.1)

        if "cluster leadership acquired" not in str(self.consul_container.logs()):
            print(self.consul_container.logs())
            self.shutdown()
            raise Exception('Timed out while waiting for Consul to acquire cluster leadership')

        diff = time.time() - start
        print("Consul available within {}sec".format(str(diff)))
        print()

    @staticmethod
    def is_port_open(port):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        result = sock.connect_ex(('127.0.0.1', port))
        sock.close()
        return result == 0

    @staticmethod
    def find_free_port():
        s = socket.socket()
        s.bind(('', 0))  # 0 means: let the OS pick one for you
        port = s.getsockname()[1]
        s.close()
        return port

    def shutdown(self):
        self.consul_container.stop()
        self.consul_container.remove()


class TestConsulKvStoreConfig(TestCase):
    consul_server = None
    consul = None

    @classmethod
    def setUpClass(cls):
        cls.consul_server = TestConsulServer()
        cls.consul = consul.Consul(port=cls.consul_server.api_port)

    @classmethod
    def tearDownClass(cls):
        cls.consul_server.shutdown()

    @classmethod
    def create_or_update_keys(cls, kv: {}):
        for k in kv:
            cls.consul.kv.put(k, kv[k])

    @classmethod
    def delete_keys(cls, kv: {}):
        for k in kv:
            cls.consul.kv.delete(k)

    def test_get_known_keys(self):
        kv = {
            'foo': 'bar',
            'foo/sub': '123',
            'sub/foo/bar_dummy': 'here'
        }
        self.create_or_update_keys(kv)

        try:
            c = KvStoreConfig(kv.keys(), 'localhost:' + str(self.consul_server.api_port))
            self.assertEqual('bar', c['foo'])
            self.assertEqual('123', c['foo/sub'])
            self.assertEqual('here', c['sub/foo/bar_dummy'])
        finally:
            self.delete_keys(kv)

    def test_get_unknown_key(self):
        c = KvStoreConfig({}, 'localhost:' + str(self.consul_server.api_port))
        self.assertEqual(None, c['something'])

    def test_create_if_not_present_creates_new_key(self):
        c = KvStoreConfig({}, 'localhost:' + str(self.consul_server.api_port))
        try:
            self.assertTrue(c.create_if_not_present("something/new", "hello"))
            self.assertEqual("hello", c['something/new'])
        finally:
            self.delete_keys({'something/new'})

    def test_create_if_not_present_does_not_change_existing_key(self):
        kv = {'foo': 'bar'}
        self.create_or_update_keys(kv)

        try:
            c = KvStoreConfig(kv.keys(), 'localhost:' + str(self.consul_server.api_port))
            self.assertFalse(c.create_if_not_present("foo", "hello"))
            self.assertEqual("bar", c['foo'])
        finally:
            self.delete_keys(kv)

    class _KeyUpdateHandler:
        def __init__(self):
            self.updated_key = None
            self.updated_value = None
            self.update_event = threading.Event()

        def __call__(self, *args, **kwargs):
            self.updated_key = args[0]
            self.updated_value = args[1]
            self.update_event.set()

    def test_monitoring_existing_key_update(self):
        kv = {'foo': 'bar'}
        self.create_or_update_keys(kv)
        c = KvStoreConfig(kv.keys(), 'localhost:' + str(self.consul_server.api_port))

        handler = self._KeyUpdateHandler()

        self.assertTrue(c.start_monitoring(handler), msg="Failed to start monitoring")
        self.create_or_update_keys({'foo': 'baz'})
        self.assertTrue(handler.update_event.wait(timeout=5), msg="Timeout while waiting for update callback")
        self.assertEqual('foo', handler.updated_key)
        self.assertEqual('baz', handler.updated_value)
        self.delete_keys(kv)

    def test_monitoring_nonexisting_key_update(self):
        kv = {'foo': 'bar'}
        c = KvStoreConfig(kv.keys(), 'localhost:' + str(self.consul_server.api_port))
        self.assertEqual(None, c['foo'])

        handler = self._KeyUpdateHandler()

        self.assertTrue(c.start_monitoring(handler), msg="Failed to start monitoring")
        self.create_or_update_keys({'foo': 'bar'})
        self.assertTrue(handler.update_event.wait(timeout=5), msg="Timeout while waiting for update callback")
        self.assertEqual('foo', handler.updated_key)
        self.assertEqual('bar', handler.updated_value)
        self.delete_keys(kv)

    def test_monitoring_deleted_key_update(self):
        kv = {'foo': 'bar'}
        self.create_or_update_keys(kv)
        c = KvStoreConfig(kv.keys(), 'localhost:' + str(self.consul_server.api_port))
        self.assertEqual('bar', c['foo'])

        handler = self._KeyUpdateHandler()

        self.assertTrue(c.start_monitoring(handler), msg="Failed to start monitoring")
        self.delete_keys(kv)
        self.assertTrue(handler.update_event.wait(timeout=5), msg="Timeout while waiting for update callback")
        self.assertEqual('foo', handler.updated_key)
        self.assertEqual(None, handler.updated_value)

    def test_get_source_http(self):
        c = KvStoreConfig({}, 'http://localhost:1234')
        self.assertEqual("Consul@http://localhost:1234", c.get_source())

    def test_get_source_https(self):
        c = KvStoreConfig({}, 'https://localhost:1234')
        self.assertEqual("Consul@https://localhost:1234", c.get_source())

    def test_get_source_default_scheme(self):
        c = KvStoreConfig({}, 'localhost:5678')
        self.assertEqual("Consul@http://localhost:5678", c.get_source())

    def test_get_source_default_port(self):
        c = KvStoreConfig({}, 'http://localhost')
        self.assertEqual("Consul@http://localhost:8500", c.get_source())

    def test_get_source_default_scheme_port(self):
        c = KvStoreConfig({}, 'localhost')
        self.assertEqual("Consul@http://localhost:8500", c.get_source())

    def test_get_source_default_all(self):
        c = KvStoreConfig({})
        self.assertEqual("Consul@http://127.0.0.1:8500", c.get_source())

    def test_get_source_ip(self):
        c = KvStoreConfig({}, 'https://192.168.111.222')
        self.assertEqual("Consul@https://192.168.111.222:8500", c.get_source())

Test notes:

• Fires up a temporary docker container to test with

Answer 1

Thank you for the "pip install" advice! That's helpful.

            parts = urlparse(kvstore_endpoint, scheme=self.CONSUL_DEFAULT_SCHEME)
            if parts.hostname:
                args['host'] = parts.hostname
            if parts.port:
                args['port'] = parts.port
            if parts.scheme:
                args['scheme'] = parts.scheme

I feel you're being very forgiving, here. I would prefer to be able to make stronger assertions about what is true at this point. Maybe you're trying to support 'file:///x.txt' URLs? No, I don't really believe that. If we managed to parse a webserver address, I'd expect all three of those to be supplied or defaulted, or we should raise a 'missing part' exception. I think all three are present for all unit tests.

def create_if_not_present(self, full_key: str, value: Union[str, bytes]) -> bool:

Maybe value really does need to be bytes sometimes, but it makes me nervous about callers being sloppy with encoding. Might try sticking with str, see what that impacts, and convince those callers to be more careful.

On the whole your code is easily readable, thank you. You even separated the standard and vendor libraries, kudos. Tiny nit: in get_source(), PEP8 asks for blanks on either side of + operator. I recommend letting flake8 take a pass over *.py.

Things start to get interesting with the (well-named) create_if_not_present(). I wouldn't have minded a comment that mentioned https://python-consul.readthedocs.io/en/latest/#consul.base.Consul.KV.put , but that's just because I'm new to Consul, perhaps it's overkill. The bool type is helpful, but I feel it's still worth adding a brief docstring:

"""Predicate, returns False if the update has not taken place."""

Hmm, return val.decode(), interesting, thank you for teaching me something. I usually specify utf8, but that is default. I wouldn't mind seeing a -> str: return type annotation on _decode_consul_data_value().

In _monitor(), please phrase it this way:

    index_map = {key: index for key, index in results}

(or k, i as you later phrase it).

The idea is that [the KvConfigStore interface] can be swapped for [a non-Consul] implementation if needed

That's an interesting remark. I would believe it more if it mentioned a particular competing implementation, and especially if there was rudimentary code support for it. The most violence happens to the code upon initial encounter with 2nd implementation. After that's working, adding support for a 3rd implementation would likely be not such a big deal. If you want to get a feel for amount of risk mitigation you're getting, I'd advise supporting some 2nd implementation now. But I suspect that Consul will be a good match for your needs, and that this "swap out" goal is not an important one to you, that it would be better to revisit the goal after there's a particular bridge you have to cross.

BackgroundTask looks nice enough.

TestConsulServer's is_open_port() and find_free_port() both call .close(), which suggests that you might insert a with clause.

The dotted quad hardcode is not a big wart, but to eliminate it you might pip install and then consult netifaces.ifaddresses('eth0')[AF_LINK]

You probably want setUpClass to store this often used value: 'localhost:%s' % self.consul_server.api_port

Test coverage looks very good. I imagine coverage measurements come up mostly green. This is some nice solid library code.

Answer 2

Thanks for the interesting question.

Since you already have an answer focusing on the code itself, I'll focus more on the high-level bits.

Why Consul?

First, as you noted, containers bring many benefits, but they also bring new problems. Configuration and service discovery are two of them, and Consul solves them both. However, there are many other problems to solve. In the future, you're likely to need to:

• configure secrets,
• deploy multiple containers using rolling updates on different nodes for high availability,
• store logs,
• balance load,
• control version your configuration,
• add liveness probes (also known as health checks), and so on.

While it's possible to do handle each problem individually, I think it's much better to use something like Kubernetes, Docker Swarm or Mesos to do this. It's more complicated at first, and it forces you to do things as the authors of the tool consider they should be done, but it also avoids a number of headaches. In those cases, the configuration is done through environment variables, which works very well for static configuration across dev, test and production environments.

(Also, receiving live configuration changes is something I've never heard about. Configuration does not change that often, and I believe it's simpler to start a new container when it changes. But it probably provides other benefits too.)

Okay, now that I have said this, I'll consider in the rest of my answer that using Consul was indeed the right choice for reasons that are specific to your team.

The code itself

This will be shorter since the code is actually good, surprisingly so for someone who says to be new to Python. asyncio is well used, but that's less surprising if you come from C#! A few comments:

• Please follow PEP8, especially for the length of lines. For Code Review, actually following the 80 chars limit is nice. For your code, you can stick to 99 chars. Consider integrating flake8 and yapf in your editor.
• Consider decoupling KvStoreConfig from BackgroundTask, you could have a third class or function knowing about the two, but KvStoreConfig does not necessarily needs to know about BackroundTask. But that can probably wait for the next KvStoreConfig implementations.
• Consider using pytest which contains many nice features. The most useful here is probably fixtures instead of the global setUp/tearDown.
• Consider writing a test that checks for a value that changes twice. If you had forgotten the while True, this test would not pass, but you would still have 100% coverage.