Automation Blog

from Stefan Schnell

Ansible is an IT process automation engine for configuration management, application deployment and orchestration. The use of this software and technology replaces manual repetitive tasks and executes them automatically. The spectrum of applications of Ansible has a different focus than VCF Automation. From this perspective, combining both products appears interesting and promises advantages. This post shows an approach that allows to invoke templates at the Ansible Automation Platform (AAP) from VCF Automation, jobs to be monitored and the results to be returned.

Ansible Job Monitor

The following Python action contains several functions: The main function first detects the template ID. Here it is necessary to enter the name of the template as well as the user name and password for the AAP. Optionally it is possible to enter data for inventory and credential. If the credentials are configured in the AAP and used in this call, an ssh_password is not required. The template is then launched and the ID generated from the job is monitored in a loop. If the job status is not running, pending or waiting, the job output is read. 

"""
Ansible job monitor

@author Stefan Schnell <mail@stefan-schnell.de>
@version 0.5.1

@param {string} in_userName - Name of the Ansible user
@param {SecureString} in_password - Password of the Ansible user
@param {string} in_jobTemplateName - Name of the AAP job template
@param {string} in_data - Additional data to send to the server
@param {SecureString} in_sshPassword - Password of the SSH connection
@param {string} in_inventoryName - Name of the AAP inventory
@param {string} in_credentialName - Name of the AAP credential
@outputType Properties

Important hint:
The "Prompt on Launch" field, next to the "Inventory" and/or
"Credential" field in the Ansible Automation Platform, must be enabled,
when using the parameters in_inventoryName and/or in_credentialName.

@example
in_userName = myAAPName
in_password = secret
in_jobTemplateName = myAAPTestTemplate
in_data = {"extra_vars": {"name": "Stefan", "password": "secret"}}
in_sshPassword = secret
in_inventoryName = myAAPInventoryName
in_credentialName = myAAPCredentialName
"""

import ast
import base64
import http.client
import json
import math
import re
import ssl
import time
import urllib.error
import urllib.request


def ansibleApiCall(
    userName: str,
    userPassword: str,
    apiCall: str,
    method: str = "GET",
    data: str | None = None
) -> dict | bytes | None:
    """ Sends a request to the Ansible endpoint
    """

    url: str = "https://yourAnsibleAutomationPlatformURL"
    url += apiCall

    response: http.client.HTTPResponse | None = None
    result: dict | bytes | None = None

    try:

        request = urllib.request.Request(
            url = url,
            method = method,
            data = data
        )

        print(f"Ansible {method} request: {url}")
        if data:
            print(f"with {data}")

        authorization: str = base64.b64encode(
            bytes(userName + ":" + userPassword, "UTF-8")
        ).decode("UTF-8")
        authorization = "Basic " + authorization

        request.add_header("Authorization", authorization)
        request.add_header("Content-Type", "application/json")

        response = urllib.request.urlopen(
            request,
            context = ssl.create_default_context()
            #context = ssl._create_unverified_context()
        )

        responseCode: int = response.status
        responseRead: bytes = response.read()

        if responseCode in (200, 201, 202, 204):
            if len(responseRead) > 0:
                result = json.loads(responseRead)
            else:
                result = response
        else:
            raise urllib.error.HTTPError(
                url,
                responseCode,
                f"API call failed with status code {responseCode}",
                response.headers,
                None
            )

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at ansibleApiCall: {err}"
        ) from err

    finally:
        if response is not None:
            response.close()

    return result


def jsonStringToDict(
    jsonString: str
) -> dict:
    """ Converts JSON string to dictionary via Abstract Syntax Tree (AST)
    """

    jsonDict: dict = {}

    try:
        jsonDict = json.loads(jsonString)
    except:
        try:
            jsonDict = ast.literal_eval(jsonString)
        except Exception as err:
            raise Exception("An error occurred at " + \
                "JSON string to dictionary") from err

    return jsonDict


def getIdOfInventory(
    userName: str,
    userPassword: str,
    inventoryName: str
) -> int:
    """ Detects the ID of an Ansible inventory
    """

    try:

        cntInventories: int = int(jsonStringToDict(str(
            ansibleApiCall(
                userName,
                userPassword,
                "/api/controller/v2/inventories/?page_size=1"
            )
        ))["count"])

        inventories: dict = {"count": cntInventories, "results": []}

        pageSize: int = 64
        totalPages = math.ceil(cntInventories / pageSize)

        for page in range(1, totalPages + 1):

            pageResponse = jsonStringToDict(str(
                ansibleApiCall(
                    userName,
                    userPassword,
                    f"/api/controller/v2/inventories/"
                    f"?page_size={pageSize}&page={page}"
                )
            ))

            inventories["results"].extend(pageResponse.get("results", []))

        for inventory in inventories["results"]:
            if (inventory["name"] == inventoryName):
                return inventory["id"]

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at getIdOfInventory: {err}"
        ) from err


def getIdOfCredential(
    userName: str,
    userPassword: str,
    credentialName: str
) -> int:
    """ Detects the ID of an Ansible credential
    """

    try:

        cntCredentials: int = int(jsonStringToDict(str(
            ansibleApiCall(
                userName,
                userPassword,
                "/api/controller/v2/credentials/?page_size=1"
            )
        ))["count"])

        credentials: dict = {"count": cntCredentials, "results": []}

        pageSize: int = 64
        totalPages = math.ceil(cntCredentials / pageSize)

        for page in range(1, totalPages + 1):

            pageResponse = jsonStringToDict(str(
                ansibleApiCall(
                    userName,
                    userPassword,
                    f"/api/controller/v2/credentials/"
                    f"?page_size={pageSize}&page={page}"
                )
            ))

            credentials["results"].extend(pageResponse.get("results", []))

        for credential in credentials["results"]:
            if (credential["name"] == credentialName):
                return credential["id"]

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at getIdOfCredential: {err}"
        ) from err


def getIdOfJobTemplate(
    userName: str,
    userPassword: str,
    jobTemplateName: str
) -> tuple[ int, bool, bool ]:
    """ Detects the ID of an Ansible job template
    """

    try:

        cntTemplates: int = int(jsonStringToDict(str(
            ansibleApiCall(
                userName,
                userPassword,
                "/api/controller/v2/job_templates/?page_size=1"
            )
        ))["count"])

        templates: dict = {"count": cntTemplates, "results": []}

        pageSize: int = 32
        totalPages = math.ceil(cntTemplates / pageSize)

        for page in range(1, totalPages + 1):

            pageResponse = jsonStringToDict(str(
                ansibleApiCall(
                    userName,
                    userPassword,
                    f"/api/controller/v2/job_templates/"
                    f"?page_size={pageSize}&page={page}"
                )
            ))

            templates["results"].extend(pageResponse.get("results", []))

        for template in templates["results"]:
            if (template["name"] == jobTemplateName):
                return (
                    template["id"],
                    template["ask_inventory_on_launch"],
                    template["ask_credential_on_launch"]
                )

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at getIdOfJobTemplate: {err}"
        ) from err

    return None, None, None


def launchJobTemplate(
    userName: str,
    userPassword: str,
    templateId: str,
    data: str | None = None
) -> dict:
    """ Invokes an Ansible job template
    """

    try:

        result: str = str(
            ansibleApiCall(
                userName,
                userPassword,
                f"/api/controller/v2/job_templates/{templateId}/launch/",
                "POST",
                data.encode() if data else None
            )
        )

        return jsonStringToDict(result)

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at launchJobTemplate: {err}"
        ) from err


def getJobOutput(
    userName: str,
    userPassword: str,
    jobId: str
) -> dict:
    """ Gets the output of an executed Ansible job
    """

    try:

        result: str = str(
            ansibleApiCall(
                userName,
                userPassword,
                f"/api/controller/v2/jobs/{jobId}/stdout/?format=json"
            )
        )

        return jsonStringToDict(result)

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at getJobOutput: {err}"
        ) from err


def getJobEvents(
    userName: str,
    userPassword: str,
    jobId: str
) -> dict:
    """ Gets the events of an executed Ansible job
    """

    try:

        result: str = str(
            ansibleApiCall(
                userName,
                userPassword,
                f"/api/controller/v2/jobs/{jobId}/job_events"
            )
        )

        return jsonStringToDict(result)

    except Exception as err:
        raise RuntimeError(
            f"An error occurred at getJobEvents: {err}"
        ) from err


def handler(context: dict, inputs: dict) -> dict:

    outputs: dict = {}
    jobOutput: dict = {"content": ""}
    jobEvents: dict = {}

    userName: str = inputs["in_userName"]
    userPassword: str = inputs["in_password"]
    nameOfJobTemplate: str = inputs["in_jobTemplateName"]

    try:

        jobTemplateId: int
        askInventory: bool
        askCredential: bool

        jobTemplateId, askInventory, askCredential = getIdOfJobTemplate(
            userName,
            userPassword,
            nameOfJobTemplate
        )

        if not jobTemplateId:
            raise ValueError("Error at job template Id")

        payload: dict = {}


        if inputs.get("in_data"):
            additionalPayload: dict = json.loads(inputs["in_data"])
            payload.update(additionalPayload)

        if inputs.get("in_inventoryName") and askInventory:

            inventoryId: int = getIdOfInventory(
                userName,
                userPassword,
                inputs["in_inventoryName"]
            )

            if not inventoryId:
                raise ValueError("Error at inventory id")

            payload["inventory"] = inventoryId

        if inputs.get("in_credentialName") and askCredential:

            credentialId: int = getIdOfCredential(
                userName,
                userPassword,
                inputs["in_credentialName"]
            )

            if not credentialId:
                raise ValueError("Error at credential id")

            payload["credentials"] = [credentialId]

        else:

            if inputs["in_sshPassword"]:

                payload["credential_passwords"] = {
                    "ssh_password": inputs["in_sshPassword"]
                }

        data: str = json.dumps(payload)

        newJob: dict = launchJobTemplate(
            userName,
            userPassword,
            str(jobTemplateId),
            data
        )

        newJobId: int = newJob["id"]

        if not newJobId:
            raise ValueError("Error at new job Id")

        jobStatus: str | None = None
        jobSummary: dict | None = None

        while (
            jobStatus is None or
            jobStatus in ("running", "pending", "waiting")
        ):

            jobSummary = jsonStringToDict(str(ansibleApiCall(
                userName,
                userPassword,
                f"/api/controller/v2/jobs/{newJobId}"
            )))

            jobStatus: str = jobSummary["status"]

            time.sleep(0.125)

        print(f"Job with Id {newJobId} done, with the status {jobStatus}")

        jobOutput = getJobOutput(
            userName,
            userPassword,
            str(newJobId)
        )

        jobEvents = getJobEvents(
            userName,
            userPassword,
            str(newJobId)
        )["results"]

        changedValue: dict | None = None
        for jobEvent in jobEvents:
            if jobEvent.get("event") == "playbook_on_stats":
                changedValue = (
                    jobEvent.get("event_data", {}).get("changed", {})
                )
                break

        if changedValue is not None:
            print(f"Changed value: {sum(changedValue.values())}")
        else:
            print("Can not determine Changed value")

            totalChanges: int = 0
            for jobEvent in jobEvents:
                if (
                    jobEvent.get("event") == "runner_on_ok" and
                    jobEvent.get("changed") is True
                ):
                    total_changes += 1

            print(f"Changed value from fallback: {total_changes}")

        outputs = {
            "status": "done",
            "error": None,
            "newJobId": newJobId,
            "jobSummary": jobSummary,
            "jobEvents": jobEvents,
            "jobOutput": jobOutput["content"]
        }

    except Exception as err:

        errorMessage: str = (
            str(err) if str(err) 
            else err.__class__.__name__
        )

        print(f"ERROR: {errorMessage}")

        outputs = {
            "status": "incomplete",
            "error": errorMessage,
            "newJobId": None,
            "jobSummary": None,
            "jobEvents": None,
            "jobOutput": None
        }

    return outputs

The following images show the execution of the above action in VCF Automation. The Test Ping template is executed, which performs a ping against the local target system, the managed node.

vcf automation call of ansible job template
result of vcf automation call of ansible job template
The following image shows the same execution but from the perspective of the Ansible Automation Platform.

result of vcf automation call in ansible automation platform

Conclusion

The seamless integration of Ansible IT process automation into VCF data center automation is easily possible. The combination of Red Hat Ansible and VMware Cloud Foundation Automation allows an expansion of the automation spectrum.