Lso client

The LSO client, which interacts with LSO running externally.

LSO is responsible for executing Ansible playbooks, that deploy subscriptions.

_LSOState

Bases: TypedDict

An expanded state that must contain at least the required keys for the execution of an Ansible playbook.

Source code in gso/services/lso_client.py

class _LSOState(TypedDict):  # noqa: PYI049
    """An expanded state that must contain at least the required keys for the execution of an Ansible playbook."""

    playbook_name: str
    extra_vars: dict[str, Any]
    inventory: dict[Literal["all"], dict[Literal["hosts"], dict[str, Any] | None]]
    __extra_values__: Any  # This is feature unavailable in python 3.12

_send_request(parameters, callback_route)

Send a request to LSO. The callback address is derived using the process ID provided.

Parameters:

Name	Type	Description	Default
parameters	dict	JSON body for the request, which will almost always at least consist of a subscription object, and a boolean value to indicate a dry run.	required
callback_route	str	The callback route that should be used to resume the workflow.	required

Raises:

Type	Description
HTTPError	When receiving a non-successful status code from LSO.

Source code in gso/services/lso_client.py

def _send_request(parameters: dict, callback_route: str) -> None:
    """Send a request to LSO. The callback address is derived using the process ID provided.

    Args:
        parameters: JSON body for the request, which will almost always at least consist of a subscription object, and a
            boolean value to indicate a dry run.
        callback_route: The callback route that should be used to resume the workflow.

    Raises:
        HTTPError: When receiving a non-successful status code from LSO.
    """
    oss = settings.load_oss_params()
    params = oss.PROVISIONING_PROXY

    # Build up a callback URL of the Provisioning Proxy to return its results to.
    callback_url = f"{oss.GENERAL.internal_hostname}{callback_route}"
    debug_msg = f"[provisioning proxy] Callback URL set to {callback_url}"
    logger.debug(debug_msg)

    parameters.update({"callback": callback_url})
    url = f"{params.scheme}://{params.api_base}/api/playbook"

    request = requests.post(url, json=parameters, timeout=10)
    request.raise_for_status()

_execute_playbook(playbook_name, callback_route, inventory, extra_vars)

Execute a playbook remotely through the provisioning proxy.

When providing this method with an inventory, the format should be compatible with the Ansible YAML-based format. For example, an inventory consisting of two hosts, which each a unique host variable assigned to them looks as follows:

"inventory": {
    "all": {
        "hosts": {
            "host1.local": {
                "foo": "bar"
            },
            "host2.local": {
                "key": "value"
            },
            "host3.local": None
        }
    }
}

Danger

Note the fact that the collection of all hosts is a dictionary, and not a list of strings. Ansible expects each host to be a key-value pair. The key is the FQDN of a host, and the value always null.

The extra vars can be a simple dict consisting of key-value pairs, for example:

"extra_vars": {
    "dry_run": true,
    "commit_comment": "I am a robot!",
    "verb": "deploy"
}

Parameters:

Name	Type	Description	Default
playbook_name	str	Filename of the playbook that is to be executed. It must be present on the remote system running the provisioning proxy, otherwise it will return an error.	required
callback_route	str	The endpoint at which GSO expects a callback to continue the workflow executing this step.	required
inventory	dict[str, Any]	An inventory of machines at which the playbook is targeted. Must be in YAML-compatible format.	required
extra_vars	dict[str, Any]	Any extra variables that the playbook relies on. This can include a subscription object, a boolean value indicating a dry run, a commit comment, etc. All unicode character values are decoded to prevent sending special characters to remote machines that don't support this.	required

Source code in gso/services/lso_client.py

@step("Execute Ansible playbook")
def _execute_playbook(
    playbook_name: str, callback_route: str, inventory: dict[str, Any], extra_vars: dict[str, Any]
) -> None:
    """Execute a playbook remotely through the provisioning proxy.

    When providing this method with an inventory, the format should be compatible with the Ansible YAML-based format.
    For example, an inventory consisting of two hosts, which each a unique host variable assigned to them looks as
    follows:

    ```py
    "inventory": {
        "all": {
            "hosts": {
                "host1.local": {
                    "foo": "bar"
                },
                "host2.local": {
                    "key": "value"
                },
                "host3.local": None
            }
        }
    }
    ```

    !!! danger
        Note the fact that the collection of all hosts is a dictionary, and not a list of strings. Ansible expects each
        host to be a key-value pair. The key is the FQDN of a host, and the value always `null`.

    The extra vars can be a simple dict consisting of key-value pairs, for example:

    <!-- vale off -->
    ```py
    "extra_vars": {
        "dry_run": true,
        "commit_comment": "I am a robot!",
        "verb": "deploy"
    }
    ```
    <!-- vale on -->

    Args:
        playbook_name: Filename of the playbook that is to be executed. It must be present on the remote system running
            the provisioning proxy, otherwise it will return an error.
        callback_route: The endpoint at which GSO expects a callback to continue the workflow executing this step.
        inventory: An inventory of machines at which the playbook is targeted. Must be in YAML-compatible format.
        extra_vars: Any extra variables that the playbook relies on. This can include a subscription object, a boolean
            value indicating a dry run, a commit comment, etc. All unicode character values are decoded to prevent
            sending special characters to remote machines that don't support this.
    """
    parameters = {
        "playbook_name": playbook_name,
        "inventory": inventory,
        "extra_vars": json.loads(unidecode(json.dumps(extra_vars, ensure_ascii=False))),
    }

    _send_request(parameters, callback_route)

validate_inventory_is_set(state)

Validate whether the passed Ansible inventory is empty.

If the inventory is empty, which can happen in select cases, there should be no playbook run. This conditional will prevent from calling out to LSO with an empty playbook, which would cause the Ansible runner process to hang. This in turn will result in a workflow step that is never called back to.

Source code in gso/services/lso_client.py

def validate_inventory_is_set(state: State) -> bool:
    """Validate whether the passed Ansible inventory is empty.

    If the inventory is empty, which can happen in select cases, there should be no playbook run. This conditional will
    prevent from calling out to LSO with an empty playbook, which would cause the Ansible runner process to
    hang. This in turn will result in a workflow step that is never called back to.
    """
    if "inventory" not in state:
        msg = "Missing Ansible inventory for playbook."
        raise ProcessFailureError(msg, details="Key 'inventory' not found in state.")
    if "all" not in state["inventory"] or "hosts" not in state["inventory"]["all"]:
        msg = "Malformed Ansible inventory found in state."
        raise ProcessFailureError(msg, details="Ansible inventory must be in YAML form, not string.")

    return state["inventory"]["all"]["hosts"]

lso_interaction(provisioning_step)

Interact with the provisioning proxy LSO using a callback step.

An asynchronous interaction with the provisioning proxy. This is an external system that executes Ansible playbooks to provision service subscriptions. If the playbook fails, this step will also fail, allowing for the user to retry provisioning from the UI.

Optionally, the keys lso_result_title and lso_result_extra_label can be added to the state before running this interaction. They will be used to customise the input step that shows the outcome of the LSO interaction.

Parameters:

Name	Type	Description	Default
provisioning_step	Step	A workflow step that performs an operation remotely using the provisioning proxy.	required

Returns:

Type	Description
StepList	A list of steps that is executed as part of the workflow.

Source code in gso/services/lso_client.py

def lso_interaction(provisioning_step: Step) -> StepList:
    """Interact with the provisioning proxy LSO using a callback step.

    An asynchronous interaction with the provisioning proxy. This is an external system that executes Ansible playbooks
    to provision service subscriptions. If the playbook fails, this step will also fail, allowing for the user to retry
    provisioning from the UI.

    Optionally, the keys `lso_result_title` and `lso_result_extra_label` can be added to the state before running
    this interaction. They will be used to customise the input step that shows the outcome of the LSO
    interaction.

    Args:
        provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.

    Returns:
        A list of steps that is executed as part of the workflow.
    """
    return (
        begin
        >> provisioning_step
        >> _inventory_is_not_empty(
            begin
            >> callback_step(
                name=RUNNING_ANSIBLE_PLAYBOOK_STEP_NAME, action_step=_execute_playbook, validate_step=_evaluate_results
            )
            >> step("Inject result title")(lambda: {"lso_result_title": provisioning_step.name})
            >> _show_results
        )
        >> _clean_state
    )

indifferent_lso_interaction(provisioning_step)

Interact with the provisioning proxy LSO using a callback step.

This interaction is identical from the one described in lso_interaction(), with one functional difference. Whereas the lso_interaction() will make the workflow step fail on unsuccessful interaction, this step will not. It is therefore indifferent about the outcome of the Ansible playbook that is executed.

Danger

Using this interaction requires the operator to manually evaluate the outcome of a playbook. If a playbook fails, this will not cause the workflow to fail.

Parameters:

Name	Type	Description	Default
provisioning_step	Step	A workflow step that performs an operation remotely using the provisioning proxy.	required

Returns:

Type	Description
StepList	A list of steps that is executed as part of the workflow.

Source code in gso/services/lso_client.py

def indifferent_lso_interaction(provisioning_step: Step) -> StepList:
    """Interact with the provisioning proxy LSO using a callback step.

    This interaction is identical from the one described in `lso_interaction()`, with one functional difference.
    Whereas the `lso_interaction()` will make the workflow step fail on unsuccessful interaction, this step will not.
    It is therefore indifferent about the outcome of the Ansible playbook that is executed.

    !!! danger
        Using this interaction requires the operator to manually evaluate the outcome of a playbook. If a playbook
        fails, this will not cause the workflow to fail.

    Args:
        provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.

    Returns:
        A list of steps that is executed as part of the workflow.
    """
    return (
        begin
        >> provisioning_step
        >> _inventory_is_not_empty(
            begin
            >> callback_step(
                name=RUNNING_ANSIBLE_PLAYBOOK_STEP_NAME, action_step=_execute_playbook, validate_step=_ignore_results
            )
            >> step("Inject result title")(lambda: {"lso_result_title": provisioning_step.name})
            >> _show_results
        )
        >> _clean_state
    )

anonymous_lso_interaction(provisioning_step, validation_step=_evaluate_results)

Interact with the provisioning proxy LSO without any user input.

Similar to the indifferent LSO interaction, there also is the anonymous interaction. Output is not ignored but no input step is created to display the results. A custom validation step may be given as input. This validation step should look inside the callback_result key in the current state. By default, only the return code of the playbook execution is evaluated.

Parameters:

Name	Type	Description	Default
provisioning_step	Step	A workflow step to remotely provision a subscription.	required
validation_step	Step	An optional validation step which defaults to a step that evaluates the return code.	_evaluate_results

Source code in gso/services/lso_client.py

def anonymous_lso_interaction(provisioning_step: Step, validation_step: Step = _evaluate_results) -> StepList:
    """Interact with the provisioning proxy LSO without any user input.

    Similar to the indifferent LSO interaction, there also is the anonymous interaction. Output is not ignored
    but no input step is created to display the results.
    A custom validation step may be given as input. This validation step should look inside the `callback_result` key
    in the current state. By default, only the return code of the playbook execution is evaluated.

    Args:
        provisioning_step: A workflow step to remotely provision a subscription.
        validation_step: An optional validation step which defaults to a step that evaluates the return code.
    """
    return (
        begin
        >> provisioning_step
        >> _inventory_is_not_empty(
            callback_step(
                name=RUNNING_ANSIBLE_PLAYBOOK_STEP_NAME, action_step=_execute_playbook, validate_step=validation_step
            )
        )
        >> _clean_state
    )

anonymous_indifferent_lso_interaction(provisioning_step)

Interact with the provisioning proxy LSO without any user input.

Similar to the anonymous LSO interaction, but indifferent about the outcome of the Ansible playbook that is executed

Parameters:

Name	Type	Description	Default
provisioning_step	Step	A workflow step to remotely provision a subscription.	required

Source code in gso/services/lso_client.py

def anonymous_indifferent_lso_interaction(provisioning_step: Step) -> StepList:
    """Interact with the provisioning proxy LSO without any user input.

    Similar to the anonymous LSO interaction, but indifferent about the outcome of the Ansible playbook that is executed

    Args:
         provisioning_step: A workflow step to remotely provision a subscription.
    """
    return (
        begin
        >> provisioning_step
        >> _inventory_is_not_empty(
            callback_step(
                name=RUNNING_ANSIBLE_PLAYBOOK_STEP_NAME,
                action_step=_execute_playbook,
                validate_step=_ignore_results,
            )
        )
        >> _clean_state
    )