This is an archive of the discontinued Mercurial Phabricator instance.

Differential D6926

ci: retry expired spot instance requests
Changes PlannedPublic

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

Details

Reviewers
None
Group Reviewers
hg-reviewers
Summary

EC2 spot instance requests can fail. Often the root cause is no
availability of spot instances in the given availability zone.

This commit teaches our CI system to recognize when a spot instance
request has failed and to retry it.

To do this, we create a pair of new Lambda functions, "spot instance
request monitor" and "start job." The former is invoked periodically
via CloudWatch Event. It scans all jobs waiting on a spot instance
request. If the spot instance request is cancelled, it calls into
the "start job" function, which simply calls an internal API for
trying to start a job given a job ID.

Our strategy for retrying spot instance requests is to try to use
the "next" availability zone for the given region. We will cycle
through availability zones until a spot instance request is
acted on. In my testing, this always works. Although it could
take several minutes to cycle through all availability zones and
for an availability zone to have instance availability.

There are definitely other strategies we could try. For example,
we could try another AWS region. Or we could launch an on-demand
instance after N failures. Or we could launch a difference EC2
instance type - one with hopefully more availability. Or we could
use spot fleet requests (which allow launching from multiple
instance types). The easiest is probably to launch an on-demand
instance. But until we need this functionality, I'm inclined to
not build it, as I like the consistency of always using spot
instances.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Revision Contents
Changeset List

Diff 16723

contrib/ci/README.rst

This component also contains an *instance state change* Lambda function. This component also contains an *instance state change* Lambda function.
This function is invoked whenever an EC2 instance's state change. e.g. This function is invoked whenever an EC2 instance's state change. e.g.
when a new instance is created, stops, or terminates. The role of this when a new instance is created, stops, or terminates. The role of this
function is to keep accounting for running jobs and instances up-to-date. function is to keep accounting for running jobs and instances up-to-date.
This function will update job records in DynamoDB to record that a This function will update job records in DynamoDB to record that a
job has started/finished/aborted. job has started/finished/aborted.
This component contains a *start job* Lambda function, which can be
invoked with the Job ID of a job that someone wishes to start. It
tries to ensure the job has run. In the case of an expired spot instance
request, it will try again.
The Terraform code for this component lives in ``job_executor.tf``. The Terraform code for this component lives in ``job_executor.tf``.
Spot Instance Request Monitor
-----------------------------
The *spot instance request monitor* is a component for monitoring the
state of spot instance requests.
We utilize EC2 spot instances to run jobs. Spot instances, unlike
on-demand instances, can't be launched directly. Instead, you create
a request for spot instances and this request is eventually fulfilled
by EC2, if possible. Often, the request is fulfilled immediately and
an EC2 instance launches within a few seconds.
But sometimes a spot instance request fails. This is often due to no
spot instances being available at that time. This is where this component
plays a part.
The *spot instance request monitor* is a Lambda function that is
periodically invoked via a CloudWatch Event. It scans all jobs currently
waiting on a spot instance request. If the spot instance request
couldn't be fulfilled, it calls out to the *start job* Lambda function
to tell it to try to reschedule it. Theoretically, the job should
eventually start.
Worker Worker
------ ------
The *worker* component is the entity doing compute-heavy work. It is The *worker* component is the entity doing compute-heavy work. It is
essentially a single EC2 instance. essentially a single EC2 instance.
The Terraform code for this component lives in ``worker.tf``. The Terraform code for this component lives in ``worker.tf``.

contrib/ci/lambda_functions/ci.py

body = record['body'] body = record['body']
data = json.loads(body) data = json.loads(body)
job_id = data['job_id'] job_id = data['job_id']
start_pending_job(ec2, job_table, job_id) start_pending_job(ec2, job_table, job_id)
def handle_start_job(event, context):
"""Handler for ci-start-job function."""
ec2 = boto3.client('ec2')
dynamodb = boto3.resource('dynamodb')
job_table = dynamodb.Table(os.environ['DYNAMODB_JOB_TABLE'])
job_id = event['job_id']
print('received request to start job %s' % job_id)
start_pending_job(ec2, job_table, job_id)
def handle_spot_instance_request_monitor(event, context):
"""Handler to invoke spot instance request monitor."""
ec2 = boto3.client('ec2')
dynamodb = boto3.resource('dynamodb')
lambda_client = boto3.client('lambda')
job_table = dynamodb.Table(os.environ['DYNAMODB_JOB_TABLE'])
start_job_function = os.environ['LAMBDA_START_JOB_FUNCTION']
monitor_spot_instance_requests(ec2, job_table, lambda_client,
start_job_function)
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'])
test_result_table = dynamodb.Table(os.environ['DYNAMODB_TEST_RESULT_TABLE']) test_result_table = dynamodb.Table(os.environ['DYNAMODB_TEST_RESULT_TABLE'])
request_spot_instance_for_job(ec2, job_table, job) request_spot_instance_for_job(ec2, job_table, job)
def request_spot_instance_for_job(ec2, job_table, job): def request_spot_instance_for_job(ec2, job_table, job):
"""Request a spot instance to start a job.""" """Request a spot instance to start a job."""
job_id = job['job_id'] job_id = job['job_id']
print('requesting spot instance for job %s' % job_id) print('requesting spot instance for job %s' % job_id)
# Fresh job request.
if job['execution_state'] == 'pending':
# Pick an availability zone randomly.
availability_zones = [ availability_zones = [
az['ZoneName'] az['ZoneName']
for az in ec2.describe_availability_zones()['AvailabilityZones'] for az in ec2.describe_availability_zones()['AvailabilityZones']
if az['State'] == 'available'] if az['State'] == 'available']
# Fresh job request.
if job['execution_state'] == 'pending':
# Pick an availability zone randomly.
availability_zone = random.choice(availability_zones) availability_zone = random.choice(availability_zones)
# Looks like we previously tried to launch a spot instance for this
# job. Examine the state of that request and make sure we can
# replace it.
elif job['execution_state'] == 'spot-instance-requested':
spot_instance_requests = ec2.describe_spot_instance_requests(
SpotInstanceRequestIds=[job['spot_instance_request_id']],
)['SpotInstanceRequests']
# This should never happen assuming this code path doesn't run after
# the spot request was finalized, which should never happen, since
# we trigger things periodically.
if not spot_instance_requests:
print('unable to find spot instance request for job %s' % job_id)
return
sir = spot_instance_requests[0]
# If the spot request expired, try again on to the next availability
# zone.
if (sir['State'] == 'cancelled'
and sir['Status']['Code'] == 'schedule-expired'):
print('previous spot request for job %s expired; '
'trying different az' % job_id)
previous_az = sir['LaunchSpecification']['Placement']['AvailabilityZone']
previous_index = availability_zones.index(previous_az)
try:
availability_zone = availability_zones[previous_index + 1]
except IndexError:
availability_zone = availability_zones[0]
else:
print('unhandled spot instance request state for job %s: '
'%s; %s: %s' % (
job_id,
sir['State'],
sir['Status']['Code'],
sir['Status']['Message']))
print('cancelling job %s' % job_id)
job_table.update_item(
Key={'job_id': job_id},
UpdateExpression=(
'set execution_state = :state, '
'cancelled_reason = :reason'
),
ExpressionAttributeValues={
':state': 'cancelled',
':reason': 'spot state: %s; %s' % (
sir['State'],
sir['Status']['Code'],
),
},
)
return
else: else:
print('unhandled execution_state: %s' % job['execution_state']) print('unhandled execution_state: %s' % job['execution_state'])
return return
start_params = json.loads(job['start_params']) start_params = json.loads(job['start_params'])
user_data_template = start_params['user_data_template'] user_data_template = start_params['user_data_template']
user_data_params = start_params['user_data_params'] user_data_params = start_params['user_data_params']
ec2_instance_config = start_params['ec2_instance_launch_config'] 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)
launch_spec = dict(ec2_instance_config) launch_spec = dict(ec2_instance_config)
launch_spec['Placement'] = { launch_spec['Placement'] = {
'AvailabilityZone': availability_zone, 'AvailabilityZone': availability_zone,
} }
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
# by Amazon. That's fine. We're a CI system. If the instance is terminated, # by Amazon. That's fine. We're a CI system. If the instance is terminated,
# we can just retry the job. # we can just retry the job.
# #
# The max bid price is the on-demand price. So in the typical case we save # The max bid price is the on-demand price. So in the typical case we save
# $$$. If we're unlucky we pay the on-demand rate. You can't lose. # $$$. If we're unlucky we pay the on-demand rate. You can't lose. Unless
# there are no available spot instances. But we handle this by setting a
# short request validity window and retrying in a different availability
# zone. Eventually we should find someone willing to satisfy our request.
res = ec2.request_spot_instances( res = ec2.request_spot_instances(
BlockDurationMinutes=60, BlockDurationMinutes=60,
ValidUntil=datetime.datetime.utcnow() + datetime.timedelta(minutes=10), ValidUntil=datetime.datetime.utcnow() + datetime.timedelta(minutes=1),
LaunchSpecification=launch_spec, LaunchSpecification=launch_spec,
) )
spot_instance_request_id = res['SpotInstanceRequests'][0]['SpotInstanceRequestId'] spot_instance_request_id = res['SpotInstanceRequests'][0]['SpotInstanceRequestId']
print('spot instance request id: %s' % spot_instance_request_id) print('spot instance request id: %s' % spot_instance_request_id)
print('recording spot instance state for job %s' % job_id) print('recording spot instance state for job %s' % job_id)
job_table.update_item( job_table.update_item(
Key={'job_id': job_id}, Key={'job_id': job_id},
UpdateExpression=( UpdateExpression=(
'set execution_state = :state, ' 'set execution_state = :state, '
'spot_instance_request_id = :sir, ' 'spot_instance_request_id = :sir, '
'spot_instance_request_count = spot_instance_request_count + :incr' 'spot_instance_request_count = spot_instance_request_count + :incr'
), ),
ExpressionAttributeValues={ ExpressionAttributeValues={
':state': 'spot-instance-requested', ':state': 'spot-instance-requested',
':sir': spot_instance_request_id, ':sir': spot_instance_request_id,
':incr': 1, ':incr': 1,
} }
) )
def monitor_spot_instance_requests(ec2, job_table, lambda_client,
start_job_function):
"""Looks for stale spot instance requests and updates state accordingly."""
print('monitoring for stale spot instance requests')
# We could call the EC2 API directly. But we want DynamoDB to be our
# source of truth. So key off its state.
res = job_table.query(
IndexName='execution_state',
KeyConditionExpression=Key('execution_state').eq('spot-instance-requested'),
Select='ALL_PROJECTED_ATTRIBUTES',
)
print('found %d jobs in spot-instance-requested state' % len(res['Items']))
for item in res['Items']:
print('job %s is in spot-instance-requested' % item['job_id'])
# If the spot instance request is expired, retrigger scheduling.
request_id = item['spot_instance_request_id']
print('checking state of %s' % request_id)
spot_instance_requests = ec2.describe_spot_instance_requests(
SpotInstanceRequestIds=[request_id],
)['SpotInstanceRequests']
if not spot_instance_requests:
print('could not find %s; weird' % request_id)
continue
sir = spot_instance_requests[0]
print('spot instance request %s is in state %s: %s' % (
request_id, sir['State'], sir['Status']['Code']))
if (sir['State'] == 'cancelled'
and sir['Status']['Code'] == 'schedule-expired'):
print('spot instance request %s for job %s has expired; '
'retrying scheduling' % (request_id, item['job_id']))
payload = json.dumps({
'job_id': item['job_id'],
})
lambda_client.invoke(
FunctionName=start_job_function,
InvocationType='Event',
Payload=payload,
)
def react_to_instance_state_change(job_table, instance, state): def react_to_instance_state_change(job_table, instance, state):
"""React to a CI worker instance state change.""" """React to a CI worker instance state change."""
now = decimal.Decimal(time.time()) now = decimal.Decimal(time.time())
# CI workers advertise their job info via tags. However, the tags cannot # CI workers advertise their job info via tags. However, the tags cannot
# be set for spot instances and are instead encoded in user data. So when # be set for spot instances and are instead encoded in user data. So when
# a spot instance starts, detect that here and set the tags so they can be # a spot instance starts, detect that here and set the tags so they can be
# seen by future handlers. # seen by future handlers.

contrib/ci/lambda_functions/web.py

else: else:
fail_count = 'n/a' fail_count = 'n/a'
if 'skip_count' in job_info: if 'skip_count' in job_info:
skip_count = '%d' % job_info['skip_count'] skip_count = '%d' % job_info['skip_count']
else: else:
skip_count = 'n/a' skip_count = 'n/a'
if job_info['execution_state'] in ('pending', 'spot-instance-requested', 'running'): if job_info['execution_state'] in (
'pending', 'spot-instance-requested', 'cancelled', 'running'):
job_state = job_info['execution_state'] job_state = job_info['execution_state']
elif job_info['execution_state'] == 'done': elif job_info['execution_state'] == 'done':
exit_clean = job_info.get('exit_clean') exit_clean = job_info.get('exit_clean')
if exit_clean is None: if exit_clean is None:
job_state = 'unknown' job_state = 'unknown'
elif exit_clean is True: elif exit_clean is True:
job_state = 'completed' job_state = 'completed'
elif exit_clean is False: elif exit_clean is False:

contrib/ci/terraform/job_executor.tf

# And allow that CloudWatch-initiated trigger of Lambda to work. # And allow that CloudWatch-initiated trigger of Lambda to work.
resource "aws_lambda_permission" "ci_instance_state_change_allow_cloudwatch" { resource "aws_lambda_permission" "ci_instance_state_change_allow_cloudwatch" {
statement_id = "AllowExecutionFromCloudWatch" statement_id = "AllowExecutionFromCloudWatch"
action = "lambda:InvokeFunction" action = "lambda:InvokeFunction"
function_name = aws_lambda_function.ci_instance_state_change.function_name function_name = aws_lambda_function.ci_instance_state_change.function_name
principal = "events.amazonaws.com" principal = "events.amazonaws.com"
source_arn = aws_cloudwatch_event_rule.trigger_instance_state_change.arn source_arn = aws_cloudwatch_event_rule.trigger_instance_state_change.arn
} }
resource "aws_cloudwatch_log_group" "lambda_ci_start_job" {
name = "/aws/lambda/${aws_lambda_function.ci_start_job.function_name}"
retention_in_days = 7
}
resource "aws_iam_role" "lambda_ci_start_job" {
name = "lambda-ci-start-job"
description = "For Lambda function to trigger job start"
assume_role_policy = data.aws_iam_policy_document.assume_role_lambda.json
}
# Lambda function for starting a job. This is similar to `run_pending_job`
# except it isn't a handler for SQS events.
resource "aws_lambda_function" "ci_start_job" {
function_name = "ci-start-job"
description = "Starts a scheduled CI job"
filename = data.archive_file.lambda_ci.output_path
handler = "ci.handle_start_job"
source_code_hash = data.archive_file.lambda_ci.output_base64sha256
runtime = "python3.7"
timeout = 60
role = aws_iam_role.lambda_ci_start_job.arn
environment {
variables = {
DYNAMODB_JOB_TABLE = aws_dynamodb_table.ci_job.name
}
}
}
# Inherit the policy from run_pending_job.
resource "aws_iam_role_policy" "lambda_ci_start_job_run_pending_job" {
role = aws_iam_role.lambda_ci_start_job.name
name = "run-pending-job"
policy = data.aws_iam_policy_document.lambda_ci_run_pending_job.json
}
# Add supplement with additional policy.
data "aws_iam_policy_document" "lambda_ci_start_job" {
# Allow Lambda function to write CloudWatch events.
statement {
effect = "Allow"
actions = [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
]
resources = [
aws_cloudwatch_log_group.lambda_ci_start_job.arn]
}
}
resource "aws_iam_role_policy" "lambda_ci_start_job" {
role = aws_iam_role.lambda_ci_start_job.name
name = aws_iam_role.lambda_ci_start_job.name
policy = data.aws_iam_policy_document.lambda_ci_start_job.json
}

contrib/ci/terraform/spot_instance_request_monitor.tf

  • This file was added.
# Defines resources to monitor spot instance requests.
resource "aws_cloudwatch_log_group" "lambda_ci_spot_instance_request_monitor" {
name = "/aws/lambda/${aws_lambda_function.ci_spot_instance_request_monitor.function_name}"
retention_in_days = 7
}
resource "aws_iam_role" "lambda_ci_spot_instance_request_monitor" {
name = "lambda-ci-spot-instance-request-monitor"
description = "For Lambda function monitoring spot instance requests"
assume_role_policy = data.aws_iam_policy_document.assume_role_lambda.json
}
# Function that monitors spot instance requests and retries failed ones.
resource "aws_lambda_function" "ci_spot_instance_request_monitor" {
function_name = "ci-spot-instance-request-monitor"
description = "Monitors spot instance requests and triggers activity"
filename = data.archive_file.lambda_ci.output_path
handler = "ci.handle_spot_instance_request_monitor"
source_code_hash = data.archive_file.lambda_ci.output_base64sha256
runtime = "python3.7"
timeout = 60
role = aws_iam_role.lambda_ci_spot_instance_request_monitor.arn
environment {
variables = {
DYNAMODB_JOB_TABLE = aws_dynamodb_table.ci_job.name
LAMBDA_START_JOB_FUNCTION = aws_lambda_function.ci_start_job.function_name
}
}
}
data "aws_iam_policy_document" "lambda_ci_spot_instance_request_monitor" {
# Allow Lambda function to write CloudWatch events.
statement {
effect = "Allow"
actions = [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
]
resources = [aws_cloudwatch_log_group.lambda_ci_spot_instance_request_monitor.arn]
}
# Allow querying spot instance requests.
statement {
effect = "Allow"
actions = [
"ec2:DescribeSpotInstanceRequests",
]
resources = ["*"]
}
# Allow fetching job state from DynamoDB.
statement {
effect = "Allow"
actions = [
"dynamodb:Query",
]
resources = [
aws_dynamodb_table.ci_job.arn,
"${aws_dynamodb_table.ci_job.arn}/*",
]
}
# Allow invoking the start job Lambda function.
statement {
effect = "Allow"
actions = ["lambda:InvokeFunction"]
resources = [aws_lambda_function.ci_start_job.arn]
}
}
resource "aws_iam_role_policy" "lambda_ci_spot_instance_request_monitor" {
role = aws_iam_role.lambda_ci_spot_instance_request_monitor.name
name = aws_iam_role.lambda_ci_spot_instance_request_monitor.name
policy = data.aws_iam_policy_document.lambda_ci_spot_instance_request_monitor.json
}
# Periodically trigger the Lambda function so state is continuously monitored.
resource "aws_cloudwatch_event_rule" "trigger_ci_spot_instance_request_monitor" {
name = "trigger-ci-spot-instance-request-monitor"
description = "Trigger monitoring of spot instance requests"
schedule_expression = "rate(1 minute)"
}
resource "aws_cloudwatch_event_target" "ci_spot_instance_request_monitor" {
rule = aws_cloudwatch_event_rule.trigger_ci_spot_instance_request_monitor.name
arn = aws_lambda_function.ci_spot_instance_request_monitor.arn
}
resource "aws_lambda_permission" "ci_spot_instance_request_monitor_allow_cloudwatch" {
statement_id = "AllowExecutionFromCloudWatch"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.ci_spot_instance_request_monitor.function_name
principal = "events.amazonaws.com"
source_arn = aws_cloudwatch_event_rule.trigger_ci_spot_instance_request_monitor.arn
}

contrib/ci/terraform/storage.tf

resource "aws_dynamodb_table" "ci_job" { resource "aws_dynamodb_table" "ci_job" {
name = "ci-job" name = "ci-job"
billing_mode = "PAY_PER_REQUEST" billing_mode = "PAY_PER_REQUEST"
attribute { attribute {
name = "job_id" name = "job_id"
type = "S" type = "S"
} }
attribute {
name = "execution_state"
type = "S"
}
hash_key = "job_id" hash_key = "job_id"
# This allows us to easily query for jobs in a specific state.
global_secondary_index {
name = "execution_state"
hash_key = "execution_state"
range_key = "job_id"
projection_type = "ALL"
}
} }
# Tracks results for individual tests in each job. # Tracks results for individual tests in each job.
resource "aws_dynamodb_table" "ci_test_result" { resource "aws_dynamodb_table" "ci_test_result" {
name = "ci-test-result" name = "ci-test-result"
billing_mode = "PAY_PER_REQUEST" billing_mode = "PAY_PER_REQUEST"
attribute { attribute {