Completed
Branch master (df7763)
by Fabian
06:23
created

RunAction.run()   A

Complexity

Conditions 1

Size

Total Lines 10

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 4
CRAP Score 1

Importance

Changes 1
Bugs 0 Features 0
Metric Value
c 1
b 0
f 0
dl 0
loc 10
ccs 4
cts 4
cp 1
rs 9.4285
cc 1
crap 1
1 1
from datetime import datetime
2 1
from json import dumps
3
4 1
import boto3
5 1
from botocore.exceptions import ClientError, NoCredentialsError
6
7
8 1
class EcsClient(object):
9 1
    def __init__(self, access_key_id=None, secret_access_key=None, region=None, profile=None):
10 1
        session = boto3.session.Session(aws_access_key_id=access_key_id,
11
                                        aws_secret_access_key=secret_access_key,
12
                                        region_name=region,
13
                                        profile_name=profile)
14 1
        self.boto = session.client(u'ecs')
15
16 1
    def describe_services(self, cluster_name, service_name):
17 1
        return self.boto.describe_services(cluster=cluster_name, services=[service_name])
18
19 1
    def describe_task_definition(self, task_definition_arn):
20 1
        return self.boto.describe_task_definition(taskDefinition=task_definition_arn)
21
22 1
    def list_tasks(self, cluster_name, service_name):
23 1
        return self.boto.list_tasks(cluster=cluster_name, serviceName=service_name)
24
25 1
    def describe_tasks(self, cluster_name, task_arns):
26 1
        return self.boto.describe_tasks(cluster=cluster_name, tasks=task_arns)
27
28 1
    def register_task_definition(self, family, containers, volumes):
29 1
        return self.boto.register_task_definition(family=family, containerDefinitions=containers, volumes=volumes)
30
31 1
    def deregister_task_definition(self, task_definition_arn):
32 1
        return self.boto.deregister_task_definition(taskDefinition=task_definition_arn)
33
34 1
    def update_service(self, cluster, service, desired_count, task_definition):
35 1
        return self.boto.update_service(
36
            cluster=cluster,
37
            service=service,
38
            desiredCount=desired_count,
39
            taskDefinition=task_definition
40
        )
41
42 1
    def run_task(self, cluster, task_definition, count, started_by, overrides):
43 1
        return self.boto.run_task(
44
            cluster=cluster,
45
            taskDefinition=task_definition,
46
            count=count,
47
            startedBy=started_by,
48
            overrides=overrides
49
        )
50
51
52 1
class EcsService(dict):
53 1
    def __init__(self, cluster, iterable=None, **kwargs):
54 1
        self._cluster = cluster
55 1
        super(EcsService, self).__init__(iterable, **kwargs)
56
57 1
    def set_desired_count(self, desired_count):
58 1
        self[u'desiredCount'] = desired_count
59
60 1
    def set_task_definition(self, task_definition):
61 1
        self[u'taskDefinition'] = task_definition.arn
62
63 1
    @property
64
    def cluster(self):
65 1
        return self._cluster
66
67 1
    @property
68
    def name(self):
69 1
        return self.get(u'serviceName')
70
71 1
    @property
72
    def task_definition(self):
73 1
        return self.get(u'taskDefinition')
74
75 1
    @property
76
    def desired_count(self):
77 1
        return self.get(u'desiredCount')
78
79 1
    @property
80
    def deployment_created_at(self):
81 1
        for deployment in self.get(u'deployments'):
82 1
            if deployment.get(u'status') == u'PRIMARY':
83 1
                return deployment.get(u'createdAt')
84 1
        return datetime.now()
85
86 1
    @property
87
    def deployment_updated_at(self):
88 1
        for deployment in self.get(u'deployments'):
89 1
            if deployment.get(u'status') == u'PRIMARY':
90 1
                return deployment.get(u'updatedAt')
91 1
        return datetime.now()
92
93 1
    @property
94
    def errors(self):
95 1
        errors = {}
96 1
        for event in self.get('events'):
97 1
            if u'unable' in event[u'message'] and event[u'createdAt'] >= self.deployment_updated_at:
98 1
                errors[event[u'createdAt'].isoformat()] = 'ERROR: %s' % event[u'message']
99 1
        return errors
100
101 1
    @property
102
    def older_errors(self):
103 1
        errors = {}
104 1
        for event in self.get('events'):
105 1
            if u'unable' in event[u'message'] and \
106
                                    self.deployment_created_at <= event[u'createdAt'] <= self.deployment_updated_at:
107 1
                errors[event[u'createdAt'].isoformat()] = 'ERROR: %s' % event[u'message']
108 1
        return errors
109
110
111 1
class EcsTaskDefinition(dict):
112 1
    def __init__(self, iterable=None, **kwargs):
113 1
        super(EcsTaskDefinition, self).__init__(iterable, **kwargs)
114 1
        self._diff = []
115
116 1
    @property
117
    def containers(self):
118 1
        return self.get(u'containerDefinitions')
119
120 1
    @property
121
    def container_names(self):
122 1
        for container in self.get(u'containerDefinitions'):
123 1
            yield container[u'name']
124
125 1
    @property
126
    def volumes(self):
127 1
        return self.get(u'volumes')
128
129 1
    @property
130
    def arn(self):
131 1
        return self.get(u'taskDefinitionArn')
132
133 1
    @property
134
    def family(self):
135 1
        return self.get(u'family')
136
137 1
    @property
138
    def revision(self):
139 1
        return self.get(u'revision')
140
141 1
    @property
142
    def family_revision(self):
143 1
        return '%s:%d' % (self.get(u'family'), self.get(u'revision'))
144
145 1
    @property
146
    def diff(self):
147 1
        return self._diff
148
149 1
    def get_overrides(self):
150 1
        override = dict()
151 1
        overrides = []
152 1
        for diff in self.diff:
153 1
            if override.get('name') != diff.container:
154 1
                override = dict(name=diff.container)
155 1
                overrides.append(override)
156 1
            if diff.field == 'command':
157 1
                override['command'] = self.get_overrides_command(diff.value)
158 1
            elif diff.field == 'environment':
159 1
                override['environment'] = self.get_overrides_environment(diff.value)
160 1
        return overrides
161
162 1
    def get_overrides_command(self, command):
163 1
        return command.split(' ')
164
165 1
    def get_overrides_environment(self, environment_dict):
166 1
        return [{"name": e, "value": environment_dict[e]} for e in environment_dict]
167
168 1
    def set_images(self, tag=None, **images):
169 1
        self.validate_container_options(**images)
170 1
        for container in self.containers:
171 1
            if container[u'name'] in images:
172 1
                new_image = images[container[u'name']]
173 1
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
174 1
                self._diff.append(diff)
175 1
                container[u'image'] = new_image
176 1
            elif tag:
177 1
                image_definition = container[u'image'].rsplit(u':', 1)
178 1
                new_image = u'%s:%s' % (image_definition[0], tag.strip())
179 1
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
180 1
                self._diff.append(diff)
181 1
                container[u'image'] = new_image
182
183 1
    def set_commands(self, **commands):
184 1
        self.validate_container_options(**commands)
185 1
        for container in self.containers:
186 1
            if container[u'name'] in commands:
187 1
                new_command = commands[container[u'name']]
188 1
                diff = EcsTaskDefinitionDiff(container[u'name'], u'command', new_command, container.get(u'command'))
189 1
                self._diff.append(diff)
190 1
                container[u'command'] = [new_command]
191
192 1
    def set_environment(self, environment_list):
193 1
        environment = {}
194
195 1
        for env in environment_list:
196 1
            environment.setdefault(env[0], {})
197 1
            environment[env[0]][env[1]] = env[2]
198
199 1
        self.validate_container_options(**environment)
200 1
        for container in self.containers:
201 1
            if container[u'name'] in environment:
202 1
                self.apply_container_environment(container, environment[container[u'name']])
203
204 1
    def apply_container_environment(self, container, new_environment):
205 1
        old_environment = {env['name']: env['value'] for env in container.get('environment', {})}
206 1
        merged_environment = old_environment.copy()
207 1
        merged_environment.update(new_environment)
208
209 1
        diff = EcsTaskDefinitionDiff(container[u'name'], u'environment', merged_environment, old_environment)
210 1
        self._diff.append(diff)
211
212 1
        container[u'environment'] = [{"name": e, "value": merged_environment[e]} for e in merged_environment]
213
214 1
    def validate_container_options(self, **container_options):
215 1
        for container_name in container_options:
216 1
            if container_name not in self.container_names:
217 1
                raise UnknownContainerError(u'Unknown container: %s' % container_name)
218
219
220 1
class EcsTaskDefinitionDiff(object):
221 1
    def __init__(self, container, field, value, old_value):
222 1
        self.container = container
223 1
        self.field = field
224 1
        self.value = value
225 1
        self.old_value = old_value
226
227 1
    def __repr__(self):
228 1
        return u"Changed %s of container '%s' to: %s (was: %s)" % \
229
               (self.field, self.container, dumps(self.value), dumps(self.old_value))
230
231
232 1
class EcsAction(object):
233 1
    def __init__(self, client, cluster_name, service_name):
234 1
        self._client = client
235 1
        self._cluster_name = cluster_name
236 1
        self._service_name = service_name
237
238 1
        try:
239 1
            self._service = self.get_service()
240 1
        except IndexError:
241 1
            raise ConnectionError(u'An error occurred when calling the DescribeServices operation: Service not found.')
242 1
        except ClientError as e:
243 1
            raise ConnectionError(str(e))
244 1
        except NoCredentialsError:
245 1
            raise ConnectionError(u'Unable to locate credentials. Configure credentials by running "aws configure".')
246
247 1
    def get_service(self):
248 1
        services_definition = self._client.describe_services(self._cluster_name, self._service_name)
249 1
        return EcsService(self._cluster_name, services_definition[u'services'][0])
250
251 1
    def get_current_task_definition(self, service):
252 1
        task_definition_payload = self._client.describe_task_definition(service.task_definition)
253 1
        task_definition = EcsTaskDefinition(task_definition_payload[u'taskDefinition'])
254 1
        return task_definition
255
256 1
    def get_task_definition(self, task_definition):
257 1
        task_definition_payload = self._client.describe_task_definition(task_definition)
258 1
        task_definition = EcsTaskDefinition(task_definition_payload[u'taskDefinition'])
259 1
        return task_definition
260
261 1
    def update_task_definition(self, task_definition):
262 1
        response = self._client.register_task_definition(task_definition.family, task_definition.containers,
263
                                                         task_definition.volumes)
264 1
        new_task_definition = EcsTaskDefinition(response[u'taskDefinition'])
265 1
        self._client.deregister_task_definition(task_definition.arn)
266 1
        return new_task_definition
267
268 1
    def update_service(self, service):
269 1
        response = self._client.update_service(service.cluster, service.name, service.desired_count,
270
                                               service.task_definition)
271 1
        return EcsService(self._cluster_name, response[u'service'])
272
273 1
    def is_deployed(self, service):
274 1
        if len(service[u'deployments']) != 1:
275 1
            return False
276 1
        running_tasks = self._client.list_tasks(service.cluster, service.name)
277 1
        if not running_tasks[u'taskArns']:
278 1
            return service.desired_count == 0
279 1
        return service.desired_count == self.get_running_tasks_count(service, running_tasks[u'taskArns'])
280
281 1
    def get_running_tasks_count(self, service, task_arns):
282 1
        running_count = 0
283 1
        tasks_details = self._client.describe_tasks(self._cluster_name, task_arns)
284 1
        for task in tasks_details[u'tasks']:
285 1
            if task[u'taskDefinitionArn'] == service.task_definition and task[u'lastStatus'] == u'RUNNING':
286 1
                running_count += 1
287 1
        return running_count
288
289 1
    @property
290
    def client(self):
291 1
        return self._client
292
293 1
    @property
294
    def service(self):
295 1
        return self._service
296
297 1
    @property
298
    def cluster_name(self):
299 1
        return self._cluster_name
300
301 1
    @property
302
    def service_name(self):
303 1
        return self._service_name
304
305
306 1
class DeployAction(EcsAction):
307 1
    def deploy(self, task_definition):
308 1
        self._service.set_task_definition(task_definition)
309 1
        return self.update_service(self._service)
310
311
312 1
class ScaleAction(EcsAction):
313 1
    def scale(self, desired_count):
314 1
        self._service.set_desired_count(desired_count)
315 1
        return self.update_service(self._service)
316
317
318 1
class RunAction(EcsAction):
319 1
    def __init__(self, client, cluster_name):
320 1
        self._client = client
321 1
        self._cluster_name = cluster_name
322 1
        self.started_tasks = []
323
324 1
    def run(self, task_definition, count, started_by):
325 1
        result = self._client.run_task(
326
            cluster=self._cluster_name,
327
            task_definition=task_definition.family_revision,
328
            count=count,
329
            started_by=started_by,
330
            overrides=dict(containerOverrides=task_definition.get_overrides())
331
        )
332 1
        self.started_tasks = result['tasks']
333 1
        return True
334
335
336 1
class EcsError(Exception):
337 1
    pass
338
339
340 1
class ConnectionError(EcsError):
341 1
    pass
342
343
344 1
class UnknownContainerError(EcsError):
345
    pass
346