swarm_mode_by_polling_manager

tifayuki · tifayuki · commit dc0c9ca786e6 · 2016-11-03T14:49:48.000-07:00
diff --git a/README.md b/README.md
@@ -57,20 +57,26 @@ That's it - the haproxy container will start querying Docker Cloud's API for an
 
 Docker 1.12 supports SwarmMode natively. `dockercloud/haproxy` will auto config itself to load balance all the services running on the same network:
 
-1. Create a new network using `docker create` command
+1. Create a new network using `docker network create -d overlay <name>` command
 
-2. Launch `dockercloud/haproxy` service on that network
+2. Launch `dockercloud/haproxy` service on that network on manager nodes.
 
 3. Launch your application services that need to be load balanced on the same network.
 
+    **Note**
+    - You **HAVE TO** set the environment variable `SERVICE_PORTS=<port1>, <port2>` in your application service, which are the ports you would like to expose.
+    - For `dockercloud/haproxy` service:
+      If you mount `/var/run/docker.sock`, it can only be run on swarm manager nodes.
+      If you want the haproxy service to run on worker nodes, you need to setup DOCKER_HOST envvar that points to the manager address.
+
 * If your application services need to access other services(database, for example), you can attach your application services to two different network, one is for database and the other one for the proxy
 * This feature is still experimental, please let us know if you find any bugs or have any suggestions.
 
 #### example of docker swarm mode support
 
     docker network create -d overlay proxy
-    docker service create --name haproxy --network proxy --mount target=/var/run/docker.sock,source=/var/run/docker.sock,type=bind -p 80:80 dockercloud/haproxy
-    docker service create --name app --network proxy dockercloud/hello-world
+    docker service create --name haproxy --network proxy --mount target=/var/run/docker.sock,source=/var/run/docker.sock,type=bind -p 80:80 --constraint "node.role == manager" dockercloud/haproxy
+    docker service create -e SERVICE_PORTS="80" --name app --network proxy --constraint "node.role != manager" dockercloud/hello-world
     docker service scale app=2
     docker service update --env-add VIRTUAL_HOST=web.org app
 
@@ -243,6 +249,7 @@ Settings here can overwrite the settings in HAProxy, which are only applied to t
 |TCP_PORTS|comma separated ports(e.g. 9000, 9001, 2222/ssl). The port listed in `TCP_PORTS` will be load-balanced in TCP mode. Port ends with `/ssl` indicates that port needs SSL termination.
 |VIRTUAL_HOST_WEIGHT|an integer of the weight of an virtual host, used together with `VIRTUAL_HOST`, default:0. It affects the order of acl rules of the virtual hosts. The higher weight one virtual host has, the more priority that acl rules applies.|
 |VIRTUAL_HOST|specify virtual host and virtual path. Format: `[scheme://]domain[:port][/path], ...`. wildcard `*` can be used in `domain` and `path` part|
+|SERVICE_PORTS|comma separated ports(e.g. 80, 8080), which are the ports you would like to expose in your application service. This envvar is swarm mode only, and it is **MUST** be set in swarm mode|
 
 Check [the HAProxy configuration manual](http://cbonte.github.io/haproxy-dconv/configuration-1.5.html) for more information on the above.
 
diff --git a/haproxy/config.py b/haproxy/config.py
@@ -65,6 +65,7 @@ def parse_extra_frontend_settings(envvars):
 STATS_PORT = os.getenv("STATS_PORT", "1936")
 TIMEOUT = os.getenv("TIMEOUT", "connect 5000, client 50000, server 50000")
 NBPROC = int(os.getenv("NBPROC", 1))
+SWARM_MODE_POLLING_INTERVAL = int(os.getenv("SWARM_MODE_POLLING_INTERVAL", 5))
 
 # global
 RUNNING_MODE = None
@@ -76,6 +77,7 @@ def parse_extra_frontend_settings(envvars):
 HAPROXY_RUN_COMMAND = ['/usr/sbin/haproxy', '-f', HAPROXY_CONFIG_FILE, '-db', '-q']
 API_RETRY = 10  # seconds
 PID_FILE = "/tmp/dockercloud-haproxy.pid"
+SERVICE_PORTS_ENVVAR_NAME = "SERVICE_PORTS"
 
 # regular expressions
 SERVICE_NAME_MATCH = re.compile(r"(.+)_\d+$")
diff --git a/haproxy/eventhandler.py b/haproxy/eventhandler.py
@@ -97,30 +97,26 @@ def listen_docker_events_compose_mode():
         add_haproxy_run_task("Reconnect docker events")
 
 
-def listen_docker_events_swarm_mode():
+def polling_service_status_swarm_mode():
     while True:
+        time.sleep(config.SWARM_MODE_POLLING_INTERVAL)
         try:
             try:
                 docker = docker_client()
             except:
                 docker = docker_client(os.environ)
 
-            docker.ping()
-            for event in docker.events(decode=True):
-                logger.debug(event)
-                action = event.get("Action", "")
-                type = event.get("Type", "")
-                container = event.get("Actor", {}).get("Attributes", {}).get("container", "")
-                network = event.get("Actor", {}).get("Attributes", {}).get("name")
-                id = event.get("Actor", {}).get("ID", "")
-                if type == "network" and action in ["connect", "disconnect"] and id in Haproxy.cls_swarm_networks:
-                    if action == "connect":
-                        msg = "Docker event: container %s %s to network %s" % (container, action, network)
-                    else:
-                        msg = "Docker event: container %s %s from network %s" % (container, action, network)
-                    add_haproxy_run_task(msg)
+            tasks = docker.tasks(filters={"desired-state": "running"})
+            linked_tasks = set()
+            for task in tasks:
+                task_nets = [network.get("Network", {}).get("ID", "") for network in
+                             task.get("NetworksAttachments", [])]
+                task_service_id = task.get("ServiceID", "")
+                if task_service_id != Haproxy.cls_service_id and Haproxy.cls_nets.intersection(set(task_nets)):
+                    task_id = task.get("ID", "")
+                    linked_tasks.add(task_id)
+
+            if Haproxy.cls_linked_tasks != linked_tasks:
+                add_haproxy_run_task("Tasks are updated")
         except APIError as e:
             logger.info("Docker API error: %s" % e)
-
-        time.sleep(1)
-        add_haproxy_run_task("Reconnect docker events")
diff --git a/haproxy/haproxycfg.py b/haproxy/haproxycfg.py
@@ -26,7 +26,8 @@
 
 
 def add_haproxy_run_task(msg=None):
-    logger.info("=> Add task: %s", msg)
+    if msg:
+        logger.info("=> Add task: %s", msg)
     gevent.spawn(tasks.put, (config.RUNNING_MODE, msg))
 
 
@@ -38,7 +39,8 @@ def run_haproxy():
         while not tasks.empty():
             if mode != RunningMode.CloudMode:
                 delay = 0.1
-            logger.info("=> Task accumulated, skip: %s", msg)
+            if msg:
+                logger.info("=> Task accumulated, skip: %s", msg)
             mode, msg = tasks.get()
             time.sleep(delay)
             continue
@@ -50,11 +52,13 @@ def run_haproxy():
 
 class Haproxy(object):
     cls_linked_services = set()
-    cls_swarm_networks = []
+    cls_linked_tasks = set()
     cls_cfg = None
     cls_process = None
     cls_certs = []
     cls_ca_certs = []
+    cls_nets = set()
+    cls_service_id = ""
 
     def __init__(self, running_mode=RunningMode.LegacyMode):
         logger.info("==========BEGIN==========")
@@ -114,8 +118,10 @@ def _init_swarm_mode_links():
             logger.info("Docker API error, regressing to legacy links mode: %s" % e)
             return None
         haproxy_container_id = os.environ.get("HOSTNAME", "")
-        links, Haproxy.cls_linked_services, Haproxy.cls_swarm_networks = SwarmModeLinkHelper.get_swarm_mode_links(
-            docker, haproxy_container_id)
+        Haproxy.cls_service_id, Haproxy.cls_nets = SwarmModeLinkHelper.get_swarm_mode_haproxy_id_nets(docker,
+                                                                                                      haproxy_container_id)
+        links, Haproxy.cls_linked_tasks = SwarmModeLinkHelper.get_swarm_mode_links(docker, Haproxy.cls_service_id,
+                                                                                   Haproxy.cls_nets)
         logger.info("Linked service: %s", ", ".join(SwarmModeLinkHelper.get_service_links_str(links)))
         logger.info("Linked container: %s", ", ".join(SwarmModeLinkHelper.get_container_links_str(links)))
         return links
diff --git a/haproxy/helper/swarm_mode_link_helper.py b/haproxy/helper/swarm_mode_link_helper.py
@@ -1,70 +1,88 @@
 import logging
+
 import compose_mode_link_helper
+from haproxy.config import SERVICE_PORTS_ENVVAR_NAME
 
 logger = logging.getLogger("haproxy")
 
 
-def get_swarm_mode_links(docker, haproxy_container_short_id):
+def get_swarm_mode_haproxy_id_nets(docker, haproxy_container_short_id):
     try:
         haproxy_container = docker.inspect_container(haproxy_container_short_id)
     except Exception as e:
         logger.info("Docker API error, regressing to legacy links mode: %s" % e)
-        return {}, set()
+        return "", set()
     labels = haproxy_container.get("Config", {}).get("Labels", {})
-    service_id = labels.get("com.docker.swarm.node.id", "")
-    service_name = labels.get("com.docker.swarm.service.name", "")
-    task_id = labels.get("com.docker.swarm.task.id", "")
-    task_name = labels.get("com.docker.swarm.task.name", "")
-    if not (service_id and service_name and task_id and task_name):
+    haproxy_service_id = labels.get("com.docker.swarm.service.id", "")
+    if not haproxy_service_id:
         logger.info("Dockercloud haproxy is not running in a service in SwarmMode")
-        return {}, set()
+        return "", set()
 
-    nets = haproxy_container.get("NetworkSettings", {}).get("Networks", {})
-    net_ids = [network.get("NetworkID", "") for network in nets.values()]
+    haproxy_nets = set([network.get("NetworkID", "") for network in
+                        haproxy_container.get("NetworkSettings", {}).get("Networks", {}).values()])
 
-    linked_containers_ids = []
-    for net_id in net_ids:
-        network_inspect = docker.inspect_network(net_id)
-        linked_containers_ids.extend(network_inspect.get("Containers", {}).keys())
+    return haproxy_service_id, haproxy_nets
 
-    linked_containers = {}
-    for linked_containers_id in linked_containers_ids:
-        try:
-            linked_container = docker.inspect_container(linked_containers_id)
-        except Exception as e:
-            logger.info("Docker API error: %s" % e)
-            continue
-        if linked_container.get("Config", {}).get("Labels", {}).get("com.docker.swarm.service.name",
-                                                                    "") != service_name:
-            linked_containers[linked_containers_id] = linked_container
 
-    links, services = _calc_links(linked_containers)
-    return links, services, net_ids
+def get_swarm_mode_links(docker, haproxy_service_id, haproxy_nets):
+    services = docker.services()
+    tasks = docker.tasks(filters={"desired-state": "running"})
+    links, linked_tasks = get_task_links(tasks, services, haproxy_service_id, haproxy_nets)
+    return links, linked_tasks
 
 
-def _calc_links(containers):
+def get_task_links(tasks, services, haproxy_service_id, haproxy_nets):
+    services_id_name = {s.get("ID"): s.get("Spec", {}).get("Name", "") for s in services}
     links = {}
-    services = set()
-    for container_id, container in containers.items():
-        container_name = container.get("Name").lstrip("/")
-        service_name = container.get("Config", {}).get("Labels", {}).get("com.docker.swarm.service.name", "")
-        container_evvvars = get_container_envvars(container)
-        endpoints = get_container_endpoints(container, container_name)
-        links[container_id] = {"service_name": service_name,
-                               "container_envvars": container_evvvars,
-                               "container_name": container_name,
-                               "endpoints": endpoints,
-                               }
-        services.add(service_name)
-    return links, services
-
-
-def get_container_endpoints(container, container_name):
-    return compose_mode_link_helper.get_container_endpoints(container, container_name)
-
-
-def get_container_envvars(container):
-    return compose_mode_link_helper.get_container_envvars(container)
+    linked_tasks = set()
+    for task in tasks:
+        task_nets = [network.get("Network", {}).get("ID", "") for network in task.get("NetworksAttachments", [])]
+        task_service_id = task.get("ServiceID", "")
+        task_nets_attached = haproxy_nets.intersection(set(task_nets))
+        if task_service_id != haproxy_service_id and task_nets_attached:
+            task_id = task.get("ID", "")
+            task_slot = "%d" % task.get("Slot", 0)
+            task_service_id = task.get("ServiceID", "")
+            task_service_name = services_id_name.get(task_service_id, "")
+            container_name = ".".join([task_service_name, task_slot, task_id])
+            task_envvars = get_task_envvars(task.get("Spec", {}).get("ContainerSpec", {}).get("Env", []))
+
+            service_ports = ""
+            for task_envvar in task_envvars:
+                if task_envvar["key"] == SERVICE_PORTS_ENVVAR_NAME:
+                    service_ports = task_envvar["value"]
+            task_ports = [x.strip() for x in service_ports.strip().split(",") if x.strip()]
+
+            task_ips = []
+            for network_attachment in task.get("NetworksAttachments", []):
+                if network_attachment.get("Network", {}).get("ID", "") in task_nets_attached:
+                    task_ips = network_attachment.get("Addresses", [])
+                    break
+
+            if task_ips:
+                task_ip = task_ips[0].split("/")[0]
+            else:
+                task_ip = container_name
+
+            task_endpoints = {"%s/tcp" % port: "tcp://%s:%s" % (task_ip, port) for port in task_ports}
+
+            links[task_id] = {"endpoints": task_endpoints, "container_name": container_name,
+                              "service_name": task_service_name, "container_envvars": task_envvars}
+            linked_tasks.add(task_id)
+    return links, linked_tasks
+
+
+def get_task_envvars(envvars):
+    new_envvars = []
+    for _envvar in envvars:
+        terms = _envvar.split("=", 1)
+        envvar = {"key": terms[0]}
+        if len(terms) == 2:
+            envvar["value"] = terms[1]
+        else:
+            envvar["value"] = ""
+        new_envvars.append(envvar)
+    return new_envvars
 
 
 def get_service_links_str(links):
diff --git a/haproxy/main.py b/haproxy/main.py
@@ -15,7 +15,7 @@
 import config
 from config import DEBUG, PID_FILE, HAPROXY_CONTAINER_URI, HAPROXY_SERVICE_URI, API_AUTH
 from eventhandler import on_user_reload, listen_docker_events_compose_mode, listen_dockercloud_events, \
-    listen_docker_events_swarm_mode
+    polling_service_status_swarm_mode
 from haproxy import __version__
 import haproxycfg
 from haproxycfg import add_haproxy_run_task, run_haproxy, Haproxy
@@ -58,7 +58,7 @@ def main():
         gevent.spawn(listen_docker_events_compose_mode)
     elif config.RUNNING_MODE == RunningMode.SwarmMode:
         add_haproxy_run_task("Initial start - Swarm Mode")
-        gevent.spawn(listen_docker_events_swarm_mode)
+        gevent.spawn(polling_service_status_swarm_mode)
     elif config.RUNNING_MODE == RunningMode.LegacyMode:
         add_haproxy_run_task("Initial start - Legacy Mode")
 
diff --git a/tests/unit/helper/test_swarm_mode_link_helper.py b/tests/unit/helper/test_swarm_mode_link_helper.py
diff --git a/tests/unit/parser/test_legacy_parser.py b/tests/unit/parser/test_legacy_parser.py
