ZooKeeper configuration plugin

admin/packaging.py

@@ -1004,6 +1004,8 @@ def omnibus_package_builder(
                      flocker_node_path),
                     (FilePath('/opt/flocker/bin/flocker-node-era'),
                      flocker_node_path),
+                    (FilePath('/opt/flocker/bin/flocker-zk'),
+                     flocker_node_path),
                 ]
             ),
             BuildPackage(
@@ -1116,10 +1118,13 @@ class DockerBuild(object):
         to build.
     """
     def run(self):
-        check_call(
-            ['docker', 'build',
-             '--pull', '--tag', self.tag,
-             self.build_directory.path])
+        cmd = ['docker', 'build']
+        for env_key in ('http_proxy', 'https_proxy'):
+            env_val = os.environ.get(env_key)
+            if env_val is not None:
+                cmd.extend(['--build-arg', '='.join((env_key, env_val))])
+        cmd.extend(['--pull', '--tag', self.tag, self.build_directory.path])
+        check_call(cmd)
 
 
 @attributes(['tag', 'volumes', 'command'])
@@ -1138,9 +1143,14 @@ def run(self):
         for container, host in self.volumes.iteritems():
             volume_options.extend(
                 ['--volume', '%s:%s' % (host.path, container.path)])
-
+        proxy_options = []
+        for env_key in ('http_proxy', 'https_proxy'):
+            env_val = os.environ.get(env_key)
+            if env_val is not None:
+                proxy_options.extend(['-e', '='.join((env_key, env_val))])
         result = call(
             ['docker', 'run', '--rm'] +
+            proxy_options +
             volume_options + [self.tag] + self.command)
         if result:
             raise SystemExit(result)

docs/flocker-features/configuration-stores.rst

@@ -0,0 +1,53 @@
+====================
+Configuration Stores
+====================
+
+The :ref:`Flocker control service <control-service>` maintains information about the state of the cluster and must store it across restarts. The default persistence mechanism for control service data is to save it as a file on the control node.
+
+By introducing a configuration store plugin architecture, Flocker opens up the possibility of using a variety of persistence mechanisms. Using a network-based configuration store is an important step toward building a highly available Flocker control service.
+
+
+Listing and activating configuration stores
+===========================================
+
+Configuration store plugins are added to Flocker as a community effort. Run ``flocker-control --help`` for details on which configuration stores are supported by your Flocker distribution, and how to use them.
+
+Activating a specific configuration store requires changing the command line arguments in the flocker-control startup script. For example when running flocker-control under systemd on CentOS 7:
+
+#. Modify ``/usr/lib/systemd/system/flocker-control.service`` and add the ``--configuration-store-plugin`` parameter::
+
+    ExecStart=/usr/sbin/flocker-control --configuration-store-plugin=myplugin --extra-pluginvar=foo
+
+#. Execute::
+
+    systemctl daemon-reload
+    systemctl restart flocker-control
+
+
+Directory configuration store
+=============================
+
+The default. Saves the cluster state as a flat file in a configurable location, by default in ``/var/lib/flocker``. Example::
+
+
+    flocker-control --data-path=/path/to/share
+
+
+ZooKeeper configuration store
+=============================
+
+Saves the configuration in `ZooKeeper <https://zookeeper.apache.org/>`_ , a highly available key-value store designed for maintaining configuration information. Example::
+
+    flocker-control --configuration-store-plugin=zookeeper --zookeeper-hosts=192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181
+
+
+Helper script
+-------------
+
+A bundled tool, `flocker-zk`, helps migrate and examine configuration data held in ZooKeeper.
+
+
+Log entries
+-----------
+
+The log message type `flocker:control:store:zookeeper` identifies entries originating from this plugin.

docs/flocker-features/index.rst

@@ -12,3 +12,4 @@ Flocker Features
    security
    storage-backends
    storage-profiles
+   configuration-stores

docs/gettinginvolved/infrastructure/packaging.rst

@@ -187,7 +187,7 @@ To run the ``flocker-control`` container:
         clusterhqci/flocker-control:master
 
 flocker-docker-plugin
---------------------
+---------------------
 
 The ``flocker-docker-plugin`` Docker image is built using the same ``docker build ...`` command line as for ``flocker-dataset-agent`` but substituting the ``docker-plugin``.
 

flocker/cli/zookeeper_script.py

@@ -0,0 +1,127 @@
+# Copyright (C) 2017 Nokia Corporation and/or its subsidiary(-ies).
+# See LICENSE file for details.
+
+"""
+Command line tool for manipulating ZooKeeper-held Flocker control data.
+"""
+
+from argparse import ArgumentParser, RawDescriptionHelpFormatter
+from contextlib import contextmanager
+import logging
+import datetime
+import sys
+
+from kazoo.client import KazooClient
+from kazoo.exceptions import NoNodeError
+
+ZK_CONFIG_PATH = '/flocker/configuration'
+ZOOKEEPER_HOSTS = 'localhost:2181'
+MIGRATE_FROM = '/var/lib/flocker/current_configuration.json'
+EPILOG = '''get:
+  Print ZooKeeper node contents to stdout.
+
+put:
+  Pipe from stdin to ZooKeeper.
+
+migrate:
+  Store %s in ZooKeeper.
+''' % MIGRATE_FROM
+
+
+class ScriptError(RuntimeError):
+    '''Known errors we want to catch'''
+
+
+@contextmanager
+def zk_client(zkhosts):
+    '''Minimal ZooKeeper client'''
+    _client = KazooClient(
+        hosts=zkhosts,
+        connection_retry={'max_tries': 4, 'delay': 1},
+        command_retry={'max_tries': 4, 'delay': 1}
+        )
+    try:
+        _client.start()
+        yield _client
+    finally:
+        _client.stop()
+
+
+def get(zkhosts, node):
+    '''Get node contents from ZooKeeper'''
+    with zk_client(zkhosts) as client:
+        try:
+            content, stat = client.get(node)
+        except NoNodeError:
+            raise ScriptError('No such node %s' % node)
+        mtime = datetime.datetime.fromtimestamp(stat.mtime / 1000).isoformat()
+        logging.info(
+            'Fetched node. Version:%s, mtime:%s, size:%s',
+            stat.version, mtime, len(content))
+        return content
+
+
+def put(zkhosts, node, content):
+    '''Store something to a ZooKeeper node'''
+    with zk_client(zkhosts) as client:
+        client.ensure_path(node)
+        stat = client.set(node, content)
+        mtime = datetime.datetime.fromtimestamp(stat.mtime / 1000).isoformat()
+        logging.info(
+            'Stored node. Version:%s, mtime:%s, size:%s',
+            stat.version, mtime, len(content))
+        return stat.version
+
+
+def main():
+    '''Entrypoint for the command line tool'''
+    parser = ArgumentParser(description=__doc__, epilog=EPILOG,
+                            formatter_class=RawDescriptionHelpFormatter)
+    parser.add_argument('action',
+                        nargs='?',
+                        choices=['put', 'get', 'migrate'],
+                        default='get',
+                        help='default: get; see below for details')
+    parser.add_argument('--zookeeper-hosts', '-z',
+                        help='default: %s' % ZOOKEEPER_HOSTS)
+    parser.add_argument('--node',
+                        '-n',
+                        help='default: %s' % ZK_CONFIG_PATH)
+    parser.add_argument('-v',
+                        dest='verbose',
+                        action='count',
+                        default=0,
+                        help='increase verbosity')
+    parser.add_argument('-q',
+                        dest='quiet',
+                        action='count',
+                        default=0,
+                        help='decrease verbosity')
+    parser.set_defaults(zookeeper_hosts=ZOOKEEPER_HOSTS, node=ZK_CONFIG_PATH)
+    args = parser.parse_args()
+    log_level = logging.WARNING - (args.verbose * 10) + (args.quiet * 10)
+    logging.basicConfig(
+        level=log_level,
+        format='%(message)s',
+        handlers=[logging.StreamHandler(sys.stderr)])
+    try:
+        if args.action == 'get':
+            sys.stdout.write(get(args.zookeeper_hosts, args.node))
+        elif args.action == 'put':
+            print put(args.zookeeper_hosts, args.node, sys.stdin.read())
+        elif args.action == 'migrate':
+            with open(MIGRATE_FROM, 'rb') as dir_cfg_store:
+                content = dir_cfg_store.read()
+                print put(args.zookeeper_hosts, args.node, content)
+    except ScriptError, exc:
+        logging.error(str(exc))
+        sys.exit(2)
+    except Exception, exc:  # pylint: disable=broad-except
+        if log_level <= logging.DEBUG:
+            raise
+        else:
+            logging.error(str(exc))
+            sys.exit(3)
+
+if __name__ == '__main__':
+    main()

flocker/control/configuration_store/test/test_zookeeper.py

@@ -0,0 +1,42 @@
+# Copyright (C) 2017 Nokia Corporation and/or its subsidiary(-ies).
+# See LICENSE file for details.
+
+"""
+Tests for ``flocker.control.configuration_store.zookeeper``.
+"""
+import subprocess
+from time import sleep
+
+from ....testtools import AsyncTestCase
+from ..testtools import IConfigurationStoreTestsMixin
+from ..zookeeper import ZooKeeperConfigurationStore
+
+
+class ZooKeeperConfigurationStoreInterfaceTests(IConfigurationStoreTestsMixin,
+                                                AsyncTestCase):
+    """
+    Tests for ``ZooKeeperConfigurationStore``.
+    """
+    def _wait_for_zk(self, maxtries=15):
+        cmd = ['docker', 'logs', self.docker_id]
+        for _ in range(maxtries):
+            subpr = subprocess.Popen(
+                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            stdout, _stderr = subpr.communicate()
+            if 'binding to port' in stdout:
+                return
+            sleep(1)
+        raise RuntimeError('Timed out waiting for Zookeeper container')
+
+    def setUp(self):
+        super(ZooKeeperConfigurationStoreInterfaceTests, self).setUp()
+        self.docker_id = subprocess.check_output(
+            ['docker', 'run', '-d', '--network=host', 'zookeeper:3.4']).strip()
+        self._wait_for_zk()
+        self.store = ZooKeeperConfigurationStore(
+            zookeeper_hosts='localhost:2181'
+        )
+
+    def tearDown(self):
+        super(ZooKeeperConfigurationStoreInterfaceTests, self).tearDown()
+        subprocess.check_output(['docker', 'rm', '-f', self.docker_id])

flocker/control/configuration_store/zookeeper.py

@@ -0,0 +1,87 @@
+# Copyright (C) 2017 Nokia Corporation and/or its subsidiary(-ies).
+# See LICENSE file for details.
+
+"""
+Persistence of cluster configuration to ZooKeeper.
+"""
+
+from contextlib import contextmanager
+from datetime import datetime
+import logging
+
+from eliot import Message
+from kazoo.client import KazooClient
+from pyrsistent import PClass, field
+from twisted.internet.defer import succeed
+
+from zope.interface import implementer
+
+from .interface import IConfigurationStore
+
+
+ZK_CONFIG_PATH = '/flocker/configuration'
+
+
+# produce Eliot logs from standard logger
+class EliotLogHandler(logging.Handler):
+    def emit(self, record):
+        Message.new(
+            message_type=u'flocker:control:store:zookeeper',
+            message=record.getMessage()
+        ).write()
+
+
+logger = logging.getLogger("kazoo")
+logger.setLevel(logging.INFO)
+logger.addHandler(EliotLogHandler())
+
+
+@implementer(IConfigurationStore)
+class ZooKeeperConfigurationStore(PClass):
+    zookeeper_hosts = field(mandatory=True, type=str)
+
+    @property
+    @contextmanager
+    def client(self):
+        _client = KazooClient(
+            hosts=self.zookeeper_hosts,
+            connection_retry={'max_tries': 8, 'delay': 1},
+            command_retry={'max_tries': 8, 'delay': 1}
+            )
+        try:
+            _client.start()
+            yield _client
+        finally:
+            _client.stop()
+
+    def initialize(self):
+        with self.client as client:
+            client.ensure_path(ZK_CONFIG_PATH)
+        return succeed(None)
+
+    def get_content_sync(self):
+        with self.client as client:
+            out, stat = client.get(ZK_CONFIG_PATH)
+            mtime = datetime.fromtimestamp(stat.mtime / 1000).isoformat()
+            logger.info(
+                'Fetched configuration version: %s, mtime: %s, size: %s',
+                stat.version, mtime, len(out))
+        return out
+
+    def get_content(self):
+        return succeed(self.get_content_sync())
+
+    def set_content(self, content):
+        with self.client as client:
+            stat = client.set(ZK_CONFIG_PATH, content)
+            mtime = datetime.fromtimestamp(stat.mtime / 1000).isoformat()
+            logger.info(
+                'Stored configuration version: %s, mtime: %s, size: %s',
+                stat.version, mtime, len(content))
+        return succeed(None)
+
+
+def zookeeper_store_from_options(options):
+    return ZooKeeperConfigurationStore(
+        zookeeper_hosts=options["zookeeper-hosts"]
+    )

flocker/control/script.py

@@ -22,6 +22,7 @@
 from ._persistence import ConfigurationPersistenceService
 from ._clusterstate import ClusterStateService
 from .configuration_store.directory import directory_store_from_options
+from .configuration_store.zookeeper import zookeeper_store_from_options
 from ..common.script import (
     flocker_standard_options, FlockerScriptRunner, main_for_service,
     enable_profiling, disable_profiling)
@@ -70,6 +71,14 @@ def __unicode__(self):
         ]],
 
     ),
+    ConfigurationStorePlugin(
+        name=u'zookeeper',
+        factory=zookeeper_store_from_options,
+        options=[[
+            "zookeeper-hosts", "z", b"localhost:2181",
+            "The directory where data will be persisted.", str
+        ]],
+    )
 ]
 CONFIGURATION_STORE_PLUGINS_BY_NAME = {
     p.name: p for p in CONFIGURATION_STORE_PLUGINS