def state(name,
        tgt,
        ssh=False,
        tgt_type='glob',
        ret='',
        ret_config=None,
        ret_kwargs=None,
        highstate=None,
        sls=None,
        top=None,
        saltenv=None,
        test=None,
        pillar=None,
        pillarenv=None,
        expect_minions=True,
        fail_minions=None,
        allow_fail=0,
        exclude=None,
        concurrent=False,
        timeout=None,
        batch=None,
        queue=False,
        subset=None,
        orchestration_jid=None,
        **kwargs):
    '''
    Invoke a state run on a given target

    name
        An arbitrary name used to track the state execution

    tgt
        The target specification for the state run.

        .. versionadded: 2016.11.0

        Masterless support: When running on a masterless minion, the ``tgt``
        is ignored and will always be the local minion.

    tgt_type
        The target type to resolve, defaults to ``glob``

    ret
        Optionally set a single or a list of returners to use

    ret_config
        Use an alternative returner configuration

    ret_kwargs
        Override individual returner configuration items

    highstate
        Defaults to None, if set to True the target systems will ignore any
        sls references specified in the sls option and call state.highstate
        on the targeted minions

    top
        Should be the name of a top file. If set state.top is called with this
        top file instead of state.sls.

    sls
        A group of sls files to execute. This can be defined as a single string
        containing a single sls file, or a list of sls files

    test
        Pass ``test=true`` or ``test=false`` through to the state function. This
        can be used to overide a test mode set in the minion's config file. If
        left as the default of None and the 'test' mode is supplied on the
        command line, that value is passed instead.

    pillar
        Pass the ``pillar`` kwarg through to the state function

    pillarenv
        The pillar environment to grab pillars from

        .. versionadded:: 2017.7.0

    saltenv
        The default salt environment to pull sls files from

    ssh
        Set to `True` to use the ssh client instead of the standard salt client

    roster
        In the event of using salt-ssh, a roster system can be set

    expect_minions
        An optional boolean for failing if some minions do not respond

    fail_minions
        An optional list of targeted minions where failure is an option

    allow_fail
        Pass in the number of minions to allow for failure before setting
        the result of the execution to False

    exclude
        Pass exclude kwarg to state

    concurrent
        Allow multiple state runs to occur at once.

        WARNING: This flag is potentially dangerous. It is designed
        for use when multiple state runs can safely be run at the same
        Do not use this flag for performance optimization.

    queue
        Pass ``queue=true`` through to the state function

    batch
        Execute the command :ref:`in batches <targeting-batch>`. E.g.: ``10%``.

        .. versionadded:: 2016.3.0

    subset
        Number of minions from the targeted set to randomly use

        .. versionadded:: 2017.7.0

    asynchronous
        Run the salt command but don't wait for a reply.

        NOTE: This flag conflicts with subset and batch flags and cannot be
        used at the same time.

        .. versionadded:: neon


    Examples:

    Run a list of sls files via :py:func:`state.sls <salt.state.sls>` on target
    minions:

    .. code-block:: yaml

        webservers:
          salt.state:
            - tgt: 'web*'
            - sls:
              - apache
              - django
              - core
            - saltenv: prod

    Run sls file via :py:func:`state.sls <salt.state.sls>` on target
    minions with exclude:

    .. code-block:: yaml

        docker:
          salt.state:
            - tgt: 'docker*'
            - sls: docker
            - exclude: docker.swarm
            - saltenv: prod

    Run a full :py:func:`state.highstate <salt.state.highstate>` on target
    mininons.

    .. code-block:: yaml

        databases:
          salt.state:
            - tgt: role:database
            - tgt_type: grain
            - highstate: True
    '''
    cmd_kw = {'arg': [], 'kwarg': {}, 'ret': ret, 'timeout': timeout}

    if ret_config:
        cmd_kw['ret_config'] = ret_config

    if ret_kwargs:
        cmd_kw['ret_kwargs'] = ret_kwargs

    state_ret = {'name': name,
                 'changes': {},
                 'comment': '',
                 'result': True}

    try:
        allow_fail = int(allow_fail)
    except ValueError:
        state_ret['result'] = False
        state_ret['comment'] = 'Passed invalid value for \'allow_fail\', must be an int'
        return state_ret

    cmd_kw['tgt_type'] = tgt_type
    cmd_kw['ssh'] = ssh
    if 'roster' in kwargs:
        cmd_kw['roster'] = kwargs['roster']
    cmd_kw['expect_minions'] = expect_minions
    cmd_kw['asynchronous'] = kwargs.pop('asynchronous', False)
    if highstate:
        fun = 'state.highstate'
    elif top:
        fun = 'state.top'
        cmd_kw['arg'].append(top)
    elif sls:
        fun = 'state.sls'
        if isinstance(sls, list):
            sls = ','.join(sls)
        cmd_kw['arg'].append(sls)
    else:
        state_ret['comment'] = 'No highstate or sls specified, no execution made'
        state_ret['result'] = False
        return state_ret

    if test is not None or __opts__.get('test'):
        cmd_kw['kwarg']['test'] = test if test is not None else __opts__.get('test')

    if pillar:
        cmd_kw['kwarg']['pillar'] = pillar

    if pillarenv is not None:
        cmd_kw['kwarg']['pillarenv'] = pillarenv

    if saltenv is not None:
        cmd_kw['kwarg']['saltenv'] = saltenv

    if exclude is not None:
        cmd_kw['kwarg']['exclude'] = exclude

    cmd_kw['kwarg']['queue'] = queue

    if isinstance(concurrent, bool):
        cmd_kw['kwarg']['concurrent'] = concurrent
    else:
        state_ret['comment'] = ('Must pass in boolean for value of \'concurrent\'')
        state_ret['result'] = False
        return state_ret

    if batch is not None:
        cmd_kw['batch'] = six.text_type(batch)
    if subset is not None:
        cmd_kw['subset'] = subset

    masterless = __opts__['__role'] == 'minion' and \
                 __opts__['file_client'] == 'local'
    if not masterless:
        _fire_args({'type': 'state', 'tgt': tgt, 'name': name, 'args': cmd_kw})
        cmd_ret = __salt__['saltutil.cmd'](tgt, fun, **cmd_kw)
    else:
        if top:
            cmd_kw['topfn'] = ''.join(cmd_kw.pop('arg'))
        elif sls:
            cmd_kw['mods'] = ''.join(cmd_kw.pop('arg'))
        cmd_kw.update(cmd_kw.pop('kwarg'))
        tmp_ret = __salt__[fun](**cmd_kw)
        cmd_ret = {__opts__['id']: {
            'ret': tmp_ret,
            'out': tmp_ret.get('out', 'highstate') if
                isinstance(tmp_ret, dict) else 'highstate'
        }}

    if cmd_kw['asynchronous']:
        state_ret['__jid__'] = cmd_ret.get('jid')
        state_ret['changes'] = cmd_ret
        if int(cmd_ret.get('jid', 0)) > 0:
            state_ret['result'] = True
            state_ret['comment'] = 'State submitted successfully.'
        else:
            state_ret['result'] = False
            state_ret['comment'] = 'State failed to run.'
        return state_ret

    try:
        state_ret['__jid__'] = cmd_ret[next(iter(cmd_ret))]['jid']
    except (StopIteration, KeyError):
        pass

    changes = {}
    fail = set()
    no_change = set()

    if fail_minions is None:
        fail_minions = ()
    elif isinstance(fail_minions, six.string_types):
        fail_minions = [minion.strip() for minion in fail_minions.split(',')]
    elif not isinstance(fail_minions, list):
        state_ret.setdefault('warnings', []).append(
            '\'fail_minions\' needs to be a list or a comma separated '
            'string. Ignored.'
        )
        fail_minions = ()

    if not cmd_ret and expect_minions:
        state_ret['result'] = False
        state_ret['comment'] = 'No minions returned'
        return state_ret

    for minion, mdata in six.iteritems(cmd_ret):
        if mdata.get('out', '') != 'highstate':
            log.warning('Output from salt state not highstate')

        m_ret = False

        if 'return' in mdata and 'ret' not in mdata:
            mdata['ret'] = mdata.pop('return')

        m_state = True
        if mdata.get('failed', False):
            m_state = False
        else:
            try:
                m_ret = mdata['ret']
            except KeyError:
                m_state = False
            if m_state:
                m_state = __utils__['state.check_result'](m_ret, recurse=True)

        if not m_state:
            if minion not in fail_minions:
                fail.add(minion)
            changes[minion] = m_ret
            continue
        try:
            for state_item in six.itervalues(m_ret):
                if isinstance(state_item, dict):
                    if 'changes' in state_item and state_item['changes']:
                        changes[minion] = m_ret
                        break
            else:
                no_change.add(minion)
        except AttributeError:
            log.error("m_ret did not have changes %s %s", type(m_ret), m_ret)
            no_change.add(minion)

    if changes:
        state_ret['changes'] = {'out': 'highstate', 'ret': changes}
    if len(fail) > allow_fail:
        state_ret['result'] = False
        state_ret['comment'] = 'Run failed on minions: {0}'.format(', '.join(fail))
    else:
        state_ret['comment'] = 'States ran successfully.'
        if changes:
            state_ret['comment'] += ' Updating {0}.'.format(', '.join(changes))
        if no_change:
            state_ret['comment'] += ' No changes made to {0}.'.format(', '.join(no_change))
    if test or __opts__.get('test'):
        if state_ret['changes'] and state_ret['result'] is True:
            # Test mode with changes is the only case where result should ever be none
            state_ret['result'] = None
    return state_ret
Invoke a state run on a given target

    name
        An arbitrary name used to track the state execution

    tgt
        The target specification for the state run.

        .. versionadded: 2016.11.0

        Masterless support: When running on a masterless minion, the ``tgt``
        is ignored and will always be the local minion.

    tgt_type
        The target type to resolve, defaults to ``glob``

    ret
        Optionally set a single or a list of returners to use

    ret_config
        Use an alternative returner configuration

    ret_kwargs
        Override individual returner configuration items

    highstate
        Defaults to None, if set to True the target systems will ignore any
        sls references specified in the sls option and call state.highstate
        on the targeted minions

    top
        Should be the name of a top file. If set state.top is called with this
        top file instead of state.sls.

    sls
        A group of sls files to execute. This can be defined as a single string
        containing a single sls file, or a list of sls files

    test
        Pass ``test=true`` or ``test=false`` through to the state function. This
        can be used to overide a test mode set in the minion's config file. If
        left as the default of None and the 'test' mode is supplied on the
        command line, that value is passed instead.

    pillar
        Pass the ``pillar`` kwarg through to the state function

    pillarenv
        The pillar environment to grab pillars from

        .. versionadded:: 2017.7.0

    saltenv
        The default salt environment to pull sls files from

    ssh
        Set to `True` to use the ssh client instead of the standard salt client

    roster
        In the event of using salt-ssh, a roster system can be set

    expect_minions
        An optional boolean for failing if some minions do not respond

    fail_minions
        An optional list of targeted minions where failure is an option

    allow_fail
        Pass in the number of minions to allow for failure before setting
        the result of the execution to False

    exclude
        Pass exclude kwarg to state

    concurrent
        Allow multiple state runs to occur at once.

        WARNING: This flag is potentially dangerous. It is designed
        for use when multiple state runs can safely be run at the same
        Do not use this flag for performance optimization.

    queue
        Pass ``queue=true`` through to the state function

    batch
        Execute the command :ref:`in batches <targeting-batch>`. E.g.: ``10%``.

        .. versionadded:: 2016.3.0

    subset
        Number of minions from the targeted set to randomly use

        .. versionadded:: 2017.7.0

    asynchronous
        Run the salt command but don't wait for a reply.

        NOTE: This flag conflicts with subset and batch flags and cannot be
        used at the same time.

        .. versionadded:: neon


    Examples:

    Run a list of sls files via :py:func:`state.sls <salt.state.sls>` on target
    minions:

    .. code-block:: yaml

        webservers:
          salt.state:
            - tgt: 'web*'
            - sls:
              - apache
              - django
              - core
            - saltenv: prod

    Run sls file via :py:func:`state.sls <salt.state.sls>` on target
    minions with exclude:

    .. code-block:: yaml

        docker:
          salt.state:
            - tgt: 'docker*'
            - sls: docker
            - exclude: docker.swarm
            - saltenv: prod

    Run a full :py:func:`state.highstate <salt.state.highstate>` on target
    mininons.

    .. code-block:: yaml

        databases:
          salt.state:
            - tgt: role:database
            - tgt_type: grain
            - highstate: True