Repository: openaddresses/machine
Pull request #463 (2016-11-27 20:19:24)

Make task instances police themselves
Task instances launched by `openaddr-run-ec2-command` will now talk to Slack and upload their own logs to S3. If this works well, the next step will be to remove the long-term supervision responsibility from the script running under cron, so that the [central scheduled task instance](https://github.com/openaddresses/machine/blob/master/docs/components.md#scheduled-tasks) has fewer long-running jobs to worry about, and can be restarted or replaced more frequently.


Patch:
diff --git a/chef/crontab/recipes/default.rb b/chef/crontab/recipes/default.rb
index 5b3e2d0..d2a0910 100644
--- a/chef/crontab/recipes/default.rb
+++ b/chef/crontab/recipes/default.rb
@@ -61,15 +61,15 @@ end
 file "/etc/cron.d/openaddr_crontab-collect-extracts" do
     content <<-CRONTAB
 PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin
-SLACK_URL=#{slack_url}
 LC_ALL=C.UTF-8
 # Archive collection, every other day at 11am UTC (4am PDT)
-0 11	*/2 * *	#{username}	( \
-  curl -X POST -d '{"text": "Starting new collection zips..."}' $SLACK_URL -s ; \
+0 11	*/2 * *	#{username}	\
   openaddr-run-ec2-command \
   -a "#{aws_access_id}" \
   -s "#{aws_secret_key}" \
+  -b "#{aws_s3_bucket}" \
   --sns-arn "#{aws_sns_arn}" \
+  --slack-url "#{slack_url}" \
   --verbose \
   -- \
     openaddr-collect-extracts \
@@ -79,8 +79,6 @@ LC_ALL=C.UTF-8
     -b "#{aws_s3_bucket}" \
     --sns-arn "#{aws_sns_arn}" \
     --verbose \
-  && curl -X POST -d '{"text": "Completed <https://#{cname}|new collection zips>."}' $SLACK_URL -s \
-  || curl -X POST -d '{"text": "Failed to complete new collection zips."}' $SLACK_URL -s ) \
   >> /var/log/openaddr_crontab/collect-extracts.log 2>&1
 CRONTAB
 end
@@ -88,16 +86,16 @@ end
 file "/etc/cron.d/openaddr_crontab-index-tiles" do
     content <<-CRONTAB
 PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin
-SLACK_URL=#{slack_url}
 LC_ALL=C.UTF-8
 # Index into tiles, every third day at 11am UTC (4am PDT)
-0 11	*/3 * *	#{username}	( \
-  curl -X POST -d '{"text": "Starting new spatial index..."}' $SLACK_URL -s ; \
+0 11	*/3 * *	#{username}	\
   openaddr-run-ec2-command \
   --hours 9 \
   -a "#{aws_access_id}" \
   -s "#{aws_secret_key}" \
+  -b "#{aws_s3_bucket}" \
   --sns-arn "#{aws_sns_arn}" \
+  --slack-url "#{slack_url}" \
   --verbose \
   -- \
     openaddr-index-tiles \
@@ -107,8 +105,6 @@ LC_ALL=C.UTF-8
     -b "#{aws_s3_bucket}" \
     --sns-arn "#{aws_sns_arn}" \
     --verbose \
-  && curl -X POST -d '{"text": "Completed <https://#{cname}|new spatial index>."}' $SLACK_URL -s \
-  || curl -X POST -d '{"text": "Failed to complete new spatial index."}' $SLACK_URL -s ) \
   >> /var/log/openaddr_crontab/index-tiles.log 2>&1
 CRONTAB
 end
@@ -116,18 +112,18 @@ end
 file "/etc/cron.d/openaddr_crontab-dotmap" do
     content <<-CRONTAB
 PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin
-SLACK_URL=#{slack_url}
 LC_ALL=C.UTF-8
 # Generate OpenAddresses dot map, every fifth day at 11am UTC (4am PDT)
-0 11	*/5 * *	#{username}	( \
-  curl -X POST -d '{"text": "Starting new dot map..."}' $SLACK_URL -s ; \
+0 11	*/5 * *	#{username}	\
   openaddr-run-ec2-command \
   --role dotmap \
   --hours 16 \
   --instance-type r3.large \
   -a "#{aws_access_id}" \
   -s "#{aws_secret_key}" \
+  -b "#{aws_s3_bucket}" \
   --sns-arn "#{aws_sns_arn}" \
+  --slack-url "#{slack_url}" \
   --verbose \
   -- \
     openaddr-update-dotmap \
@@ -136,8 +132,6 @@ LC_ALL=C.UTF-8
     -a "#{aws_access_id}" \
     -s "#{aws_secret_key}" \
     --sns-arn "#{aws_sns_arn}" \
-  && curl -X POST -d '{"text": "Completed <https://openaddresses.io|new dot map>."}' $SLACK_URL -s \
-  || curl -X POST -d '{"text": "Failed to complete new dot map."}' $SLACK_URL -s ) \
   >> /var/log/openaddr_crontab/dotmap.log 2>&1
 CRONTAB
 end
diff --git a/openaddr/run_ec2_ami.py b/openaddr/run_ec2_ami.py
index bbfe01d..af45a63 100644
--- a/openaddr/run_ec2_ami.py
+++ b/openaddr/run_ec2_ami.py
@@ -18,18 +18,24 @@ parser.add_argument('-a', '--access-key', default=environ.get('AWS_ACCESS_KEY_ID
 parser.add_argument('-s', '--secret-key', default=environ.get('AWS_SECRET_ACCESS_KEY', None),
                     help='Optional AWS secret key name. Defaults to value of AWS_SECRET_ACCESS_KEY environment variable.')
 
+parser.add_argument('-b', '--bucket', default=environ.get('AWS_S3_BUCKET', None),
+                    help='S3 bucket name. Defaults to value of AWS_S3_BUCKET environment variable.')
+
 parser.add_argument('--sns-arn', default=environ.get('AWS_SNS_ARN', None),
                     help='Optional AWS Simple Notification Service (SNS) resource. Defaults to value of AWS_SNS_ARN environment variable.')
 
 parser.add_argument('--role', default='openaddr',
                     help='Machine chef role to execute. Defaults to "openaddr".')
 
-parser.add_argument('--hours', default=12, type=int,
+parser.add_argument('--hours', default=12, type=float,
                     help='Number of hours to allow before giving up. Defaults to 12 hours.')
 
 parser.add_argument('--instance-type', default='m3.medium',
                     help='EC2 instance type. Defaults to "m3.medium".')
 
+parser.add_argument('--slack-url',
+                    help='Slack POST URL.')
+
 parser.add_argument('-v', '--verbose', help='Turn on verbose logging',
                     action='store_const', dest='loglevel',
                     const=logging.DEBUG, default=logging.INFO)
@@ -45,14 +51,15 @@ def main():
     ''' 
     '''
     args = parser.parse_args()
-    instance, deadline = False, time() + args.hours * 3600
+    instance, deadline, lifespan = False, time() + (args.hours + 1) * 3600, int(args.hours * 3600)
     setup_logger(args.access_key, args.secret_key, args.sns_arn, log_level=args.loglevel)
 
     try:
         ec2 = connect_ec2(args.access_key, args.secret_key)
         autoscale = connect_autoscale(args.access_key, args.secret_key)
         instance = request_task_instance(ec2, autoscale, args.instance_type,
-                                         args.role, args.command)
+                                         args.role, lifespan, args.command,
+                                         args.bucket, args.slack_url)
 
         while True:
             instance.update()
diff --git a/openaddr/util/__init__.py b/openaddr/util/__init__.py
index f5147c4..14ddd19 100644
--- a/openaddr/util/__init__.py
+++ b/openaddr/util/__init__.py
@@ -8,6 +8,7 @@ from tempfile import mkstemp
 from os import close
 import ftplib, httmock
 import io, zipfile
+import json
 
 from ..compat import quote
 
@@ -50,7 +51,18 @@ def set_autoscale_capacity(autoscale, cloudwatch, capacity):
     if measure['Maximum'] > .9:
         group.set_capacity(capacity)
 
-def request_task_instance(ec2, autoscale, instance_type, chef_role, command):
+def _command_messages(command):
+    '''
+    '''
+    strings = dict(
+        message_starting = 'Starting {}...'.format(command),
+        message_complete = 'Completed {}'.format(command),
+        message_failed = 'Failed {}'.format(command),
+        )
+    
+    return { k: quote(json.dumps(dict(text=v))) for (k, v) in strings.items() }
+
+def request_task_instance(ec2, autoscale, instance_type, chef_role, lifespan, command, bucket, slack_url):
     '''
     '''
     group_name = 'CI Workers {0}.x'.format(*get_version().split('.'))
@@ -59,10 +71,22 @@ def request_task_instance(ec2, autoscale, instance_type, chef_role, command):
     (config, ) = autoscale.get_all_launch_configurations(names=[group.launch_config_name])
     (image, ) = ec2.get_all_images(image_ids=[config.image_id])
     keypair = ec2.get_all_key_pairs()[0]
+
+    yyyymmdd = datetime.now().strftime('%Y-%m-%d')
     
     with open(join(dirname(__file__), 'templates', 'task-instance-userdata.sh')) as file:
-        userdata_kwargs = dict(role=chef_role, command=' '.join(map(quote, command)))
-        userdata_kwargs.update(version=quote(get_version()))
+        userdata_kwargs = dict(
+            role = quote(chef_role),
+            command = ' '.join(map(quote, command)),
+            lifespan = quote(str(lifespan)),
+            version = quote(get_version()),
+            access_key = quote(ec2.aws_access_key_id),
+            secret_key = quote(ec2.aws_secret_access_key),
+            log_prefix = quote('logs/{}-{}'.format(yyyymmdd, command[0])),
+            bucket = quote(bucket or 'data.openaddresses.io'),
+            slack_url = quote(slack_url or ''),
+            **_command_messages(command[0])
+            )
     
         run_kwargs = dict(instance_type=instance_type, security_groups=['default'],
                           instance_initiated_shutdown_behavior='terminate',
@@ -71,7 +95,7 @@ def request_task_instance(ec2, autoscale, instance_type, chef_role, command):
 
     reservation = image.run(**run_kwargs)
     (instance, ) = reservation.instances
-    instance.add_tag('Name', 'Scheduled {} {}'.format(datetime.now().strftime('%Y-%m-%d'), command[0]))
+    instance.add_tag('Name', 'Scheduled {} {}'.format(yyyymmdd, command[0]))
     
     _L.info('Started EC2 instance {} from AMI {}'.format(instance, image))
     
diff --git a/openaddr/util/templates/task-instance-userdata.sh b/openaddr/util/templates/task-instance-userdata.sh
index 28669b7..7af0a98 100644
--- a/openaddr/util/templates/task-instance-userdata.sh
+++ b/openaddr/util/templates/task-instance-userdata.sh
@@ -1,12 +1,45 @@
-#!/bin/sh -ex
-apt-get update -y
+#!/bin/bash -ex
+
+# Tell Slack all about it
+function notify_slack
+{{
+    if [ {slack_url} ]; then
+        curl -s -o /dev/null -X POST -d $1 {slack_url} || true
+    fi
+}}
+
+notify_slack {message_starting}
+
+# Bail out with a log message
+function shutdown_with_log
+{{
+    if [ $1 = 0 ]; then
+        notify_slack {message_complete}
+    else
+        notify_slack {message_failed}
+    fi
+    
+    mkdir /tmp/task
+    gzip -c /var/log/cloud-init-output.log > /tmp/task/cloud-init-output.log.gz
+    echo {command} > /tmp/task/command
+    echo $1 > /tmp/task/status
+    
+    AWS_ACCESS_KEY_ID={access_key} AWS_SECRET_ACCESS_KEY={secret_key} \
+        aws s3 cp /tmp/task s3://{bucket}/{log_prefix}/ --recursive --acl private
+    
+    shutdown -h now
+}}
+
+# Bail out when the timer reaches zero
+( sleep {lifespan}; shutdown_with_log 2 ) &
 
 # (Re)install machine.
 cd /home/ubuntu/machine
 sudo -u ubuntu git fetch origin {version}
 sudo -u ubuntu git rebase FETCH_HEAD
-chef/run.sh {role}
 
-LC_ALL="C.UTF-8" {command} 2>&1
+apt-get update -y
+chef/run.sh {role}
 
-shutdown -h now
+# Run the actual command
+LC_ALL="C.UTF-8" {command} && shutdown_with_log 0 || shutdown_with_log 1 2>&1
diff --git a/setup.py b/setup.py
index 250ea0a..9add235 100644
--- a/setup.py
+++ b/setup.py
@@ -130,6 +130,9 @@ setup(
 
         # http://pythonhosted.org/itsdangerous/
         'itsdangerous == 0.24',
+        
+        # https://aws.amazon.com/cli/
+        'awscli == 1.11.21',
 
         ] + conditional_requirements
 )


Test patch:
diff --git a/openaddr/tests/util.py b/openaddr/tests/util.py
index 185da78..1467a1f 100644
--- a/openaddr/tests/util.py
+++ b/openaddr/tests/util.py
@@ -88,13 +88,14 @@ class TestUtilities (unittest.TestCase):
         
         autoscale.get_all_groups.return_value = [group]
         autoscale.get_all_launch_configurations.return_value = [config]
+        ec2.aws_access_key_id, ec2.aws_secret_access_key = 'this', 'that'
         ec2.get_all_images.return_value = [image]
         ec2.get_all_key_pairs.return_value = [keypair]
         
         image.run.return_value = reservation
         reservation.instances = [instance]
         
-        util.request_task_instance(ec2, autoscale, 'm3.medium', chef_role, command)
+        util.request_task_instance(ec2, autoscale, 'm3.medium', chef_role, 60, command, 'bucket-name', None)
         
         autoscale.get_all_groups.assert_called_once_with([expected_group_name])
         autoscale.get_all_launch_configurations.assert_called_once_with(names=[group.launch_config_name])
@@ -107,6 +108,9 @@ class TestUtilities (unittest.TestCase):
         self.assertEqual(image_run_kwargs['key_name'], keypair.name)
         
         self.assertIn('chef/run.sh {}'.format(quote(chef_role)), image_run_kwargs['user_data'])
+        self.assertIn('s3://bucket-name/logs/', image_run_kwargs['user_data'])
+        self.assertIn('AWS_ACCESS_KEY_ID={}'.format(quote(ec2.aws_access_key_id)), image_run_kwargs['user_data'])
+        self.assertIn('AWS_SECRET_ACCESS_KEY={}'.format(quote(ec2.aws_secret_access_key)), image_run_kwargs['user_data'])
         for (arg1, arg2) in zip(command, command[1:]):
             self.assertIn(quote(arg1)+' '+quote(arg2), image_run_kwargs['user_data'])