python handler.py

@manager.task(name='schedule')
def schedule(playbook, variables, tags, is_local=None):
    slack.api_token = settings.SLACK_TOKEN
    vault_pass = utils.read_vault_file(settings.VAULT_PASSWORD_FILE)
    playbook_file = settings.PLAYBOOK_BASE_PATH + '/' + playbook + '.yml'

    stats = callbacks.AggregateStats()
    playbook_cb = callbacks.PlaybookCallbacks(verbose=utils.VERBOSITY)
    runner_cb = callbacks.PlaybookRunnerCallbacks(stats, verbose=utils.VERBOSITY)

    inventory = ansible.inventory.Inventory(settings.ANSIBLE_INVENTORY, vault_password=vault_pass)

    variables["datetime"] = time.strftime("%Y-%m-%d / %H:%M:%S")

    if "ec2volumes-cleanup" in tags:
        variables["skip_from"] = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        variables["skip_to"] = datetime.now().strftime('%Y-%m-%d')

    if is_local:
        inventory.subset('localhost')

    playbook_run = ansible.playbook.PlayBook(
        playbook=playbook_file,
        module_path=C.DEFAULT_MODULE_PATH,
        inventory=inventory,
        forks=C.DEFAULT_FORKS,
        remote_user=C.DEFAULT_REMOTE_USER,
        remote_pass=None,
        callbacks=playbook_cb,
        runner_callbacks=runner_cb,
        stats=stats,
        timeout=C.DEFAULT_TIMEOUT,
        transport=C.DEFAULT_TRANSPORT,
        become=C.DEFAULT_BECOME,
        become_method=C.DEFAULT_BECOME_METHOD,
        become_user=None,
        become_pass=None,
        extra_vars=variables,
        private_key_file=C.DEFAULT_PRIVATE_KEY_FILE,
        only_tags=tags,
        skip_tags=None,
        check=False,
        diff=False,
        vault_password=vault_pass,
        force_handlers=False
    )

    failed_hosts = []
    unreachable_hosts = []

    try:
        playbook_run.run()

        hosts = sorted(playbook_run.stats.processed.keys())
        display(callbacks.banner("PLAY RECAP"))
        playbook_cb.on_stats(playbook_run.stats)

        for h in hosts:
            t = playbook_run.stats.summarize(h)
            if t['failures'] > 0:
                failed_hosts.append(h)
            if t['unreachable'] > 0:
                unreachable_hosts.append(h)

        for h in hosts:
            t = playbook_run.stats.summarize(h)

            display("%s : %s %s %s %s" % (
                hostcolor(h, t),
                colorize('ok', t['ok'], 'green'),
                colorize('changed', t['changed'], 'yellow'),
                colorize('unreachable', t['unreachable'], 'red'),
                colorize('failed', t['failures'], 'red')),
                screen_only=True
            )

            display("%s : %s %s %s %s" % (
                hostcolor(h, t, False),
                colorize('ok', t['ok'], None),
                colorize('changed', t['changed'], None),
                colorize('unreachable', t['unreachable'], None),
                colorize('failed', t['failures'], None)),
                log_only=True
            )

        if len(unreachable_hosts) > 0 or len(failed_hosts) > 0:
            message_color = "yellow"
            message = "Unstable " + playbook + " (" + str(variables) + " - " + str(tags) + ") scheduled task - " + \
                      str(len(unreachable_hosts)) + " unreachable and " + str(len(failed_hosts)) + \
                      " failed hosts - inspect logs"
            return_code = 2
        else:
            message_color = "green"
            message = "Stable " + playbook + " (" + str(variables) + " - " + str(tags) + \
                      ") scheduled task completed - all hosts succeeded."
            return_code = 0

        slack.chat.post_message(settings.SLACK_ROOM, message, username=settings.SLACK_SENDER)
        return return_code

    except errors.AnsibleError, e:
        display("ERROR: %s" % e, color='red')
        message_color = "red"
        message = "Failed " + playbook + " (" + str(variables) + " - " + str(tags) + \
                  + ") schedule task - there was an error executing the ansible playbook."
        return_code = 1
        slack.chat.post_message(settings.SLACK_ROOM, message, username=settings.SLACK_SENDER)
        return return_code

scheduler definiton

manager = Celery('deploy', broker=settings.CELERY_BROKER)
manager.conf.update(
    CELERY_DEFAULT_QUEUE='default',
    CELERY_QUEUES=tuple(celery_queues_list),
    CELERY_RESULT_BACKEND=settings.CELERY_BROKER,
    CELERYBEAT_SCHEDULE={
        'backups': {
            'task': 'schedule',
            'schedule': crontab(minute='0', hour='*/3'),
            'args': ('backup', {}, ['ec2volumes'], True)
        },
        'cleanup': {
            'task': 'schedule',
            'schedule': crontab(minute=0, hour=0),
            'args': ('backup', {}, ['ec2volumes-cleanup'], True)
        }
    }
)

tasks.yml

---
- name: Snapshot EC2 volumes
  ec2_snapshot:
    aws_access_key: "{{ aws.key }}"
    aws_secret_key: "{{ aws.secret }}"
    description: "{{ datetime }} - {{ item.description }} - {{ item.id }}"
    region: "us-west-1"
    volume_id: "{{ item.id }}"
    wait: False
  delegate_to: localhost
  with_items:
    - "{{ ec2.volumes }}"
  connection: local
  tags:
    - ec2volumes

- name: List existing volumes
  shell: "aws ec2 describe-snapshots --owner {{ aws.owner_id }}"
  args:
    executable: "{{ common.default_shell }}"
  connection: local
  delegate_to: localhost
  register: all_snapshots
  tags:
    - ec2volumes-cleanup

- name: Store the list of snapshots as a fact de-JSONiefied
  set_fact:
    snapshot_list: "{{ (all_snapshots.stdout|from_json).Snapshots }}"
  connection: local
  delegate_to: localhost
  tags:
    - ec2volumes-cleanup

- name: Get rid of EBS snapshots older than 2 days
  shell: "aws ec2 delete-snapshot --snapshot-id {{ item.SnapshotId }}"
  args:
    executable: "{{ common.default_shell }}"
  with_items:
    - "{{ snapshot_list }}"
  when: item.StartTime|truncate(10, True, '') != skip_from and item.StartTime|truncate(10, True, '') != skip_to
  connection: local
  delegate_to: localhost
  tags:
    - ec2volumes-cleanup