my_orchestrate.py

from collections import defaultdict
import json
import logging

import salt.minion
import salt.utils
import salt.utils.event


log = logging.getLogger(__name__)


def _get_failed_states(data):
    failed = []

    orch_state_run = sorted(data.itervalues(), key=lambda d: d['__run_num__'])
    for orch_state in orch_state_run:
        if orch_state['result']:
            continue

        # if the orchestration state doesn't have an __id__ field, we know
        # it's because it failed as a result of failing dependencies, so we
        # can just stash it and move on
        if '__id__' not in orch_state:
            # a missing __id__ also means it *does* have __sls__, which we can
            # use for nice output
            log.info('orchestration state %s failed because of failed '
                     'dependencies', orch_state['__sls__'])
            failed.append(orch_state)
            continue

        failed_states = defaultdict(list)

        # The comment for a failed orchestration state will contain the entire
        # json-encoded return from the minion. However, it's given to us
        # concatenated to some human-readable output so we figure out where
        # the json starts, then load it and extract failed states.
        json_idx = orch_state['comment'].index('{')
        json_str = orch_state['comment'][json_idx:]
        result = json.loads(json_str)

        for minion, states in result.iteritems():
            states_run = sorted(states.itervalues(),
                                key=lambda d: d['__run_num__'])

            for state in states_run:
                # if the result has an __sls__ field, that means it failed
                # because of failed dependencies, so we just skip over it
                if state['result'] or '__sls__' in state:
                    continue

                failed_states[minion].append(state)

        # torch the comment, otherwise we'll have the entire nasty
        # mess a part of our error payloads
        del orch_state['comment']
        orch_state['failed_children'] = dict(failed_states)
        failed.append(orch_state)

    return failed



def my_orchestrate(mods, run_id, saltenv='base', pillar=None, on_success=None,
                   on_failure=None):
    """Do customized orchestration.
    :param mods: orchestration states to apply
    :param run_id: an ID, ideally randomized, used to correlate events
        emitted by a specific execution of this runner.
    :param saltenv: Salt environment in which to find SLSes to apply
    :param pillar: Pillar data to pass through to states
    :param on_success: event tag to fire on success
    :param on_failure: event tag to fire on failure
    """
    event = salt.utils.event.get_master_event(__opts__, __opts__['sock_dir'],
                                              listen=False)
    log.info('starting my_orchestrate, run ID %s', run_id)

    opts = __opts__.copy()
    # We force the output to json and set the static option so we can more
    # easily parse failed state information out of the `comment` field
    # on orchestration state results.
    #
    # It's not clear if this is intended behavior but we have to do this
    # because the data returned by state.sls does not given state-by-
    # state, per-minion failure output. States that succeed appear 
    # individually in the returned data but failures only appear in the
    # single collected JSON output for the entire minion state run.
    opts.update({
        'file_client': 'local',
        'output': 'json',
        'static': True,
    })

    minion = salt.minion.MasterMinion(opts)
    result = minion.functions['state.sls'](mods, saltenv=saltenv, pillar=pillar)

    failed = None
    if not salt.utils.check_state_result(result):
        failed = _get_failed_states(result)

    ev_data = {
        'status': 'failed' if failed else 'succeeded',
        'pillar': pillar,
        'run_id': run_id,
    }

    if not failed and on_success:
        log.info('firing my_orchestrate success event')
        event.fire_event(ev_data, on_success)

    elif failed and on_failure:
        log.info('firing my_orchestrate failure event')
        event.fire_event(ev_data, on_failure)
        log.error('Deploy failed:\n\n%s', pformat(failed))

    return {
        'data': {minion.opts['id']: result},
        'retcode': 1 if failed else 0,
    }

my_reactor.sls

{% set tag_parts = tag.split('/') %}
{% set app = tag_parts[-1] %}
{% set env = data['post']['env'] %}
{% set version = data['post']['version'] %}
{% set pkg = '%s==%s' | format(app, version) %}

{% set success_event = 'my_org/deploy/%s/status/succeeded' | format(app) %}
{% set failure_event = 'my_org/deploy/%s/status/failed' | format(app) %}

{% set run_id = salt.cmd.run('openssl rand -hex 4') %}


handle_deploy:
  runner.my_orchestrate.my_orchestrate:
    - mods:
      - orch.deploy_{{ app }}
    - pillar:
        deploy_env: {{ env }}
        deploy_package: {{ pkg }}
    - run_id: {{ run_id }}
    - on_success: {{ success_event }}
    - on_failure: {{ failure_event }}

send_start_notification:
  runner.slack.message:
    - url: https://hooks.slack.com/services/a/b/c
    - attachments:
      - title: deploy STARTED
        mrkdwn_in:
          - text
        color: good
        text: |
          run ID: `{{ run_id }}`
          package: `{{ pkg }}`
          environment: `{{ env }}`

Worth noting, it's probably not the best idea to fire off events this way. I wrote in the event firing because I had somehow missed that you can fire events using the fire_event state requisite. Whoops.