UnknownTaskDefinitionError

Completed

Pull Request — develop (#19)

by Fabian

created 2017-06-25 14:38 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	2
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
wmc	0
c	1
b	0
f	0
dl	0
loc	2
ccs	2
cts	2
cp	1

from datetime import datetime
from json import dumps

import boto3
from botocore.exceptions import ClientError, NoCredentialsError


class EcsClient(object):
    def __init__(self, access_key_id=None, secret_access_key=None, region=None, profile=None):
        session = boto3.session.Session(aws_access_key_id=access_key_id,
                                        aws_secret_access_key=secret_access_key,
                                        region_name=region,
                                        profile_name=profile)
        self.boto = session.client(u'ecs')

    def describe_services(self, cluster_name, service_name):
        return self.boto.describe_services(cluster=cluster_name, services=[service_name])

    def describe_task_definition(self, task_definition_arn):
        try:
            return self.boto.describe_task_definition(taskDefinition=task_definition_arn)
        except ClientError:
            raise UnknownTaskDefinitionError('Unknown task definition arn: %s' % task_definition_arn)

    def list_tasks(self, cluster_name, service_name):
        return self.boto.list_tasks(cluster=cluster_name, serviceName=service_name)

    def describe_tasks(self, cluster_name, task_arns):
        return self.boto.describe_tasks(cluster=cluster_name, tasks=task_arns)

    def register_task_definition(self, family, containers, volumes, role_arn):
        return self.boto.register_task_definition(
            family=family,
            containerDefinitions=containers,
            volumes=volumes,
            taskRoleArn=role_arn or ''
        )

    def deregister_task_definition(self, task_definition_arn):
        return self.boto.deregister_task_definition(taskDefinition=task_definition_arn)

    def update_service(self, cluster, service, desired_count, task_definition):
        return self.boto.update_service(
            cluster=cluster,
            service=service,
            desiredCount=desired_count,
            taskDefinition=task_definition
        )

    def run_task(self, cluster, task_definition, count, started_by, overrides):
        return self.boto.run_task(
            cluster=cluster,
            taskDefinition=task_definition,
            count=count,
            startedBy=started_by,
            overrides=overrides
        )


class EcsService(dict):
    def __init__(self, cluster, iterable=None, **kwargs):
        self._cluster = cluster
        super(EcsService, self).__init__(iterable, **kwargs)

    def set_desired_count(self, desired_count):
        self[u'desiredCount'] = desired_count

    def set_task_definition(self, task_definition):
        self[u'taskDefinition'] = task_definition.arn

    @property
    def cluster(self):
        return self._cluster

    @property
    def name(self):
        return self.get(u'serviceName')

    @property
    def task_definition(self):
        return self.get(u'taskDefinition')

    @property
    def desired_count(self):
        return self.get(u'desiredCount')

    @property
    def deployment_created_at(self):
        for deployment in self.get(u'deployments'):
            if deployment.get(u'status') == u'PRIMARY':
                return deployment.get(u'createdAt')
        return datetime.now()

    @property
    def deployment_updated_at(self):
        for deployment in self.get(u'deployments'):
            if deployment.get(u'status') == u'PRIMARY':
                return deployment.get(u'updatedAt')
        return datetime.now()

    @property
    def errors(self):
        errors = {}
        for event in self.get('events'):
            if u'unable' in event[u'message'] and event[u'createdAt'] >= self.deployment_updated_at:
                errors[event[u'createdAt'].isoformat()] = 'ERROR: %s' % event[u'message']
        return errors

    @property
    def older_errors(self):
        errors = {}
        for event in self.get('events'):
            if u'unable' in event[u'message'] and \
                                    self.deployment_created_at <= event[u'createdAt'] <= self.deployment_updated_at:
                errors[event[u'createdAt'].isoformat()] = 'ERROR: %s' % event[u'message']
        return errors


class EcsTaskDefinition(dict):
    def __init__(self, iterable=None, **kwargs):
        super(EcsTaskDefinition, self).__init__(iterable, **kwargs)
        self._diff = []

    @property
    def containers(self):
        return self.get(u'containerDefinitions')

    @property
    def container_names(self):
        for container in self.get(u'containerDefinitions'):
            yield container[u'name']

    @property
    def volumes(self):
        return self.get(u'volumes')

    @property
    def arn(self):
        return self.get(u'taskDefinitionArn')

    @property
    def family(self):
        return self.get(u'family')

    @property
    def role_arn(self):
        return self.get(u'taskRoleArn')

    @property
    def revision(self):
        return self.get(u'revision')

    @property
    def family_revision(self):
        return '%s:%d' % (self.get(u'family'), self.get(u'revision'))

    @property
    def diff(self):
        return self._diff

    def get_overrides(self):
        override = dict()
        overrides = []
        for diff in self.diff:
            if override.get('name') != diff.container:
                override = dict(name=diff.container)
                overrides.append(override)
            if diff.field == 'command':
                override['command'] = self.get_overrides_command(diff.value)
            elif diff.field == 'environment':
                override['environment'] = self.get_overrides_environment(diff.value)
        return overrides

    def get_overrides_command(self, command):
        return command.split(' ')

    def get_overrides_environment(self, environment_dict):
        return [{"name": e, "value": environment_dict[e]} for e in environment_dict]

    def set_images(self, tag=None, **images):
        self.validate_container_options(**images)
        for container in self.containers:
            if container[u'name'] in images:
                new_image = images[container[u'name']]
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
                self._diff.append(diff)
                container[u'image'] = new_image
            elif tag:
                image_definition = container[u'image'].rsplit(u':', 1)
                new_image = u'%s:%s' % (image_definition[0], tag.strip())
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
                self._diff.append(diff)
                container[u'image'] = new_image

    def set_commands(self, **commands):
        self.validate_container_options(**commands)
        for container in self.containers:
            if container[u'name'] in commands:
                new_command = commands[container[u'name']]
                diff = EcsTaskDefinitionDiff(container[u'name'], u'command', new_command, container.get(u'command'))
                self._diff.append(diff)
                container[u'command'] = [new_command]

    def set_environment(self, environment_list):
        environment = {}

        for env in environment_list:
            environment.setdefault(env[0], {})
            environment[env[0]][env[1]] = env[2]

        self.validate_container_options(**environment)
        for container in self.containers:
            if container[u'name'] in environment:
                self.apply_container_environment(container, environment[container[u'name']])

    def apply_container_environment(self, container, new_environment):
        old_environment = {env['name']: env['value'] for env in container.get('environment', {})}
        merged_environment = old_environment.copy()
        merged_environment.update(new_environment)

        diff = EcsTaskDefinitionDiff(container[u'name'], u'environment', merged_environment, old_environment)
        self._diff.append(diff)

        container[u'environment'] = [{"name": e, "value": merged_environment[e]} for e in merged_environment]

    def validate_container_options(self, **container_options):
        for container_name in container_options:
            if container_name not in self.container_names:
                raise UnknownContainerError(u'Unknown container: %s' % container_name)

    def set_role_arn(self, role_arn):
        if role_arn:
            diff = EcsTaskDefinitionDiff(None, u'role_arn', role_arn, self[u'taskRoleArn'])
            self[u'taskRoleArn'] = role_arn
            self._diff.append(diff)


class EcsTaskDefinitionDiff(object):
    def __init__(self, container, field, value, old_value):
        self.container = container
        self.field = field
        self.value = value
        self.old_value = old_value

    def __repr__(self):
        if self.container:
            return u"Changed %s of container '%s' to: %s (was: %s)" % (
                self.field,
                self.container,
                dumps(self.value),
                dumps(self.old_value)
            )
        else:
            return u"Changed %s to: %s (was: %s)" % (
                self.field,
                dumps(self.value),
                dumps(self.old_value)
            )



class EcsAction(object):
    def __init__(self, client, cluster_name, service_name):
        self._client = client
        self._cluster_name = cluster_name
        self._service_name = service_name

        try:
            self._service = self.get_service()
        except IndexError:
            raise ConnectionError(u'An error occurred when calling the DescribeServices operation: Service not found.')
        except ClientError as e:
            raise ConnectionError(str(e))
        except NoCredentialsError:
            raise ConnectionError(u'Unable to locate credentials. Configure credentials by running "aws configure".')

    def get_service(self):
        services_definition = self._client.describe_services(self._cluster_name, self._service_name)
        return EcsService(self._cluster_name, services_definition[u'services'][0])

    def get_current_task_definition(self, service):
        task_definition_payload = self._client.describe_task_definition(service.task_definition)
        task_definition = EcsTaskDefinition(task_definition_payload[u'taskDefinition'])
        return task_definition

    def get_task_definition(self, task_definition):
        task_definition_payload = self._client.describe_task_definition(task_definition)
        task_definition = EcsTaskDefinition(task_definition_payload[u'taskDefinition'])
        return task_definition

    def update_task_definition(self, task_definition):
        response = self._client.register_task_definition(
            task_definition.family,
            task_definition.containers,
            task_definition.volumes,
            task_definition.role_arn
        )
        new_task_definition = EcsTaskDefinition(response[u'taskDefinition'])
        self._client.deregister_task_definition(task_definition.arn)
        return new_task_definition

    def update_service(self, service):
        response = self._client.update_service(service.cluster, service.name, service.desired_count,
                                               service.task_definition)
        return EcsService(self._cluster_name, response[u'service'])

    def is_deployed(self, service):
        if len(service[u'deployments']) != 1:
            return False
        running_tasks = self._client.list_tasks(service.cluster, service.name)
        if not running_tasks[u'taskArns']:
            return service.desired_count == 0
        return service.desired_count == self.get_running_tasks_count(service, running_tasks[u'taskArns'])

    def get_running_tasks_count(self, service, task_arns):
        running_count = 0
        tasks_details = self._client.describe_tasks(self._cluster_name, task_arns)
        for task in tasks_details[u'tasks']:
            if task[u'taskDefinitionArn'] == service.task_definition and task[u'lastStatus'] == u'RUNNING':
                running_count += 1
        return running_count

    @property
    def client(self):
        return self._client

    @property
    def service(self):
        return self._service

    @property
    def cluster_name(self):
        return self._cluster_name

    @property
    def service_name(self):
        return self._service_name


class DeployAction(EcsAction):
    def deploy(self, task_definition):
        self._service.set_task_definition(task_definition)
        return self.update_service(self._service)


class ScaleAction(EcsAction):
    def scale(self, desired_count):
        self._service.set_desired_count(desired_count)
        return self.update_service(self._service)


class RunAction(EcsAction):
    def __init__(self, client, cluster_name):
        self._client = client
        self._cluster_name = cluster_name
        self.started_tasks = []

    def run(self, task_definition, count, started_by):
        result = self._client.run_task(
            cluster=self._cluster_name,
            task_definition=task_definition.family_revision,
            count=count,
            started_by=started_by,
            overrides=dict(containerOverrides=task_definition.get_overrides())
        )
        self.started_tasks = result['tasks']
        return True


class EcsError(Exception):
    pass


class ConnectionError(EcsError):
    pass


class UnknownContainerError(EcsError):
    pass


class UnknownTaskDefinitionError(EcsError):
    pass


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

fabfuel / ecs-deploy

Pull Request — develop (#19)

UnknownTaskDefinitionError

Complexity

Size/Duplication

Test Coverage

Importance

Duplication Side-by-Side

Filter issues like