Repository: ansible/awx
File: awx/main/tasks/receptor.py (Python)

Code before:
            if work_data.get('StateName') in RECEPTOR_ACTIVE_STATES:
                continue  # do not want to touch active work units
            logger.info(f'Reaping orphaned work unit {unit_id} with params {params}')
        receptor_ctl.simple_command(f"work release {unit_id}")


class RemoteJobError(RuntimeError):
    pass


def run_until_complete(node, timing_data=None, **kwargs):
    """
    Runs an ansible-runner work_type on remote node, waits until it completes, then returns stdout.
    """
    config_data = read_receptor_config()
    receptor_ctl = get_receptor_ctl(config_data)

    use_stream_tls = getattr(get_conn_type(node, receptor_ctl), 'name', None) == "STREAMTLS"
    kwargs.setdefault('tlsclient', get_tls_client(config_data, use_stream_tls))
    # kwargs.setdefault('ttl', '20s')
    kwargs.setdefault('payload', '')
    if work_signing_enabled(config_data):
        kwargs['signwork'] = True

    transmit_start = time.time()
    result = receptor_ctl.submit_work(worktype='local', node=node, **kwargs)

    unit_id = result['unitid']
    run_start = time.time()
    if timing_data:
        timing_data['transmit_timing'] = run_start - transmit_start
    run_timing = 0.0
    stdout = ''
    state_name = 'local var never set'

    try:
        resultfile = receptor_ctl.get_work_results(unit_id)

        while run_timing < 20.0:
            status = receptor_ctl.simple_command(f'work status {unit_id}')
            state_name = status.get('StateName')
            if state_name not in RECEPTOR_ACTIVE_STATES:
                break
            run_timing = time.time() - run_start
            time.sleep(0.5)
        else:
            raise RemoteJobError(f'Receptor job timeout on {node} after {run_timing} seconds, state remains in {state_name}')

        if timing_data:
            timing_data['run_timing'] = run_timing


Review comment (bug):
ok, I think I see what's going on.

https://github.com/ansible/awx/blob/devel/tools/docker-compose/ansible/roles/sources/templates/receptor-awx.conf.j2#L42

The `ansible-runner` type is used for execution nodes. The task normally bypasses receptor for running itself locally. But since the bug was on execution nodes, only, you changed this so the test will run the same logic path (even though not on an execution node). You obviously have to reset this change back.

What I would look into - keep `test_tagless_image` the same as it is now. That will now be the "control" node version of the test. Then make your changes here configurable via a kwarg to this method. You will need some additional logic, and a new kwarg on the task to tell it to run locally _as if_ it were an execution node. Introduce a `test_tagless_image_execution_node` test that does the same stuff but calls it with that stuff.

Code after:
            if work_data.get('StateName') in RECEPTOR_ACTIVE_STATES:
                continue  # do not want to touch active work units
            logger.info(f'Reaping orphaned work unit {unit_id} with params {params}')
        receptor_ctl.simple_command(f"work release {unit_id}")


class RemoteJobError(RuntimeError):
    pass


def run_until_complete(node, timing_data=None, worktype='ansible-runner', ttl='20s', **kwargs):
    """
    Runs an ansible-runner work_type on remote node, waits until it completes, then returns stdout.
    """

    config_data = read_receptor_config()
    receptor_ctl = get_receptor_ctl(config_data)

    use_stream_tls = getattr(get_conn_type(node, receptor_ctl), 'name', None) == "STREAMTLS"
    kwargs.setdefault('tlsclient', get_tls_client(config_data, use_stream_tls))
    if ttl is not None:
        kwargs['ttl'] = ttl
    kwargs.setdefault('payload', '')
    if work_signing_enabled(config_data):
        kwargs['signwork'] = True

    transmit_start = time.time()
    result = receptor_ctl.submit_work(worktype=worktype, node=node, **kwargs)

    unit_id = result['unitid']
    run_start = time.time()
    if timing_data:
        timing_data['transmit_timing'] = run_start - transmit_start
    run_timing = 0.0
    stdout = ''
    state_name = 'local var never set'

    try:
        resultfile = receptor_ctl.get_work_results(unit_id)

        while run_timing < 20.0:
            status = receptor_ctl.simple_command(f'work status {unit_id}')
            state_name = status.get('StateName')
            if state_name not in RECEPTOR_ACTIVE_STATES:
                break
            run_timing = time.time() - run_start
            time.sleep(0.5)
        else:
            raise RemoteJobError(f'Receptor job timeout on {node} after {run_timing} seconds, state remains in {state_name}')

        if timing_data: