workflow/controller/scheduler/mesos_marathon.py at 188396efb41bbf071b4e3c8efd497e29c8c30ccd · drycc/workflow · 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
121
122
123
124
125
126
import re
import time

from django.conf import settings
from marathon import MarathonClient
from marathon.models import MarathonApp
from docker import Client

from .states import JobState
from .fleet import FleetHTTPClient

# turn down standard marathon logging

MATCH = re.compile(
    '(?P<app>[a-z0-9-]+)_?(?P<version>v[0-9]+)?\.?(?P<c_type>[a-z-_]+)?.(?P<c_num>[0-9]+)')
RETRIES = 3
POLL_ATTEMPTS = 30
POLL_WAIT = 100


class MarathonHTTPClient(object):

    def __init__(self, target, auth, options, pkey):
        self.target = settings.MARATHON_HOST
        self.auth = auth
        self.options = options
        self.pkey = pkey
        self.registry = settings.REGISTRY_HOST + ':' + settings.REGISTRY_PORT
        self.client = MarathonClient('http://'+self.target+':8180')
        self.fleet = FleetHTTPClient('/var/run/fleet.sock', auth, options, pkey)

    # helpers
    def _app_id(self, name):
        return name.replace('_', '.')

    # container api
    def create(self, name, image, command='', **kwargs):
        """Create a container"""
        app_id = self._app_id(name)
        l = locals().copy()
        l.update(re.match(MATCH, name).groupdict())
        image = self.registry + '/' + image
        mems = kwargs.get('memory', {}).get(l['c_type'])
        m = 0
        if mems:
            mems = mems.lower()
            if mems[-2:-1].isalpha() and mems[-1].isalpha():
                mems = mems[:-1]
            m = int(mems[:-1])
        c = 0.5
        cpu = kwargs.get('cpu', {}).get(l['c_type'])
        if cpu:
            c = cpu
        cmd = "docker run --name {name} -P {image} {command}".format(**locals())
        self.client.create_app(app_id, MarathonApp(cmd=cmd, mem=m, cpus=c))
        self.client.scale_app(app_id, 0, force=True)
        for _ in xrange(POLL_ATTEMPTS):
            if self.client.get_app(self._app_id(name)).tasks_running == 0:
                return
            time.sleep(1)

    def start(self, name):
        """Start a container"""
        self.client.scale_app(self._app_id(name), 1, force=True)
        for _ in xrange(POLL_ATTEMPTS):
            if self.client.get_app(self._app_id(name)).tasks_running == 1:
                break
            time.sleep(1)
        host = self.client.get_app(self._app_id(name)).tasks[0].host
        self._waitforcontainer(host, name)

    def stop(self, name):
        """Stop a container"""
        raise NotImplementedError

    def destroy(self, name):
        """Destroy a container"""
        try:
            host = self.client.get_app(self._app_id(name)).tasks[0].host
            self.client.delete_app(self._app_id(name), force=True)
            self._delete_container(host, name)
        except:
            self.client.delete_app(self._app_id(name), force=True)

    def _get_container_state(self, host, name):
        docker_cli = Client("tcp://{}:2375".format(host), timeout=1200, version='1.17')
        try:
            if docker_cli.inspect_container(name)['State']['Running']:
                return JobState.up
        except:
            return JobState.destroyed

    def _waitforcontainer(self, host, name):
        for _ in xrange(POLL_WAIT):
            if self._get_container_state(host, name) == JobState.up:
                return
            time.sleep(1)
        raise RuntimeError("App container Not Started")

    def _delete_container(self, host, name):
        docker_cli = Client("tcp://{}:2375".format(host), timeout=1200, version='1.17')
        if docker_cli.inspect_container(name)['State']:
            docker_cli.remove_container(name, force=True)

    def run(self, name, image, entrypoint, command):  # noqa
        """Run a one-off command"""
        return self.fleet.run(name, image, entrypoint, command)

    def state(self, name):
        try:
            for _ in xrange(POLL_ATTEMPTS):
                if self.client.get_app(self._app_id(name)).tasks_running == 1:
                    return JobState.up
                elif self.client.get_app(self._app_id(name)).tasks_running == 0:
                    return JobState.created
                time.sleep(1)
        except:
            return JobState.destroyed

    def attach(self, name):
        """
        Attach to a job's stdin, stdout and stderr
        """
        raise NotImplementedError

SchedulerClient = MarathonHTTPClient