[PYTHON] I want to set a life cycle in the task definition of ECS
Introduction
When writing batch processing etc. using the task schedule function of ECS, if the task is updated many times using ʻecs-cli, the revision` of the task definition will increase.
It is troublesome to delete the revision (unregister) by clicking from the console screen of aws, and there is also a method to delete it using deregister-task-definition of ʻawscli`, but it is also troublesome to execute it from the command line. ..
How to delete with awscli https://docs.aws.amazon.com/cli/latest/reference/ecs/deregister-task-definition.html
So I decided to run Lambda to unregister (unregister) the revision.
What was used
--For deploying ServerlessFrameWork lambda --serverless-python-requirements For deploying external Python packages --Pipenv package management
Get a list of ECS task definitions
Use ecs.list_task_definitions in boto3 to get all the task definitions.
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs.html#ECS.Client.list_task_definitions
I set sort ='DESC' as an argument, but I set it to 'DESC' because it is necessary when using ʻitertools.groupby` later.
import json
import boto3
import logging
logger = logging.Logger(__name__)
logger.setLevel(logging.DEBUG)
def get_task_definitions():
"""Get ECS task definition
Returns:
list --Task definition arn list
"""
client = boto3.client('ecs')
try:
res = client.list_task_definitions(
status='ACTIVE',
sort='DESC')
logger.info(json.dumps(res))
except Exception as e:
logger.info(e)
else:
logger.info('Successful task definition acquisition')
list_task_definitions = res['taskDefinitionArns']
return list_task_definitions
You can get it like this
[
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_a:3',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_a:2',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_a:1',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_b:5',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_b:4',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_b:3',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_b:2',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_b:1',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_c:5',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_c:4',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_c:3',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_c:2',
'arn:aws:ecs:ap-northeast-1:XXXXXXXXXXXXX:task-definition/task_c:1'
]
Format task definition data
Now that we have the task definition, we will group it by the same task name and format the data so that we can see how many revisions the task has. I got the task definition, but to be exact, it is arn of the task definition, so I format it only to the task name.
#The shape you want to shape
{
'task_a': ['3', '2', '1'],
'task_b': ['5', '4', '3', '2', '1'],
'task_c': ['5', '4', '3', '2', '1']
}
ʻUse itertools.groupby to group them. I set sort in ʻecs.list_task_definitions because I need to sort in advance to use ʻitertools.groupby`.
a = [('a', '3'), ('a', '2'), ('a', '1'), ('b', '4'), ('b', '3'),
('b', '2'), ('b', '1'), ('c', '2'), ('c', '1')]
for key, group in groupby(a, lambda x: x[0]):
for team in group:
print(team)
print('')
It will be grouped like this, so I will format it into a dictionary.
('a', '3')
('a', '2')
('a', '1')
('b', '4')
('b', '3')
('b', '2')
('b', '1')
('c', '2')
('c', '1')
from itertools import groupby
def groupby_task_definitions(list_task_definitions):
"""Put together a list of arns in the task definition
Arguments:
list_task_definitions {list} --Task definition arn list
Returns:
dict --Task name and revision list
Example:
{
task_a: ['4', '3', '2', '1'],
task_b: ['2', 1]
}
"""
group_list = []
for task_definition in list_task_definitions:
#Cut only the task name and revision
task_definition = task_definition.rsplit('/', 1)[1]
task_name, revision = task_definition.split(':')
#Add to list as tuple
group_list.append((task_name, revision))
result_dict = {}
for key, group in groupby(group_list, lambda x: x[0]):
revision_list = []
for _, v in group:
revision_list.append(v)
result_dict[key] = revision_list
return result_dict
Delete ECS task (unregister)
Now that we have created a dictionary of task definitions, we will delete the ECS tasks.
Turn the dictionary with for and delete it withpop ()until the length of the obtained list of revisions reaches the specified number.
ʻEcs.deregister_task_definition` can be executed easily by giving the name of the task definition to the argument.
def deregister_task_definition(groupby_task_definitions_dict, leave):
"""Delete the task definition, leaving up to the two latest revisions
Arguments:
groupby_task_definitions_dict {dict} --A dict that stores a list of task names and revisions
"""
client = boto3.client('ecs')
for name, revision_list in groupby_task_definitions_dict.items():
logger.info(name)
try:
while len(revision_list) > 2:
revision = revision_list.pop()
client.deregister_task_definition(
taskDefinition='{}:{}'.format(name, revision)
)
except Exception as e:
logger.error(e)
else:
logger.info('{} : OK!'.format(name))
Make it available on Lambda
We haven't set any events at this stage, but you can use lambda invoke or CloudWatch Events on a regular basis.
serverless.yml
We will create Lambda using Serverless Framework, so expose serverless.yml.
serverless.yml
service: ecs-task
provider:
name: aws
runtime: python3.7
region: ap-northeast-1
logRetentionInDays: 30
iamRoleStatements:
- Effect: Allow
Action:
- ecs:*
Resource: "*"
functions:
definition-lifecycle :
handler: handler.main
plugins:
- serverless-python-requirements
custom:
pythonRequirements:
usePipenv: true
handler The handler is just executing a function that does what it has done so far.
def main(event, context):
list_task_definitions = get_task_definitions()
groupby_task_definitions_dict = groupby_task_definitions(
list_task_definitions)
logger.info(groupby_task_definitions_dict)
deregister_task_definition(groupby_task_definitions_dict, leave)
Whole code
handler.py
import boto3
import json
from itertools import groupby
import logging
logger = logging.Logger(__name__)
logger.setLevel(logging.DEBUG)
def get_task_definitions():
"""Get ECS task definition
Returns:
list --Task definition arn list
"""
client = boto3.client('ecs')
try:
res = client.list_task_definitions(
status='ACTIVE',
sort='DESC')
logger.info(json.dumps(res))
except Exception as e:
logger.info(e)
else:
logger.info('Successful task definition acquisition')
list_task_definitions = res['taskDefinitionArns']
return list_task_definitions
def groupby_task_definitions(list_task_definitions):
"""Put together a list of arns in the task definition
Arguments:
list_task_definitions {list} --Task definition arn list
Returns:
dict --Task name and revision list
Example:
{
task_a: ['4', '3', '2', '1'],
task_b: ['2', 1]
}
"""
group_list = []
for task_definition in list_task_definitions:
#Cut only the task name and revision
task_definition = task_definition.rsplit('/', 1)[1]
task_name, revision = task_definition.split(':')
group_list.append((task_name, revision))
result_dict = {}
for key, group in groupby(group_list, lambda x: x[0]):
revision_list = []
for _, v in list(group):
revision_list.append(v)
result_dict[key] = revision_list
return result_dict
def deregister_task_definition(groupby_task_definitions_dict):
"""Delete the task definition, leaving up to the two latest revisions
Arguments:
groupby_task_definitions_dict {dict} --A dict that stores a list of task names and revisions
"""
client = boto3.client('ecs')
for name, revision_list in groupby_task_definitions_dict.items():
logger.info(name)
try:
while len(revision_list) > 2:
revision = revision_list.pop()
client.deregister_task_definition(
taskDefinition='{}:{}'.format(name, revision)
)
except Exception as e:
logger.error(e)
else:
logger.info('{} : OK!'.format(name))
def main(event, context):
list_task_definitions = get_task_definitions()
groupby_task_definitions_dict = groupby_task_definitions(
list_task_definitions)
logger.info(groupby_task_definitions_dict)
deregister_task_definition(groupby_task_definitions_dict)
AWS somehow
You can delete (unregister) a task definition only for tasks that are ʻActive, and then go into the ʻInactive state. In that state, it cannot be completely erased at this point, so we are waiting for AWS's response.
At this time, the INACTIVE task definition remains discoverable in your account indefinitely. However, this behavior is subject to change in the future, so make sure that you do not rely on the ACL task definition that is retained across the life cycle of the associated task and service.
https://docs.aws.amazon.com/ja_jp/AmazonECS/latest/developerguide/deregister-task-definition.html
Recommended Posts