はじめに

Terraformを導入して1年程度で管理するawsアカウントやgcpプロジェクトが10程度に増えました。
日々の差分チェックを怠ると、いざリソース追加しようとして terraform planを実行したら、
Plan: 2 to add, 4 to change, 2 to destroy.
・・・とか表示されて、tf/tfstateファイルを修正してたら中々差分が解消されず気づいたら2時間経ってたとか。。

そんなterraformあるあるを回避するには地道な日々のチェックしかないのですが、
そのチェック（terraform init --upgrade ⇒からの⇒ terraform plan）を今までは
bash scriptで日次で実行して結果をslackへ通知してました。
今回はそのbashをpythonに置換した、という内容です。

環境

CentOS（7.3）※windows((python 3.6.5, Anaconda 4.5.4))でも動きました。
Python 2.7.5とPython 3.6.5の両方で試してます。

script

外部コマンドの実行と結果の取得

terraformer.py

#!/usr/bin/env python
# coding: utf-8
import os
import subprocess

log_file_path = "/terraform/log/terraform.log"
err_log_file_path = "/terraform/log/err_terraform.log"

def logger(f):
    def wrapper(*args, **kwargs):
        with open(log_file_path, "a") as log_file, open(err_log_file_path, "a") as err_log_file:
            stdout, stderr = f(*args, **kwargs)
            if stderr == "":
                log_file.write(stdout.decode("utf-8")) # decode処理を行わないとバイトとして扱われる。
                return stdout.decode("utf-8")
            else:
                err_log_file.write(stderr.decode("utf-8"))
                return stderr.decode("utf-8")
    return wrapper

@logger
def cmd(path, command):
    os.chdir(path)
# Popenの第1引数にコマンドを指定、第2/3引数で標準出力、標準エラーにパイプを第二引数、第三引数の指定によって繋ぐ
    result = subprocess.Popen(command.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = result.communicate() # 標準出力、標準エラーの取得
    return (stdout, stderr)

cmd関数は引数pathで指定したディレクトリへ移動してcommand引数で指定したコマンドを実行し、
標準出力(stdout)と標準エラー出力(stderr)を返します。@loggerでログファイルへの書き込みをデコレートしてて、
最終的にstdoutかstderrどちらかを返します。
subprocessの Popenクラスで、子(サブ)プロセスが起動してコマンド実行して、結果を受け取ったりできるそうです。
- インスタンス化する時の、第1引数にコマンドを指定して、第2/3引数にtdout/stderrを指定して、
  stdout=subprocess.PIPEを指定すると新しいパイプが子プロセスに向けて？作られるそうです。
  ファイルオブジェクトを指定してもよいそうです。
- 最後にcommunicateメソッドでコマンドの結果をタプルで得られるそうです。

メイン処理

terraformer.py

#!/usr/bin/env python
# coding: utf-8
import re
import json

# terraform_path
aws_dir = "/terraform/aws/"
gcp_dir = "/terraform/gcp/"
# manage_config
prj_config = "/terraform/prj_config.json"
# load_configfile


def main():
    with open(prj_config, "r") as config_file: # terraform管理対象の一覧を読み込む
        prj = json.load(config_file)
    last_result = {}
    for prj_key in prj:
        dir = aws_dir if prj_key == "aws" else gcp_dir
        for prj_value in prj[prj_key]:
            #  tfファイルをgitlabで管理しているのでgit pullでアップデート
            cmd(dir+prj_value, "git pull")
            # `terraform init --upgrade` で初期化とprovider versionのアップデート
            cmd(dir+prj_value, "terraform init --upgrade")
            # terraform planの結果を格納
            result = cmd(dir+prj_value, "terraform plan")
            # 結果をextraction関数で判定してdictに格納
            last_result[prj_value] = extraction(result)
    to_slack(last_result)


def extraction(plan):
    if "No changes" in plan:
        line_extraction = re.findall("No changes.*", plan) #検索してマッチした行をリストで返す
        result = "".join(line_extraction) #リストをstrにしてる
        return result
    elif "Plan" in plan:
        line_extraction = re.findall("Plan.*", plan)
        result = "".join(line_extraction)
        return result
    elif "Error" in plan:
        line_extraction = re.findall("Error.*", plan)
        result = "".join(line_extraction)
        return result
    else:
        result = "予期せぬエラーです。ログを確認して下さい。"
        return result


if __name__ == "__main__":
    main()

まずwith open(prj_config, "r") as config_fileでterraformで管理している一覧（以下のjsonファイル）を読み込んでjson⇒dctへ変換してます。

prj_config.json

{
    "gcp": [
        "gcp_prod_a_project",
        "gcp_dev_a_project",
        "gcp_dev_b_project",
        "gcp_prod_b_project",
        "gcp_staging_a_project",
        "gcp_dev_c_project",
        "gcp_prod_c_project"
    ],
    "aws": [
        "aws_prod_a_account",
        "aws_dev_a_account"
    ]
}

次にプロジェクト（AWSはアカウント）毎に先程のcmd関数を使用して以下を実行してます。
- tfファイルをgitlabで管理しているので git pull でアップデート
- terraform init --upgrade で初期化とprovider versionのアップデート
- result = cmd(dir+prj_value, "terraform plan") で各プロジェクトのディレクトリへ移動しterraform planを実行して結果を格納
- 結果をextraction関数の引数に渡す
extraction関数はterraform planの結果の判定と必要な行の抽出を行いstrで返すようにしてます。
terraform planの結果は基本的に以下の3パターンが最後に出力されます。た…多分。
re.findall()で検索してマッチした行をリストで返し、"".join()でリストからstrに変換してます。

Plan: 0 to add, 1 to change, 0 to destroy.
No changes. Infrastructure is up-to-date.
Error: "Errorの内容"

全ての結果が格納された辞書(last_result)をto_slack関数の引数にしています。

ここでは要するに↓のdctを生成してto_slack関数の引数に渡しています。

{'gcp_dev_c_project': 'Plan: 0 to add, 1 to change, 0 to destroy.', 'gcp_dev_a_project': 'No changes. Infrastructure is up-to-date.', 'gcp_dev_b_project': 'No changes. Infrastructure is up-to-date.', 'aws_prod_a_account': 'No changes. Infrastructure is up-to-date.', 'aws_dev_a_account': 'Plan: 3 to add, 0 to change, 0 to destroy.', 'gcp_prod_a_project': 'Plan: 0 to add, 1 to change, 0 to destroy.', 'gcp_prod_b_project': 'No changes. Infrastructure is up-to-date.', 'gcp_staging_a_project': 'No changes. Infrastructure is up-to-date.', 'gcp_prod_c_project': 'Plan: 0 to add, 1 to change, 0 to destroy.'}

slackへの通知

terraformer.py

#!/usr/bin/env python
# coding: utf-8

import datetime
import requests

# slack_config
url = "https://hooks.slack.com/XXXXXXX/XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
username = "terraformer"
icon_emoji = ":terraform:"
header = """
Today's terraformplan    {0}
<https://"社内wikiのURL"|DocmentLink>  \
<https://"社内のgitlab"|ScriptFile>
""".format(datetime.date.today())


def to_slack(result_dict):
    for i in result_dict:
        result_dict[i] = re.sub(r"(\[32m)|(\[0m)", "", result_dict[i])
    # postするデータの初期値を設定
    post_json = {"text" : header, "username": username, "link_names": 1, "icon_emoji": icon_emoji, "attachments": []}
    for prj in result_dict:
        if "No changes" in result_dict[prj]:
            add_dict = {"color": "good", "title": prj, "text": result_dict[prj]} # "No changes"は"good"として緑色
            post_json["attachments"].append(add_dict)
        elif "Plan" in result_dict[prj]:
            add_dict = {"color": "warning", "title": prj, "text": result_dict[prj]}　# 何かしら差分があれは"warning"として橙色
            post_json["attachments"].append(add_dict)
        elif "Error" in result_dict[prj]:
            add_dict = {"color": "danger", "title": prj, "text": result_dict[prj]} # "Error"は"danger"として赤色
            post_json["attachments"].append(add_dict)
        else:
            add_dict = {"title": prj, "text": result_dict[prj]}
            post_json["attachments"].append(add_dict)
    requests.post(url, data=json.dumps(post_json)) #辞書(post_json)をjsonに変換してpostのデータとして渡す

引数で渡されたプロジェクト名と結果が入ったdctからjsonファイルを組み立てていきます。
slackへの通知自体はIncoming Webhooksで設定したurlへrequestsライブラリでpostしているだけです。
リッチなattachmentsを使って結果によってcolorを変えたかった（good/warning/danger ）ので、結果を判定した後に、
add_dictに結果（color、prj名、結果）を入れて、post_json["attachments"].append(add_dict) でattachmentsにネストして登録していきます。
↓のjsonを組み立てていくイメージです。

    "attachments": [
        {
            "color": "warning",
            "title": "gcp_dev_c_project",
            "text": "Plan: 0 to add, 1 to change, 0 to destroy."
        },
        {
            "color": "good",
            "title": "gcp_dev_a_project",
            "text": "No changes. Infrastructure is up-to-date."
        },
    ]

最後にrequests.post(url, data=json.dumps(post_json)) でdctからjsonに変換してpostして終了です！

※result_dict[i] = re.sub(r"(\[32m)|(\[0m)", "", result_dict[i])
⇒これは以下の様にterraform planの結果にカラーシーケンス（0m[32mとか）がついててダサかったので外しました。。

gcp_prod_a_project : No changes. Infrastructure is up-to-date.[0m[32m

結果

最後に

取りあえず目的を達成する事を念頭に、あまり拘ら無かったためさくっと出来ました。
プログラミング素人なので多分お粗末なコードなのでしょうが、先ずは自動化出来たの良しとしたいです。
bash scriptでええやんと言われればその通りです…。こういう機会を活かさないと中々勉強できないので
毎朝結果が通知されるようにしてますが、次はslack botから実行出来る様にしてみたいです。

pythonでterraform planを実行してslackへ通知する

はじめに

環境

script

外部コマンドの実行と結果の取得

メイン処理

slackへの通知

結果

最後に