Behind the Scenes at Jungle Disk - Autoscaling SQS Workers

TL;DR

Checkout the code repo for an example terraform config that uses Ansible in the user-script to setup a worker that terminates itself.

Autoscale up

AWS autoscaling is an excellent tool for managing a dynamic pool of workers. For stateless work, i.e. serving a website, allowing autoscaling to expand and contract the worker pool works very well. However, for stateful work such as processing jobs off a queue, allowing autoscaling to contract the worker pool is undesirable since it may terminate a worker currently processing a job. Instead the workers themselves should manage removing themselves from the autoscaling group.

This gist shows a snippit that watches a queue and adds a worker when the queue contains at least one message for > 5 min.

resource "aws_autoscaling_policy" "seppuku_servers" { name = "${var.environment}" autoscaling_group_name = "${aws_autoscaling_group.seppuku_servers.name}" scaling_adjustment = 1 cooldown = 300 adjustment_type = "ChangeInCapacity" } resource "aws_cloudwatch_metric_alarm" "add_workers" { alarm_name = "${var.environment}-add-workers" metric_name = "ApproximateNumberOfMessagesVisible" namespace = "AWS/SQS" statistic = "Sum" period = "300" threshold = "1" comparison_operator = "GreaterThanOrEqualToThreshold" evaluation_periods = "1" dimensions { QueueName = "${aws_sqs_queue.jobs.name}" } alarm_actions = ["${aws_autoscaling_policy.seppuku_servers.arn}"] }

Ansible on deploy

By embedding an Ansible playbook in the terraform config, instances that are created by the autoscaling policy can auto deploy themselves:

module "ansible" { source = "./ansible" bucket = "${var.ansible_bucket}" prefix = "${var.ansible_prefix}" } data "aws_ami" "fedora" { most_recent = true filter { name = "name" values = [ "Fedora-Cloud-Base-25-*-gp2-0" ] } filter { name = "virtualization-type" values = [ "hvm" ] } owners = [ "125523088429" ] # Fedora Cloud SIG } data "template_file" "user_data" { template = "${file("${path.module}/user_data.sh")}" vars { ssh_bucket = "${var.ssh_keys_bucket}" ssh_prefix = "${var.ssh_keys_prefix}" ssh_user = "fedora" ssh_keys_cron = "${var.ssh_keys_update_cron}" ansible_bucket = "${var.ansible_bucket}" ansible_prefix = "${var.ansible_prefix}" ansible_vault_file = "${var.ansible_vault_file}" } } resource "aws_launch_configuration" "worker" { name_prefix = "${var.environment}-" image_id = "${data.aws_ami.fedora.id}" instance_type = "${var.worker_instance_type}" associate_public_ip_address = "true" root_block_device { volume_type = "gp2" } security_groups = [ "${aws_security_group.worker.id}", ] iam_instance_profile = "${aws_iam_instance_profile.worker.id}" user_data = "${data.template_file.user_data.rendered}" lifecycle { create_before_destroy = true } }

The instance_profile uses the following IAM config:

# data "aws_iam_policy_document" "assume-role" { statement { actions = [ "sts:AssumeRole" ] principals { type = "Service" identifiers = ["ec2.amazonaws.com", "lambda.amazonaws.com"] } } } data "aws_iam_policy_document" "allow-worker-sqs-work" { statement { effect = "Allow" actions = [ "sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:DeleteMessageBatch", ] resources = [ "${aws_sqs_queue.jobs.arn}" ] } } data "aws_iam_policy_document" "allow-s3-ssh-keys" { statement { effect = "Allow" actions = [ "s3:GetObject", "s3:GetObjectVersion" ] resources = [ "arn:aws:s3:::${var.ssh_keys_bucket}/${var.ssh_keys_prefix}/*", ], } statement { effect = "Allow" actions = [ "s3:ListBucket", ] resources = [ "arn:aws:s3:::${var.ssh_keys_bucket}" ], condition { test = "StringLike" variable = "s3:prefix" values = [ "${var.ssh_keys_prefix}/*" ] } } } data "aws_iam_policy_document" "allow-s3-ansible" { statement { effect = "Allow" actions = [ "s3:GetObject", "s3:GetObjectVersion" ] resources = [ "arn:aws:s3:::${var.ansible_bucket}/${var.ansible_prefix}/*", ], } statement { effect = "Allow" actions = [ "s3:ListBucket", ] resources = [ "arn:aws:s3:::${var.ansible_bucket}" ], condition { test = "StringLike" variable = "s3:prefix" values = [ "${var.ansible_prefix}/*" ] } } } data "aws_iam_policy_document" "allow-s3-ansible-vault" { statement { effect = "Allow" actions = [ "s3:GetObject", "s3:GetObjectVersion" ] resources = [ "arn:aws:s3:::${var.ansible_bucket}/${var.ansible_vault_file}", ], } } data "aws_iam_policy_document" "allow-ec2-describe-tags" { statement { effect = "Allow" actions = [ "ec2:DescribeTags", ] resources = [ "*" ] } } data "aws_iam_policy_document" "allow-autoscaling-terminate" { statement { effect = "Allow" actions = [ "autoscaling:TerminateInstanceInAutoScalingGroup" ] resources = [ "*" ] } } resource "aws_iam_role" "worker" { name = "worker" path = "/${var.environment}/" assume_role_policy = "${data.aws_iam_policy_document.assume-role.json}" } resource "aws_iam_instance_profile" "worker" { name = "worker" path = "/${var.environment}/" roles = ["${aws_iam_role.worker.name}"] } resource "aws_iam_role_policy" "allow-worker-sqs-work" { name = "allow-worker-sqs-work" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-worker-sqs-work.json}" } resource "aws_iam_role_policy" "allow-s3-ssh-keys" { name = "allow-s3-ssh-keys" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-s3-ssh-keys.json}" } resource "aws_iam_role_policy" "allow-s3-ansible" { name = "allow-s3-ansible" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-s3-ansible.json}" } resource "aws_iam_role_policy" "allow-s3-ansible-vault" { name = "allow-s3-ansible-vault" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-s3-ansible-vault.json}" } resource "aws_iam_role_policy" "allow-ec2-describe-tags" { name = "allow-ec2-describe-tags" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-ec2-describe-tags.json}" } resource "aws_iam_role_policy" "allow-autoscaling-terminate" { name = "allow-autoscaling-terminate" role = "${aws_iam_role.worker.id}" policy = "${data.aws_iam_policy_document.allow-autoscaling-terminate.json}" }

NOTE See the code_repo for the ssh_keys auto deployment

#!/usr/bin/env bash dnf install -y ansible awscli jq python-boto3 python-boto \ libselinux-python cat <<"EOF" > /home/${ssh_user}/update_ssh_authorized_keys.sh #!/usr/bin/env bash set -e BUCKET_URI=s3://${ssh_bucket}/${ssh_prefix} MARKER="# KEYS_BELOW_WILL_BE_UPDATED_BY_TERRAFORM" KEYS_FILE=/home/${ssh_user}/.ssh/authorized_keys TEMP_KEYS_FILE=$(mktemp) PUB_KEYS_DIR=/home/${ssh_user}/pub_key_files/ mkdir -p $PUB_KEYS_DIR # Add marker, if not present, and copy static content. grep -Fxq "$MARKER" $KEYS_FILE || echo -e "\n$MARKER" >> $KEYS_FILE line=$(grep -n "$MARKER" $KEYS_FILE | cut -d ":" -f 1) head -n $line $KEYS_FILE > $TEMP_KEYS_FILE # Synchronize the keys from the bucket. aws s3 sync --delete $BUCKET_URI $PUB_KEYS_DIR for filename in $PUB_KEYS_DIR/*; do sed 's/\n\?$/\n/' < $filename >> $TEMP_KEYS_FILE done # Move the new authorized keys in place. mv $TEMP_KEYS_FILE $KEYS_FILE chown ${ssh_user}:${ssh_user} $KEYS_FILE chmod 600 $KEYS_FILE if selinuxenabled; then restorecon -R -v $KEYS_FILE fi EOF cat <<"EOF" > /home/${ssh_user}/.ssh/config Host * StrictHostKeyChecking no EOF chmod 600 /home/${ssh_user}/.ssh/config chown ${ssh_user}:${ssh_user} /home/${ssh_user}/.ssh/config chown ${ssh_user}:${ssh_user} /home/${ssh_user}/update_ssh_authorized_keys.sh chmod 755 /home/${ssh_user}/update_ssh_authorized_keys.sh # Execute now su ${ssh_user} -c /home/${ssh_user}/update_ssh_authorized_keys.sh # Add to cron if [ -n "${ssh_keys_cron}" ]; then croncmd="/home/${ssh_user}/update_ssh_authorized_keys.sh" cronjob="${ssh_keys_cron} $croncmd" ( crontab -u ${ssh_user} -l | grep -v "$croncmd" echo "$cronjob" ) | crontab -u ${ssh_user} - fi aws s3 cp s3://${ansible_bucket}/${ansible_vault_file} /root/.vault chmod 700 /root/.vault chown root:root /root/.vault ANSIBLE_DIR=/root/${ansible_prefix} mkdir -p $ANSIBLE_DIR aws s3 sync --delete s3://${ansible_bucket}/${ansible_prefix} $ANSIBLE_DIR #ansible-galaxy install -r $ANSIBLE_DIR/requirements.yml # NOTE(jkoelker) run it twice due to RH1398272 importing gpg key locking rpmdb ansible-playbook --vault-password-file=/root/.vault \ -i "localhost," -c local $ANSIBLE_DIR/master.yml || \ ansible-playbook --vault-password-file=/root/.vault \ -i "localhost," -c local $ANSIBLE_DIR/master.yml

# variable "bucket" {} variable "prefix" {} variable "files" { default = [ "master.yml", "templates/worker.service.j2", "templates/queue-manager.sh.j2", "templates/wrapper.sh.j2", "templates/seppuku.service.j2", "templates/seppuku.sh.j2", ] } resource "aws_s3_bucket_object" "ansible" { bucket = "${var.bucket}" key = "${var.prefix}/${element(var.files, count.index)}" source = "${path.module}/${element(var.files, count.index)}" etag = "${md5(file("${path.module}/${element(var.files, count.index)}"))}" count = "${length(var.files)}" }

The user_data.sh script installs Ansible and copies down the playbook from the s3 bucket. It then call Ansible locally to configure the instance.

# - hosts: localhost vars: sqs_visibility_timout: 18000 # NOTE(jkoelker) Since we have to pay for the full hour anyway, # wait up to 30 mins for a job sqs_wait_time: 20 sqs_max_waits: 75 pre_tasks: - name: Gather facts ec2_facts: - name: Retrieve all tags on an instance ec2_tag: region: '{{ ansible_ec2_placement_region }}' resource: '{{ ansible_ec2_instance_id }}' state: list register: ec2_tags - name: Discover SQS work queue set_fact: sqs_work: '{{ ec2_tags.tags["SQS-work"] }}' tasks: - name: Copy the seppuku script into place template: src: seppuku.sh.j2 dest: /usr/local/bin/seppuku.sh owner: root group: root mode: 0755 - name: Configure the seppuku service template: src: seppuku.service.j2 dest: /etc/systemd/system/seppuku.service - name: Enable the seppuku service systemd: name: seppuku.service enabled: yes daemon_reload: yes - name: Copy the queue manager into place template: src: queue-manager.sh.j2 dest: /usr/local/bin/queue-manager.sh owner: root group: root mode: 0755 - name: Copy the wrapper into place template: src: wrapper.sh.j2 dest: /usr/local/bin/wrapper.sh owner: root group: root mode: 0755 - name: Configure the worker service template: src: worker.service.j2 dest: /etc/systemd/system/worker.service - name: Start the worker systemd: name: worker.service state: started daemon_reload: yes

The Ansible play gathers some facts from the local environment and configures a set of systemd services and scripts.

The wrapper.sh script simply calls a worker program with a job_id. This script is a good place to put any environment variables needed for the worker program to function.

#!/bin/bash echo "Starting on {{ sqs_work }}" MESSAGE=$(aws sqs receive-message \ --region "{{ ansible_ec2_placement_region }}" \ --queue-url "{{ sqs_work }}" \ --visibility-timeout "{{ sqs_visibility_timout }}" \ --wait-time-seconds "{{ sqs_wait_time }}" \ --max-number-of-messages 1) if [[ "x${MESSAGE}" == "x" ]]; then echo "Did not recieve a job in the last {{ sqs_wait_time }} seconds." exit 10 fi HANDLE=$(echo "${MESSAGE}" | jq -r '.Messages[0].ReceiptHandle') BODY=$(echo "${MESSAGE}" | jq -r '.Messages[0].Body') MSG=$(echo "${BODY}" | jq -r '.Message') JOBID=$(echo "${MSG}" | jq -r '.job_id') if [[ "x${JOBID}" == "x" ]]; then echo "Error finding jerb_id" exit 20 fi echo "Spawning worker for job_id ${JOBID}" # NOTE(jkoelker) Always exit with a "Failure" since systemd is restarting # and we want it to shutdown the specified number of # "unsuccessful" starts EXIT=1 if ! /usr/local/bin/wrapper.sh "${JOBID}"; then echo "Error processing job ${JOBID}" EXIT=30 fi aws sqs delete-message \ --region "{{ ansible_ec2_placement_region }}" \ --queue-url "{{ sqs_work }}" \ --receipt-handle "${HANDLE}" exit ${EXIT} # vim: set ft=sh:

The queue-manager.sh script long polls the SQS job queue and pulls out the job_id and passes it to the wrapper.sh script. After the worker process has exited it removes the job from the queue.

#!/bin/bash METADATA_URL="http://169.254.169.254/latest" INSTANCE_ID=$(curl ${METADATA_URL}/meta-data/instance-id) IDENT_DOC=$(curl -s ${METADATA_URL}/dynamic/instance-identity/document) REGION=$(echo "${IDENT_DOC}" | jq -r '.region') aws autoscaling terminate-instance-in-auto-scaling-group \ --region "${REGION}" \ --instance-id "${INSTANCE_ID}" \ --should-decrement-desired-capacity # vim: set ft=sh:

The seppuku.sh script discovers its instance_id from the metadata service and then terminates itself via the autoscaling group decrementing the capacity desired. This logic serves to “scale-in” the group when jobs are scarce.

How the seppuku.sh script gets triggered abuses a bit of systemd magic.

[Unit] Description=Decrement the autoscaling group and seppuku Wants=network-online.target After=network-online.target [Service] Type=oneshot RemainAfterExit=true ExecStart=/usr/local/bin/seppuku.sh [Install] WantedBy=multi-user.target

[Unit] Description=Work the job queue StartLimitAction=reboot StartLimitIntervalSec={{ sqs_wait_time * sqs_max_waits * 2 }} StartLimitBurst={{ sqs_max_waits }} [Service] ExecStart=/usr/local/bin/queue-manager.sh Restart=always RestartSec=0

The seppuku.service is enabled, but not started by Ansible. This means that the next time the server boots up, it will start the seppuku.service.

The worker.service relies on the fact that the queue-manager.sh script always exits after one attempT to process the queue. It it setup to always restart the service without delay, but if it restarts the service more than the StartLimitBurst value in StartLimitIntervalSec seconds, it will reboot the server, causing the seppuku.service to run destroying the instance and “scaling-in” the worker pool.

Since AWS charges for a full hour for a fractional hour, it is in the best interest of cost for the workers to work as much as possible. Therefore, the Ansible variables configure the service to restart up to 75 time in 50 mins. This leaves about 10 mins for the server to reboot and remove itself at the end of the first hour.