dockercloud-haproxy/haproxy/eventhandler.py at 6ce3c0b05eab2704bacb1bb45e3b3fa120d694f8 · docker-archive/dockercloud-haproxy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import json
import logging
import os
import time

import dockercloud
from compose.cli.docker_client import docker_client
from docker.errors import APIError

import config
import helper.cloud_mode_link_helper
import helper.swarm_mode_link_helper as SwarmModeLinkHelper
from haproxycfg import add_haproxy_run_task, Haproxy
from utils import get_uuid_from_resource_uri

logger = logging.getLogger("haproxy")


def on_cloud_event(message):
    logger.debug(message)
    logger.debug(Haproxy.cls_linked_services)
    try:
        event = json.loads(message)
    except ValueError:
        logger.info("event is not a valid json message")
        return

    # 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(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())
        add_haproxy_run_task(msg)

    # Add/remove services linked to haproxy
    if event.get("state", "") == "Success" and config.HAPROXY_SERVICE_URI in event.get("parents", []):
        add_haproxy_run_task("Docker Cloud Event: New action is executed on the Haproxy container")


def on_websocket_open():
    helper.cloud_mode_link_helper.LINKED_CONTAINER_CACHE.clear()
    add_haproxy_run_task("Websocket open")


def on_websocket_close():
    logger.info("Websocket close")


def on_user_reload():
    Haproxy.cls_cfg = None
    add_haproxy_run_task("User reload")


def on_cloud_error(e):
    if isinstance(e, KeyboardInterrupt):
        exit(0)


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.on_error(on_cloud_error)
    while True:
        try:
            events.run_forever()
        except dockercloud.AuthError as e:
            logger.info("Auth error: %s, retry in 1 hour" % e)
            time.sleep(3600)


def listen_docker_events_compose_mode():
    while True:
        try:
            try:
                docker = docker_client()
            except:
                docker = docker_client(os.environ)

            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)
                    add_haproxy_run_task(msg)
        except APIError as e:
            logger.info("Docker API error: %s" % e)

        time.sleep(1)
        add_haproxy_run_task("Reconnect docker events")


def polling_service_status_swarm_mode():
    while True:
        time.sleep(config.SWARM_MODE_POLLING_INTERVAL)
        try:
            try:
                swarmMaster = docker_client()
            except:
                swarmMaster = docker_client(os.environ)

            if config.SWARM_MASTER_ADDRESS:
                swarmMaster = docker_client(os.environ, host=config.SWARM_MASTER_ADDRESS)

            services = swarmMaster.services()
            tasks = swarmMaster.tasks(filters={"desired-state": "running"})
            _, linked_tasks = SwarmModeLinkHelper.get_task_links(tasks, services, Haproxy.cls_service_id,
                                                                 Haproxy.cls_nets)
            if cmp(Haproxy.cls_linked_tasks, linked_tasks) != 0:
                add_haproxy_run_task("Tasks are updated")
        except APIError as e:
            logger.info("Docker API error: %s" % e)