reload haproxy when scale a additional service on compose v2

Author: tifayuki

README.md

@@ -166,7 +166,7 @@ Settings in this part is immutable, you have to redeploy HAProxy service to make
 
 |Environment Variable|Default|Description|
 |:-----:|:-----:|:----------|
-|ADDITIONAL_SERVICES| |list of additional services to balance (es: `prj1:web,prj2:sql`). Discovery will be based on `com.docker.compose.[project|service]` container labels. This environment variable only works on compose v2, and the referenced services must be on a network accessible to this containers.|
+|ADDITIONAL_SERVICES| |list of additional services to balance (es: `prj1:web,prj2:sql`). Discovery will be based on `com.docker.compose.[project|service]` container labels. This environment variable only works on compose v2, and the referenced services must be on a network resolvable and accessible to this containers.|
 |BALANCE|roundrobin|load balancing algorithm to use. Possible values include: `roundrobin`, `static-rr`, `source`, `leastconn`. See:[HAProxy:balance](https://cbonte.github.io/haproxy-dconv/configuration-1.5.html#4-balance)|
 |CA_CERT_FILE| |the path of a ca-cert file. This allows you to mount your ca-cert file directly from a volume instead of from envvar. If set, `CA_CERT` envvar will be ignored. Possible value: `/cacerts/cert0.pem`|
 |CA_CERT| |CA cert for haproxy to verify the client. Use the same format as `DEFAULT_SSL_CERT`|

haproxy/eventhandler.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import time
+import os
 
 import dockercloud
 from compose.cli.docker_client import docker_client
@@ -26,7 +27,7 @@ def on_cloud_event(message):
     # When service scale up/down or container start/stop/terminate/redeploy, reload the service
     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:
+                    len(Haproxy.cls_linked_services.intersection(set(event.get("parents", [])))) > 0:
         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)
@@ -74,7 +75,12 @@ def listen_dockercloud_events():
 
 def listen_docker_events():
     try:
-        docker = docker_client()
+
+        try:
+            docker = docker_client()
+        except:
+            docker = docker_client(os.environ)
+
         docker.ping()
         for event in docker.events(decode=True):
             logger.debug(event)

haproxy/haproxycfg.py

@@ -26,7 +26,7 @@ def run_haproxy(msg=None):
 
 
 class Haproxy(object):
-    cls_linked_services = []
+    cls_linked_services = set()
     cls_cfg = None
     cls_process = None
     cls_certs = []
@@ -77,23 +77,24 @@ def _init_cloud_links():
     @staticmethod
     def _init_new_links():
         try:
-            docker = docker_client()
+            try:
+                docker = docker_client()
+            except:
+                docker = docker_client(os.environ)
+
             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)
+            logger.info("Docker API error, regressing to legacy links mode: %s" % e)
             return None
         links, Haproxy.cls_linked_services = NewLinkHelper.get_new_links(docker, haproxy_container)
 
-        try:
-            if ADDITIONAL_SERVICES:
-                additional_services = ADDITIONAL_SERVICES.split(",")
-                NewLinkHelper.get_additional_links(docker, additional_services, haproxy_container,
-                                                   links, Haproxy.cls_linked_services)
-        except Exception as e:
-            logger.info("Error loading ADDITIONAL_SERVICES: %s" % str(e))
-            return None
+        if ADDITIONAL_SERVICES:
+            additional_links, additional_services= NewLinkHelper.get_additional_links(docker, ADDITIONAL_SERVICES)
+            if additional_links and additional_services:
+                links.update(additional_links)
+                Haproxy.cls_linked_services.update(additional_services)
 
         logger.info("Linked service: %s", ", ".join(NewLinkHelper.get_service_links_str(links)))
         logger.info("Linked container: %s", ", ".join(NewLinkHelper.get_container_links_str(links)))

haproxy/helper/new_link_helper.py

@@ -13,36 +13,24 @@ def get_new_links(docker, haproxy_container):
     linked_compose_services = _get_linked_compose_services(networks, project)
 
     links = _calc_links(docker, linked_compose_services, project)
-    return links, ["%s_%s" % (project, service) for service in linked_compose_services]
+    return links, set(["%s_%s" % (project, service) for service in linked_compose_services])
 
 
-def get_additional_links(docker, additional_services, haproxy_container, links, linked_services):
-    networks_data = docker.networks()
-    haproxy_networks_ids = _find_container_networks_ids(haproxy_container, networks_data)
-    for _container in docker.containers():
-        container_id = _container.get("Id", "")
-        container = docker.inspect_container(container_id)
-        compose_project = container.get("Config", {}).get("Labels", {}).get("com.docker.compose.project", "")
-        compose_service = container.get("Config", {}).get("Labels", {}).get("com.docker.compose.service", "")
-        for _service in additional_services:
-            terms = _service.strip().split(":")
+def get_additional_links(docker, additional_services):
+    links = {}
+    services = set()
+    for additional_service in additional_services.split(","):
+            terms = additional_service.strip().split(":")
             if len(terms) == 2:
-                if terms[0].strip() == compose_project and terms[1].strip() == compose_service:
-                    container_networks_ids = _find_container_networks_ids(container, networks_data)
-                    if set(container_networks_ids).intersection(haproxy_networks_ids):
-                        if _service not in linked_services:
-                            linked_services.append(_service)
-                        container_name = container.get("Name").lstrip("/")
-                        container_evvvars = _get_container_envvars(container)
-                        endpoints = _get_container_endpoints(container, container_name)
-                        links[container_id] = {"service_name": _service,
-                                               "container_envvars": container_evvvars,
-                                               "container_name": container_name,
-                                               "endpoints": endpoints,
-                                               "compose_service": compose_service,
-                                               "compose_project": compose_project}
-                    else:
-                        logger.info("Ignoring container '%s': no shared network with haproxy")
+                project = terms[0]
+                service = terms[1]
+                link = _calc_links(docker, [service], project)
+                if link:
+                    links.update(link)
+                    services.add("%s_%s" % (project, service))
+                else:
+                    logger.info("Cannot find the additional service: %s" % additional_service.strip())
+    return links, services
 
 
 def _find_container_networks_ids(container, networks_data):

haproxy/main.py

@@ -68,7 +68,10 @@ def check_link_mode(container_uri, service_uri, api_auth):
         link_mode = "new"
         reason = ""
         try:
-            docker = docker_client()
+            try:
+                docker = docker_client()
+            except:
+                docker = docker_client(os.environ)
             docker.ping()
         except Exception as e:
             reason = "unable to connect to docker daemon %s" % e

tests/unit/helper/test_new_link_helper.py

@@ -1,7 +1,7 @@
 import unittest
 
 from haproxy.helper.new_link_helper import _calc_links, _get_linked_compose_services, _get_container_envvars, \
-    _get_container_endpoints, get_container_links_str, get_service_links_str
+    _get_container_endpoints, get_container_links_str, get_service_links_str, get_additional_links
 
 container1 = {u'ExecIDs': None,
               u'State': {u'Status': u'running', u'Pid': 28121, u'OOMKilled': False, u'Dead': False, u'Paused': False,
@@ -324,6 +324,17 @@ def test_calc_links(self):
         docker = self.Docker()
         self.assertEqual(links, _calc_links(docker, ['hello', 'world'], 'tmp'))
 
+    def test_get_additional_links(self):
+        docker = self.Docker()
+
+        additional_links, additional_services = get_additional_links(docker, "tmp:hello, tmp:world")
+        self.assertEqual(links, additional_links)
+        self.assertEqual(set(["tmp_hello","tmp_world"]), additional_services)
+
+        additional_links, additional_services = get_additional_links(docker, "tmp:hello, tmp:world, tmp:aaa, aaa:bbb")
+        self.assertEqual(links, additional_links)
+        self.assertSetEqual(set(["tmp_hello", "tmp_world"]), additional_services)
+
     def test_get_container_endpoints(self):
         endpoints = {u'1936/tcp': u'tcp://tmp_lb_1:1936', u'443/tcp': u'tcp://tmp_lb_1:443',
                      u'80/tcp': u'tcp://tmp_lb_1:80'}

tests/unit/test_eventhandler.py

@@ -13,7 +13,7 @@
 class OnCloudEventTestCase(unittest.TestCase):
     def setUp(self):
         self.linked_services = Haproxy.cls_linked_services
-        Haproxy.cls_linked_services = ["/svc/a/", "/svc/b/", "/svc/c/"]
+        Haproxy.cls_linked_services = set(["/svc/a/", "/svc/b/", "/svc/c/"])
         self.service_uri = config.HAPROXY_SERVICE_URI
         config.HAPROXY_SERVICE_URI = "/svc/uuid/"