Merge pull request #11 from docker/support_compose_v2

Support compose v2 TUT-691

Author: Feng Honglin

README.md

@@ -1,7 +1,17 @@
 import os
 import re
 
-from parser import parse_extra_bind_settings
+
+def parse_extra_bind_settings(extra_bind_settings):
+    bind_dict = {}
+    if extra_bind_settings:
+        settings = re.split(r'(?<!\\),', extra_bind_settings)
+        for setting in settings:
+            term = setting.split(":", 1)
+            if len(term) == 2:
+                bind_dict[term[0].strip().replace("\,", ",")] = term[1].strip().replace("\,", ",")
+    return bind_dict
+
 
 # envvar
 DEFAULT_SSL_CERT = os.getenv("DEFAULT_SSL_CERT") or os.getenv("SSL_CERT")
@@ -28,6 +38,7 @@
 HAPROXY_SERVICE_URI = os.getenv("DOCKERCLOUD_SERVICE_API_URI")
 API_AUTH = os.getenv("DOCKERCLOUD_AUTH")
 DEBUG = os.getenv("DEBUG", False)
+LINK_MODE = ""
 
 # const
 CERT_DIR = "/certs/"
@@ -37,6 +48,7 @@
 API_RETRY = 10  # seconds
 PID_FILE = "/tmp/dockercloud-haproxy.pid"
 
+# regular expressions
 SERVICE_NAME_MATCH = re.compile(r"(.+)_\d+$")
 BACKEND_MATCH = re.compile(r"(?P<proto>tcp|udp):\/\/(?P<addr>[^:]*):(?P<port>.*)")
 SERVICE_ALIAS_MATCH = re.compile(r"_PORT_\d{1,5}_(TCP|UDP)$")

haproxy/config.py

@@ -1,7 +1,12 @@
-import logging
 import json
+import logging
+
+import dockercloud
+from compose.cli.docker_client import docker_client
+from docker.errors import APIError
 
 import config
+import helper.cloud_link_helper
 from haproxycfg import run_haproxy, Haproxy
 from utils import get_uuid_from_resource_uri
 
@@ -21,17 +26,17 @@ def on_cloud_event(message):
     if event.get("state", "") not in ["In progress", "Pending", "Terminating", "Starting", "Scaling", "Stopping"] and \
                     event.get("type", "").lower() in ["container", "service"] and \
                     len(set(Haproxy.cls_linked_services).intersection(set(event.get("parents", [])))) > 0:
-        msg = "Event: %s %s is %s" % (
+        msg = "Docker Cloud Event: %s %s is %s" % (
             event["type"], get_uuid_from_resource_uri(event.get("resource_uri", "")), event["state"].lower())
         run_haproxy(msg)
 
     # Add/remove services linked to haproxy
     if event.get("state", "") == "Success" and config.HAPROXY_SERVICE_URI in event.get("parents", []):
-        run_haproxy("Event: New action is executed on the Haproxy container")
+        run_haproxy("Docker Cloud Event: New action is executed on the Haproxy container")
 
 
 def on_websocket_open():
-    Haproxy.LINKED_CONTAINER_CACHE.clear()
+    helper.cloud_link_helper.LINKED_CONTAINER_CACHE.clear()
     run_haproxy("Websocket open")
 
 
@@ -41,3 +46,30 @@ def on_websocket_close():
 
 def on_user_reload(signum, frame):
     run_haproxy("User reload")
+
+
+def listen_dockercloud_events():
+    events = dockercloud.Events()
+    events.on_open(on_websocket_open)
+    events.on_close(on_websocket_close)
+    events.on_message(on_cloud_event)
+    events.run_forever()
+
+
+def listen_docker_events():
+    try:
+        docker = docker_client()
+        docker.ping()
+        for event in docker.events(decode=True):
+            logger.debug(event)
+            attr = event.get("Actor", {}).get("Attributes")
+            compose_project = attr.get("com.docker.compose.project", "")
+            compose_service = attr.get("com.docker.compose.service", "")
+            container_name = attr.get("name", "")
+            event_action = event.get("Action", "")
+            service = "%s_%s" % (compose_project, compose_service)
+            if service in Haproxy.cls_linked_services and event_action in ["start", "die"]:
+                msg = "Docker event: container %s %s" % (container_name, event_action)
+                run_haproxy(msg)
+    except APIError as e:
+        logger.info("Docker API error: %s" % e)

haproxy/eventhandler.py

@@ -2,22 +2,26 @@
 import logging
 from collections import OrderedDict
 
+from compose.cli.docker_client import docker_client
+
+import config
 import helper.backend_helper as BackendHelper
+import helper.cloud_link_helper as CloudLinkHelper
 import helper.config_helper as ConfigHelper
 import helper.frontend_helper as FrontendHelper
-import helper.init_helper as InitHelper
+import helper.new_link_helper as NewLinkHelper
 import helper.ssl_helper as SslHelper
 import helper.tcp_helper as TcpHelper
 import helper.update_helper as UpdateHelper
 from haproxy.config import *
-from parser import Specs
+from haproxy.parser import LegacyLinkSpecs, NewLinkSpecs
 from utils import fetch_remote_obj, prettify, save_to_file, get_service_attribute, get_bind_string
 
 logger = logging.getLogger("haproxy")
 
 
 def run_haproxy(msg=None):
-    haproxy = Haproxy(msg)
+    haproxy = Haproxy(config.LINK_MODE, msg)
     haproxy.update()
 
 
@@ -27,62 +31,78 @@ class Haproxy(object):
     cls_process = None
     cls_certs = []
 
-    cls_service_name_match = re.compile(r"(.+)_\d+$")
-
-    LINKED_CONTAINER_CACHE = {}
-
-    def __init__(self, msg=""):
+    def __init__(self, link_mode="", msg=""):
         logger.info("==========BEGIN==========")
         if msg:
             logger.info(msg)
 
+        self.link_mode = link_mode
         self.ssl_bind_string = None
         self.ssl_updated = False
         self.routes_added = []
         self.require_default_route = False
+        self.specs = None
 
-        self._initialize()
-
-    def _initialize(self):
-        if HAPROXY_CONTAINER_URI and HAPROXY_SERVICE_URI and API_AUTH:
-            haproxy_container = fetch_remote_obj(HAPROXY_CONTAINER_URI)
+        self.specs = self._initialize(self.link_mode)
 
-            haproxy_links = InitHelper.get_links_from_haproxy(haproxy_container.linked_to_container)
-            new_added_container_uris = InitHelper.get_new_added_link_uri(Haproxy.LINKED_CONTAINER_CACHE, haproxy_links)
-            new_added_containers = InitHelper.get_container_object_from_uri(new_added_container_uris)
-            InitHelper.update_container_cache(Haproxy.LINKED_CONTAINER_CACHE, new_added_container_uris,
-                                              new_added_containers)
-            linked_containers = InitHelper.get_linked_containers(Haproxy.LINKED_CONTAINER_CACHE,
-                                                                 haproxy_container.linked_to_container)
-            InitHelper.update_haproxy_links(haproxy_links, linked_containers)
-
-            logger.info("Service links: %s", ", ".join(InitHelper.get_service_links_str(haproxy_links)))
-            logger.info("Container links: %s", ", ".join(InitHelper.get_container_links_str(haproxy_links)))
-
-            Haproxy.cls_linked_services = InitHelper.get_linked_services(haproxy_links)
-            self.specs = Specs(haproxy_links)
+    @staticmethod
+    def _initialize(link_mode):
+        if link_mode == "cloud":
+            links = Haproxy._init_cloud_links()
+            specs = NewLinkSpecs(links)
+        elif link_mode == "new":
+            links = Haproxy._init_new_links()
+            if links is None:
+                specs = LegacyLinkSpecs()
+            else:
+                specs = NewLinkSpecs(links)
         else:
-            logger.info("Loading HAProxy definition from environment variables")
-            Haproxy.cls_linked_services = None
-            Haproxy.specs = Specs()
+            specs = LegacyLinkSpecs()
+        return specs
 
-    def update(self):
-        self._config_ssl()
+    @staticmethod
+    def _init_cloud_links():
+        haproxy_container = fetch_remote_obj(HAPROXY_CONTAINER_URI)
+        links = CloudLinkHelper.get_cloud_links(haproxy_container)
+        Haproxy.cls_linked_services = CloudLinkHelper.get_linked_services(links)
+        logger.info("Linked service: %s", ", ".join(CloudLinkHelper.get_service_links_str(links)))
+        logger.info("Linked container: %s", ", ".join(CloudLinkHelper.get_container_links_str(links)))
+        return links
 
-        cfg_dict = OrderedDict()
-        cfg_dict.update(self._config_global_section())
-        cfg_dict.update(self._config_defaults_section())
-        cfg_dict.update(self._config_stats_section())
-        cfg_dict.update(self._config_userlist_section(HTTP_BASIC_AUTH))
-        cfg_dict.update(self._config_tcp_sections())
-        cfg_dict.update(self._config_frontend_sections())
-        cfg_dict.update(self._config_backend_sections())
+    @staticmethod
+    def _init_new_links():
+        try:
+            docker = docker_client()
+            docker.ping()
+            container_id = os.environ.get("HOSTNAME", "")
+            haproxy_container = docker.inspect_container(container_id)
+        except Exception as e:
+            logger.info("Docker API error, regressing to legacy links mode: ", e)
+            return None
+        links, Haproxy.cls_linked_services = NewLinkHelper.get_new_links(docker, haproxy_container)
+        logger.info("Linked service: %s", ", ".join(NewLinkHelper.get_service_links_str(links)))
+        logger.info("Linked container: %s", ", ".join(NewLinkHelper.get_container_links_str(links)))
+        return links
 
-        cfg = prettify(cfg_dict)
-        self._update_haproxy(cfg)
+    def update(self):
+        if self.specs:
+            self._config_ssl()
+            cfg_dict = OrderedDict()
+            cfg_dict.update(self._config_global_section())
+            cfg_dict.update(self._config_defaults_section())
+            cfg_dict.update(self._config_stats_section())
+            cfg_dict.update(self._config_userlist_section(HTTP_BASIC_AUTH))
+            cfg_dict.update(self._config_tcp_sections())
+            cfg_dict.update(self._config_frontend_sections())
+            cfg_dict.update(self._config_backend_sections())
+
+            cfg = prettify(cfg_dict)
+            self._update_haproxy(cfg)
+        else:
+            logger.info("Internal error: Specs is not initialized")
 
     def _update_haproxy(self, cfg):
-        if HAPROXY_SERVICE_URI and HAPROXY_CONTAINER_URI and API_AUTH:
+        if self.link_mode in ["cloud", "new"]:
             if Haproxy.cls_cfg != cfg:
                 logger.info("HAProxy configuration:\n%s" % cfg)
                 Haproxy.cls_cfg = cfg
@@ -94,10 +114,10 @@ def _update_haproxy(self, cfg):
             else:
                 logger.info("HAProxy configuration remains unchanged")
             logger.info("===========END===========")
-        else:
+        elif self.link_mode in ["legacy"]:
             logger.info("HAProxy configuration:\n%s" % cfg)
-            save_to_file(HAPROXY_CONFIG_FILE, cfg)
-            UpdateHelper.run_once()
+            if save_to_file(HAPROXY_CONFIG_FILE, cfg):
+                UpdateHelper.run_once()
 
     def _config_ssl(self):
         ssl_bind_string = ""

haproxy/haproxycfg.py

@@ -3,8 +3,20 @@
 from haproxy.config import SERVICE_NAME_MATCH
 from haproxy.utils import get_uuid_from_resource_uri, fetch_remote_obj
 
+LINKED_CONTAINER_CACHE = {}
 
-def get_links_from_haproxy(container_links):
+
+def get_cloud_links(haproxy_container):
+    links = _init_links(haproxy_container.linked_to_container)
+    new_added_container_uris = _get_new_added_link_uri(LINKED_CONTAINER_CACHE, links)
+    new_added_containers = _get_container_object_from_uri(new_added_container_uris)
+    _update_container_cache(LINKED_CONTAINER_CACHE, new_added_container_uris, new_added_containers)
+    linked_containers = _get_linked_containers(LINKED_CONTAINER_CACHE, haproxy_container.linked_to_container)
+    _update_links(links, linked_containers)
+    return links
+
+
+def _init_links(container_links):
     links = {}
     for link in container_links:
         linked_container_uri = link["to_container"]
@@ -24,37 +36,30 @@ def get_links_from_haproxy(container_links):
     return links
 
 
-def get_new_added_link_uri(container_object_cache, links):
+def _get_new_added_link_uri(container_object_cache, links):
     return filter(lambda x: x not in container_object_cache, links)
 
 
-def get_container_object_from_uri(container_uris):
+def _get_container_object_from_uri(container_uris):
     pool = ThreadPool(processes=10)
     container_objects = pool.map(fetch_remote_obj, container_uris)
     return container_objects
 
 
-def update_container_cache(cache, new_container_object_uris, new_container_objects):
+def _update_container_cache(cache, new_container_object_uris, new_container_objects):
     for i, container_uri in enumerate(new_container_object_uris):
         cache[container_uri] = new_container_objects[i]
 
 
-def update_haproxy_links(links, linked_containers):
+def _update_links(links, linked_containers):
     for linked_container in linked_containers:
         linked_container_uri = linked_container.resource_uri
         linked_container_service_uri = linked_container.service
-        linked_container_name = links[linked_container_uri]["container_name"]
-
-        linked_container_envvars = {}
-        for envvar in linked_container.container_envvars:
-            if "_ENV_" not in envvar['key']:
-                linked_container_envvars["%s_ENV_%s" % (linked_container_name, envvar['key'])] = envvar['value']
-
         links[linked_container_uri]["service_uri"] = linked_container_service_uri
-        links[linked_container_uri]["container_envvars"] = linked_container_envvars
+        links[linked_container_uri]["container_envvars"] = linked_container.container_envvars
 
 
-def get_linked_containers(cache, container_links):
+def _get_linked_containers(cache, container_links):
     linked_containers = [cache[link["to_container"]] for link in container_links]
     return linked_containers
 
@@ -68,10 +73,11 @@ def get_linked_services(haproxy_links):
 
 
 def get_service_links_str(haproxy_links):
-    return sorted(set(["%s(%s)" % (link.get("service_name"), get_uuid_from_resource_uri(link.get("service_uri")))
+    return sorted(set(["%s(%s)" % (link.get("service_name"), get_uuid_from_resource_uri(link.get("service_uri", "")))
                        for link in haproxy_links.itervalues()]))
 
 
 def get_container_links_str(haproxy_links):
-    return sorted(set(["%s(%s)" % (link.get("container_name"), get_uuid_from_resource_uri(link.get("container_uri")))
-                       for link in haproxy_links.itervalues()]))
+    return sorted(
+        set(["%s(%s)" % (link.get("container_name"), get_uuid_from_resource_uri(link.get("container_uri", "")))
+             for link in haproxy_links.itervalues()]))