This is an archive of the discontinued Mercurial Phabricator instance.

Differential D6924

ci: store job start parameters in DynamoDB
Changes PlannedPublic

Authored by indygreg on Sep 30 2019, 11:57 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Summary

This makes it vastly easier to retry or retrigger a job,
since the instance launch state will be stored in the
database instead of in an ad-hoc SQS message.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Revision Contents
Changeset List

Diff 16718

contrib/ci/lambda_functions/ci.py

schedule_ci(record['repo_url']['S'], record['repo']['S'], schedule_ci(record['repo_url']['S'], record['repo']['S'],
record['node']['S'], bundle_url=bundle_url) record['node']['S'], bundle_url=bundle_url)
def handle_pending_job(event, context): def handle_pending_job(event, context):
"""Handler for starting a job from an SQS message.""" """Handler for starting a job from an SQS message."""
ec2 = boto3.client('ec2') ec2 = boto3.client('ec2')
dynamodb = boto3.resource('dynamodb')
job_table = dynamodb.Table(os.environ['DYNAMODB_JOB_TABLE'])
for record in event['Records']: for record in event['Records']:
body = record['body'] body = record['body']
data = json.loads(body) data = json.loads(body)
user_data_template = data['user_data_template'] job_id = data['job_id']
user_data_params = data['user_data_params']
ec2_instance_config = data['ec2_instance_launch_config']
start_pending_job(ec2, user_data_template, user_data_params, ec2_instance_config) start_pending_job(ec2, job_table, job_id)
def handle_job_result_s3_artifact(event, context): def handle_job_result_s3_artifact(event, context):
"""Handler called when a new S3 object job artifact is uploaded.""" """Handler called when a new S3 object job artifact is uploaded."""
dynamodb = boto3.resource('dynamodb') dynamodb = boto3.resource('dynamodb')
s3 = boto3.resource('s3') s3 = boto3.resource('s3')
job_table = dynamodb.Table(os.environ['DYNAMODB_JOB_TABLE']) job_table = dynamodb.Table(os.environ['DYNAMODB_JOB_TABLE'])
'BlockDeviceMappings': block_device_mappings, 'BlockDeviceMappings': block_device_mappings,
'EbsOptimized': True, 'EbsOptimized': True,
'IamInstanceProfile': {'Name': 'ci-worker'}, 'IamInstanceProfile': {'Name': 'ci-worker'},
'ImageId': image.id, 'ImageId': image.id,
'InstanceType': 'c5.9xlarge', 'InstanceType': 'c5.9xlarge',
'SecurityGroups': ['hg-linux-dev-1'], 'SecurityGroups': ['hg-linux-dev-1'],
} }
job_params = {
'user_data_template': LINUX_USER_DATA,
'user_data_params': user_data_params,
'ec2_instance_launch_config': config,
}
schedule_time = decimal.Decimal(time.time()) schedule_time = decimal.Decimal(time.time())
print('registering job in DynamoDB') print('registering job in DynamoDB')
job_table.put_item(Item={ job_table.put_item(Item={
'job_id': job_id, 'job_id': job_id,
'repo': repo, 'repo': repo,
'node': node, 'node': node,
'job_name': job_name, 'job_name': job_name,
'build_number': build_number, 'build_number': build_number,
'execution_state': 'pending', 'execution_state': 'pending',
'schedule_time': schedule_time, 'schedule_time': schedule_time,
# We encode as JSON because DynamoDB doesn't like some types
# like empty strings.
'start_params': json.dumps({
'user_data_template': LINUX_USER_DATA,
'user_data_params': user_data_params,
'ec2_instance_launch_config': config,
}, sort_keys=True),
}) })
print('adding job to pending queue') print('adding job to pending queue')
sqs.send_message( sqs.send_message(
QueueUrl=sqs_url, QueueUrl=sqs_url,
MessageBody=json.dumps(job_params, sort_keys=True) MessageBody=json.dumps({'job_id': job_id})
) )
def start_pending_job(ec2, user_data_template, user_data_params, ec2_instance_config): def start_pending_job(ec2, job_table, job_id):
"""Called to request the start of a pending job.""" """Called to request the start of a pending job."""
res = job_table.get_item(Key={'job_id': job_id}, ConsistentRead=True)
if 'Item' not in res:
print('unable to find job %s' % job_id)
return
job = res['Item']
start_params = json.loads(job['start_params'])
user_data_template = start_params['user_data_template']
user_data_params = start_params['user_data_params']
ec2_instance_config = start_params['ec2_instance_launch_config']
user_data = user_data_template.format(**user_data_params) user_data = user_data_template.format(**user_data_params)
print('requesting spot instance for job %s' % user_data_params['job_id']) print('requesting spot instance for job %s' % user_data_params['job_id'])
launch_spec = dict(ec2_instance_config) launch_spec = dict(ec2_instance_config)
launch_spec['UserData'] = base64.b64encode(user_data.encode('utf-8')).decode('utf-8') launch_spec['UserData'] = base64.b64encode(user_data.encode('utf-8')).decode('utf-8')
# Spot instances are substantially cheaper but can be terminated at will # Spot instances are substantially cheaper but can be terminated at will

contrib/ci/terraform/job_executor.tf

function_name = "ci-run-pending-job" function_name = "ci-run-pending-job"
description = "Reacts to pending job events on SQS queue" description = "Reacts to pending job events on SQS queue"
filename = data.archive_file.lambda_ci.output_path filename = data.archive_file.lambda_ci.output_path
handler = "ci.handle_pending_job" handler = "ci.handle_pending_job"
source_code_hash = data.archive_file.lambda_ci.output_base64sha256 source_code_hash = data.archive_file.lambda_ci.output_base64sha256
runtime = "python3.7" runtime = "python3.7"
timeout = 60 timeout = 60
role = aws_iam_role.lambda_ci_run_pending_job.arn role = aws_iam_role.lambda_ci_run_pending_job.arn
environment {
variables = {
DYNAMODB_JOB_TABLE = aws_dynamodb_table.ci_job.name
}
}
} }
resource "aws_cloudwatch_log_group" "lambda_ci_run_pending_job" { resource "aws_cloudwatch_log_group" "lambda_ci_run_pending_job" {
name = "/aws/lambda/${aws_lambda_function.ci_run_pending_job.function_name}" name = "/aws/lambda/${aws_lambda_function.ci_run_pending_job.function_name}"
retention_in_days = 7 retention_in_days = 7
} }
data "aws_iam_policy_document" "lambda_ci_run_pending_job" { data "aws_iam_policy_document" "lambda_ci_run_pending_job" {
statement { statement {
effect = "Allow" effect = "Allow"
actions = [ actions = [
"ec2:*", "ec2:*",
"iam:*", "iam:*",
] ]
resources = ["*"] resources = ["*"]
} }
# Allow querying job state in DynamoDB.
statement {
effect = "Allow"
actions = [
"dynamodb:GetItem",
]
resources = [
aws_dynamodb_table.ci_job.arn,
]
}
} }
resource "aws_iam_role_policy" "lambda_ci_handle_pending_job" { resource "aws_iam_role_policy" "lambda_ci_handle_pending_job" {
role = aws_iam_role.lambda_ci_run_pending_job.name role = aws_iam_role.lambda_ci_run_pending_job.name
name = aws_iam_role.lambda_ci_run_pending_job.name name = aws_iam_role.lambda_ci_run_pending_job.name
policy = data.aws_iam_policy_document.lambda_ci_run_pending_job.json policy = data.aws_iam_policy_document.lambda_ci_run_pending_job.json
} }