【Ansible Tower×python】requestsモジュールでWorkflowの実行状況を確認したい

はじめに

• バージョン
  • Ansible Tower 3.8.0
  • Python 3.8.5
  • pandas 1.4.1
  • jmespath 0.10.0

　Ansible TowerにおけるWorkflowの実行状況を調べたいと思ったのですが、GUIでそのまま
見ても調べるのが大変なので、requestsとpandasモジュールを用いて情報を見やすく整理
することにしました。

1. APIから情報を取得する

　今回はWorkflowの実行状況を調べたいので、Tower API リファレンスガイドから、
/api/v2/workflow_jobs/を使うことにします。実行状況を取得する方法としては、
比較的簡単な以下2つを紹介します。
　なお、Ansible Towerのアドレスは1.1.1.と仮定します。

1-1. ユーザ名/パスワードを使用する場合

　ログインするときのユーザー名とパスワードを入力すればアクセスできるので一番簡単だと思われます。

import requests

url = "https://1.1.1.1/api/v2/workflow_jobs/"
user = "user"
password = "password"

result = requests.get(
    url,
    auth=(user, password),
    verify=False,
).json()

1-2. tokenを使用する場合

　トークンについてはAnsible Towerのホーム画面から、ユーザーごとの設定に移動して
トークンを発行してください。

import requests

url = "https://1.1.1.1/api/v2/workflow_jobs/"
token = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

result = requests.get(
    url,
    headers={"Authorization": "Bearer  " + token},
    verify=False,
).json()

2. 取得した情報をDataFrameに変換し、csv出力する

2-1. APIで取得した情報の確認

　前述のコードで取得した結果が以下のようになります。（不要な変数は一部削除）
各々のWorkflowの実行状況はresults配下に格納されているので、これらの情報を
jmespathで抽出してDataFrameに変換します。
　なお、データ数（count）が200を超えると、指定したURLでは全ての情報を取得できず、
nextやpreviousに記載されたURLにアクセスする必要があるので注意が必要です。
詳細は、Ansible Tower API Guide v3.8.0 - 7. Paginationを確認してください。

{
    "count": 10,
    "next": null,
    "previous": null,
    "results": [
        {
            "id": 16,
            "type": "workflow_job",
            "summary_fields": {
                "created_by": {
                    "id": 1,
                    "username": "admin",
                    "first_name": "",
                    "last_name": ""
                }
            },
            "name": "wf_sample",
            "status": "successful",
            "started": "2022-02-13T07:41:08.468880Z",
            "finished": "2022-02-13T07:41:11.324289Z",
        }
    ]
}

2-2. Workflowの実行状況取得用コード

　今回作成したコードは以下です。

• ポイント（1）
  • jmespathはjmespath.search(検索文字列, 検索する元となるデータ) の形で書く
  • results[].{使用したいキー名（任意）: 取得する変数}の形で書く
• ポイント（2）
  • to_datetimeで指定した列をdatetime型に変換
  • tz_convert("Asia/Tokyo")でTimezoneをUTC -> Asia/Tokyoに変換
• ポイント（3）
  • DataFrame名.to_csv(csv出力するパス)でDataFrameをcsv出力する

get_workflow_result.py

import jmespath
import requests
import pandas as pd
from pandas import DataFrame


def create_workflow_dataframe(url: str, token: str) -> DataFrame:
    # Ansible TowerにPOSTして結果を取得
    api_return = requests.get(
        url,
        headers={"Authorization": "Bearer  " + token},
        verify=False,
    ).json()
    # jmespathで必要な箇所を抽出 - ポイント（1）
    jmespath_api_return = jmespath.search(
        "results[].{id: id, type: type, name: name, status: status, \
        user: summary_fields.created_by.username, \
        started: started, finished: finished}",
        api_return,
    )
    df_workflow = pd.DataFrame(jmespath_api_return)
    # 時刻のフォーマットを変換 - ポイント（2）
    df_workflow["started"] = (
        pd.to_datetime(df_workflow["started"])
        .dt.tz_convert("Asia/Tokyo")
        .dt.strftime("%Y/%m/%d %H:%M:%S")
    )
    df_workflow["finished"] = (
        pd.to_datetime(df_workflow["finished"])
        .dt.tz_convert("Asia/Tokyo")
        .dt.strftime("%Y/%m/%d %H:%M:%S")
    )
    return df_workflow


def main() -> None:
    df = create_workflow_dataframe(
        "https://1.1.1.1/api/v2/workflow_jobs/",
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    )
    # DataFrameをcsvに出力 - ポイント（3）
    df.to_csv("./workflow.csv")


if __name__ == "__main__":
    main()

2-3. csv確認

　想定通り結果を取得することが出来ました。

参考記事

Tower API リファレンスガイド
Ansible Tower API Guide v3.8.0 - 7. Pagination
Ansible Tower Administration Guide v3.4.3 - 15. トークンベースの認証
pandas.to_datetime — pandas 1.4.1 documentation
pandas.Series.dt.tz_convert — pandas 1.4.1 documentation
pandas.DataFrame.to_csv — pandas 1.4.1 documentation