本記事はDockerで始めるStackstorm再入門1/3(環境構築からOrquestaで書いたWorkflowの結果をslackに通知する)の第3部です。

0. 目次と構成図

1. Actionの実行スクリプトをshellscriptとpythonで書く場合の違い
1. pythonで書いたActionの実行スクリプトの書き方
1. mockでActionの実行スクリプトをテストする
1. Workflowで引数や結果に応じて分岐させる書き方(python-script編)

本記事では第2部で予告していたとおり、pythonでActionの実行スクリプト、より柔軟なworkflowの書き方についてお話していきます。

shellscriptでリモートリポジトリのステータスの確認とコンテナの再立ち上げを行いましたが、今度はそれをpythonでおこなってみます。
pythonでActionの実行スクリプトを書くと、shellscriptに比べてActionの返却値(return value)を柔軟に定義することができたり、mockを使ってテストコードも書くことができるので、よりきめ細かくActionを書くことができます。

Dockerで始めるStackstorm再入門2/3(条件分岐させるWorkflowと定期実行させるRuleの書き方)より

本記事でお話する構成図

※1
Stackstormコンテナ内から同コンテナ外にあるアプリ(flask/nginx)コンテナに接続するために、
ホストのDockerデーモンのソケットファイル(var/run/docker.sock)をマウントさせています。

※2
Gtihubからリポジトリをcloneする際、ssh鍵やパスワードを使わずにおこなうべく、Personal access tokensを使いました。
トークンの取得方法は下記の記事をご参照ください。
私はrepoをスコープとしてトークンを取得しました。

参考
Creating a personal access token for the command line - GitHub Help

1. Actionの実行スクリプトをshellscriptとpythonで書く場合の違い

Actionの成否判定として使えるものがshellscriptではリターンコードとstdout/stderrですが、pythonでは、加えてrunメソッドの任意の返却値も使えます。
※シェルスクリプトでいうreturn_codeはpythonスクリプトでは、exit_codeと表記されています。

local-shell-script (shellscript)
return_code(0はActionのsuceeded, 0以外がfailed)
stdout/stderr
python-script
exit_code(0はActionのsuceeded, 0以外がfailed)
stdout/stderr
runメソッドの返却値

2. pythonで書いたActionの実行スクリプトの書き方

コードをひととおり確認したい方はGithubをご参照ください。
mydemo_pack/actions/scripts/git_status.py

おおむね書き方はpythonでプログラムを書く場合と同じですが、いくつかStackstorm独自の点があるので、そこだけ共有します。

エントリーポイントはrunメソッド
Actionの結果はstatusとexit_codeの他に任意のrunメソッドの返却値で制御可能
- mydemo_packではdict型のresultを用意(write_resultメソッドで定義)
__init__メソッドで/opt/stackstorm/configs/*.yamlを読み込み、self.configとして保持
- mydemo_packでは用意していないので実質不要

参考:

[Actions — StackStorm 3.1.0 documentation](Python actions can store arbitrary configuration in the configuration file which is global to the whole pack. The configuration is stored in a file named .yaml, in the /opt/stackstorm/configs/ directory.)

/opt/stackstorm/packs/mydemo_pack/actions/scripts/git_status.py

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

import re
import traceback
from st2actions.runners.pythonrunner import Action

from common_mydemo import Common

class GitStatusAction(Action):
    """Action class for st2"""

    def __init__(self, config):
        self.config = config
        self.result = {}
        for r in [
            "command", "branch", "expected", "bool", "stdout", "stderr"        
        ]:
            self.result[r] = None 

        self.common = Common()
        return
    (略)
    

    def check_stdout(self, branch, expected, stdout):
        success = False
        ptn = self._set_regex(branch, expected)
        for line in stdout:
            if not success and ptn.search(line):
                success = True
            else:
                pass

        return success, stdout
    (略)
       

    def write_result(self, command, branch, expected, bool, stdout, stderr):
        self.result.update({
            "command": command,
            "branch": branch,
            "expected": expected,
            "bool": bool,
            "stdout": self._to_str(stdout),
            "stderr": self._to_str(stderr),
        })

        return self.result
    

    def run(self, working_dir, branch, expected):
        """ Entrypoint for st2 """

        bool = False
        command = ''
        stdout = []
        stderr = []

        try:
            # git fetchのコマンドを生成
            cmd = self._git_fetch(working_dir, branch)
            # git fetchを実行するが、成否だけ知りたいのでself.common.execute_commandメソッドの返却値のstdout, stderrは_とし、明示的に不要とした
            bool, _, _ = self.common.execute_command(cmd)
            if bool:
                # git statusのコマンドを生成
                command = self.set_command(working_dir)
                # git statusを実行し、bool==Trueである場合、stdoutの結果を後続のself.check_stdoutで判定
                bool, stdout, stderr = self.common.execute_command(command)
                if bool:
                    bool, stdout = self.check_stdout(branch, expected, stdout)
                    self.result = self.write_result(command, branch, expected, bool, stdout, stderr)
        except:
            # try文のなかでエラーを引いた場合、tracebackモジュールでキャッチし、stderrに代入
            stderr = traceback.format_exc()
            self.result = self.write_result(command, branch, expected, bool, stdout, stderr)

        #finally:
        #    self.result = self.write_result(command, branch, expected, bool, stdout, stderr)
            
        
        return self.result

コマンドの実行はsubprocessモジュールを使ってactions/scripts/common_mydemo.pyとしてまとめています。
こちらのpythonスクリプトは任意のActionのpythonスクリプトが内部的に呼ぶのでStackstormが認識する必要はないため、/opt/stackstorm/packs/mydemo_pack/*.yamlのメタファイルを別途用意する必要もないです。

/opt/stackstorm/packs/mydemo_pack/actions/scripts/common_mydemo.py

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

import sys
import traceback
from subprocess import Popen, PIPE


class Common:
    """ Common for mydemo """
    
    def __init__(self):
        pass
    
    def execute_command(self, command):
        bool = False
        try:
            stdout, stderr = Popen(
                command, shell=True, stdout=PIPE, stderr=PIPE
            ).communicate()
            stdout = stdout.splitlines()
            stderr = stderr.splitlines()
            bool = True
        except:
            stdout = None
            stderr = traceback.format_exc()
                
        return bool, stdout, stderr

3. mockでActionの実行スクリプトをテストする

サンプルのmydemo_packが手元にある方は/opt/stackstorm/packs/mydemo_pack/tests/の下にテストコードなどが用意されていることが確認できるはずです。

root@$HOSTNAME:/# tree -L 3 /opt/stackstorm/packs/mydemo_pack/tests/
/opt/stackstorm/packs/mydemo_pack/tests/
├── fixtures              # runメソッドに渡す引数を定義
│   ├── git_status.yaml
│   └── rebuild_app.yaml
├── git_status　　　　　　　# mockオブジェクトに渡すsubmoduleメソッドの値のstdout/stderrを定義
│   └── response.yaml
├── rebuild_app　　　　　　 # mockオブジェクトに渡すsubmoduleメソッドの値のstdout/stderrを定義
│   └── response.yaml
├── test_git_status.py　　# テストコード
└── test_rebuild_app.py   # 同上

3 directories, 7 files

mockを使わず、そのままテストコードからActionの実行スクリプトを走らせてみる

対象のテストコードは/opt/stackstorm/packs/mydemo_pack/tests/test_git_status.py
テストはStackstormが提供するst2tests.base.BaseActionTestCaseを継承してTestGitStatusActionインスタンスを生成して実行
生成したActionインスタンスの直下にaction_cls = $CLASSを書く
- e.g. action_cls = GitStatusAction
runメソッドに渡す引数は/opt/stackstorm/packs/mydemo_pack/tests/fixtures/内の任意のyamlで定義
- test_git_status.pyの場合は/opt/stackstorm/packs/mydemo_pack/tests/fixtures/git_status.yaml
テストの対象のActionインスタンスは以下のように生成
- action = self.get_action_instance()

テストの実行コマンド

st2-run-pack-tests -p $PACKPATH
- /opt/stackstorm/packs/tests/*.pyのテストコードを実行

root@$HOSTNAME:/# st2-run-pack-tests -p /opt/stackstorm/packs/mydemo_pack

st2-run-pack-tests -p $PACKPATH -f $TESTFILENAME
- /opt/stackstorm/packs/tests/\$TESTFILENAMEを実行

root@$HOSTNAME:/# st2-run-pack-tests -p /opt/stackstorm/packs/mydemo_pack \
> -f test_git_status

st2-run-pack-tests -p $PACKPATH -f $TESTFILENAME:$METHOD
- /opt/stackstorm/packs/tests/\$TESTFILENAMEの$METHODを実行

root@$HOSTNAME:/# st2-run-pack-tests -p /opt/stackstorm/packs/mydemo_pack \
> -f test_git_status:TestGitStatusAction.test00_no_mock_st2

参考:
Pack Testing — StackStorm 3.1.0 documentation Instantiating and obtaining class instances

テストコード

/opt/stackstorm/packs/mydemo_pack/tests/test_git_status.py

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

from st2tests.base import BaseActionTestCase
from mock import MagicMock, patch
import json
import yaml
import os
import sys
import re

BASE_DIR = '/opt/stackstorm/packs/mydemo_pack'
sys.path.append(BASE_DIR)
sys.path.append(BASE_DIR + '/actions/scripts')
sys.path.append('/opt/stackstorm/virtualenvs/mydemo_pack/lib/python2.7/site-packages')
sys.path.append('/opt/stackstorm/st2/lib/python2.7/site-packages')

input_file = "git_status.yaml"
res_file = BASE_DIR + "/tests/git_status/response.yaml"

from git_status import GitStatusAction

class TestGitStatusAction(BaseActionTestCase):

    # テスト対象のActionの実行スクリプトのclass名を書く
    action_cls = GitStatusAction

    def test00_no_mock_st2(self):
        input = yaml.load(
            self.get_fixture_content(input_file), Loader=yaml.FullLoader
        )
        
        # Actionインスタンスを生成
        action = self.get_action_instance()
        # runメソッドを実行
        result = action.run(**input)
        print('result: {r}'.format(r=result))

        self.assertEquals(len(result), 6)
        self.assertEqual(result["bool"], True)

runメソッドの引数が書かれた$PACK/tests/fixtures/*.yaml

/opt/stackstorm/packs/mydemo_pack/tests/fixtures/git_status.yaml

"working_dir": "/usr/src/app/flask-docker"
"branch": "devel-views"
"expected": "up_to_date"

mockを使って、任意のメソッドの返却値を書き換える

/opt/stackstorm/packs/mydemo_pack/tests/test_git_status.py

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

from st2tests.base import BaseActionTestCase
from mock import MagicMock, patch
import json
import yaml
import os
import sys
import re

BASE_DIR = '/opt/stackstorm/packs/mydemo_pack'
sys.path.append(BASE_DIR)
sys.path.append(BASE_DIR + '/actions/scripts')
sys.path.append('/opt/stackstorm/virtualenvs/mydemo_pack/lib/python2.7/site-packages')
sys.path.append('/opt/stackstorm/st2/lib/python2.7/site-packages')

input_file = "git_status.yaml"
res_file = BASE_DIR + "/tests/git_status/response.yaml"

from git_status import GitStatusAction

class TestGitStatusAction(BaseActionTestCase):

    action_cls = GitStatusAction

    (略)

    # mockオブジェクトで書き換えるメソッド
    @patch("common_mydemo.Common.execute_command")
    def test01_mock_st2(self, execute):
        input = yaml.load(
            self.get_fixture_content(input_file), Loader=yaml.FullLoader
        )

        def _execute_command(_cmd):
            bool = True
            #stdout = ["Your branch is up-to-date with 'origin/devel-views'"]
            #stderr = [""]

            # yamlからstdout/stderrの値を取得
            res = yaml.load(open(res_file), Loader=yaml.FullLoader)
            stdout = res["succeeded"]["up_to_date"]["stdout"]
            stderr = res["succeeded"]["up_to_date"]["stderr"]

            return bool, stdout, stderr

        # _execute_commandメソッドの値をside_effectに代入
        execute.side_effect = _execute_command

        # actionインスタンスを生成し、runメソッドを実行
        action = self.get_action_instance()
        result = action.run(**input)
        print('result: {r}'.format(r=result))

        self.assertEquals(len(result), 6)
        self.assertEqual(result["bool"], True)
        self.assertEqual(
            result["command"], 
            "cd /usr/src/app/flask-docker && sudo git status"
        )
        ans = yaml.load(open(res_file), Loader=yaml.FullLoader)
        self.assertEqual(
            result["stdout"], 
            '\n'.join(map(str, ans["succeeded"]["up_to_date"]["stdout"]))
        )
        self.assertEqual(
            result["stderr"], 
            '\n'.join(map(str, ans["succeeded"]["up_to_date"]["stderr"]))
        )

    @patch("common_mydemo.Common.execute_command")
    def test02_mock_st2_not_up_to_date(self, execute):
        input = yaml.load(
            self.get_fixture_content(input_file), Loader=yaml.FullLoader
        )

        # runメソッドの引数の値はテストコードのなかで書き換える
        input.update({
            'expected': 'not_up_to_date'
        })

        def _execute_command(_cmd):
            bool = True
            #stdout = ["Your branch is up-to-date with 'origin/devel-views'"]
            #stderr = [""]

            res = yaml.load(open(res_file), Loader=yaml.FullLoader)
            stdout = res["succeeded"]["not_up_to_date"]["stdout"]
            stderr = res["succeeded"]["not_up_to_date"]["stderr"]

            return bool, stdout, stderr

        execute.side_effect = _execute_command

        action = self.get_action_instance()
        result = action.run(**input)
        print('result: {r}'.format(r=result))

        self.assertEquals(len(result), 6)
        self.assertEqual(result["bool"], True)
        self.assertEqual(
            result["command"], 
            "cd /usr/src/app/flask-docker && sudo git status"
        )
        ans = yaml.load(open(res_file), Loader=yaml.FullLoader)
        self.assertEqual(
            result["stdout"], 
            '\n'.join(map(str, ans["succeeded"]["not_up_to_date"]["stdout"]))
        )
        self.assertEqual(
            result["stderr"], 
            '\n'.join(map(str, ans["succeeded"]["not_up_to_date"]["stderr"]))
        )

※異常処理の動作確認

.py


def _execute_command(_cmd):
  raise Exception("err_message")

execute.side_effect = _execute_command

/opt/stackstorm/packs/mydemo_pack/tests/git_status/response.yaml

"succeeded":
  "up_to_date":
    "stdout":
      - "'On branch devel-views'"
      - "Your branch is up-to-date with 'origin/devel-views'."
      - "''"
      - "'nothing to commit, working directory clean'"
    "stderr":
      - ""
  "not_up_to_date":
    "stdout":
      - "hoge"
    "stderr":
      - ""
"failed":
  "up_to_date":
    "stdout":
      - ""
    "stderr":
      - ""
  "not_up_to_date":
    "stdout":
      - ""
    "stderr":
      - "hoge"

4. Workflowで引数や結果に応じて分岐させる書き方(python-script編)

基本的には引数の値に応じて分岐させるときと同じだが、boolean値やintegerのときはstringに型変換する必要がある
- 値そのものはresult().$RETURN_VALUEで取得できる

.yaml


next:
  - when: <% str(result().result.bool) = "true" %>

Actionの返却値を次のAction以降で使うために変数化(publish)する方法

.yaml


next:
  - when: <% succeeded() and str(result().result.bool) = "true" and (ctx().expected = 'up_to_date') %>
    do: last
    publish:
      - bool: <% result().result.bool %>
      - commnad: <% result().result.command %>
      - expected: <% result().result.expected %>
      - stdout: <% result().result.stdout %>
      - stderr: <% result().result.stderr %>

値を複数行に分けて書く方法

.yaml


key: |-
   val1 <% result().result.val1 %>
   val2 <% result().result.val2 %>

Workflow全体

/opt/stackstorm/packs/mydemo_pack/actions/workflows/poll-repo-python.yaml

version: 1.0

description: poll remote repo

input:
  - working_dir
  - branch
  - expected
  - ptns
  - timeout

output:
  - failed: <% ctx().failed %>
  - action_name: <% ctx().action_name %>

tasks:
  init: # failedフラグを初期化(False)とするだけのAction
    action: core.noop
    next:
    - publish:
        - failed: False
        - action_name: 'poll-repo-python'
      do: git_status_before_merged
  
  git_status_before_merged:   # ローカルリポジトリは最新であることを想定してgit status
    action: mydemo_pack.git_status_python
    input:
      working_dir: <% ctx().working_dir %>
      branch: <% ctx().branch %>
      expected: <% ctx().expected %>   #デフォルト値は'up_to_date'
      timeout: <% ctx().timeout %>
    next:
     # ローカルリポジトリは最新であることが確認できた
      - when: <% succeeded() and str(result().result.bool) = "true" and (ctx().expected = 'up_to_date') %>
        do: last
        publish:
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [expected] <% result().result.expected %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
　　　　# ローカルリポジトリは最新ではないことが確認できた
      - when: <% succeeded() and str(result().result.bool) = "true" and (ctx().expected = 'not_up_to_date') %>
        do: git_merge
        publish:
          - failed: False
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [expected] <% result().result.expected %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
      # ローカルリポジトリは最新であることが確認できなかった
      - when: <% succeeded() and str(result().result.bool) = "false" and (ctx().expected = 'up_to_date') %>
        do: git_status_before_merged
        publish:
          - failed: True
          - expected: 'not_up_to_date'
      - when: <% failed() %>
        do: post_msg
        publish:
          - failed: True
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [expected] <% result().result.expected %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
  
  git_merge:   # git_status_before_mergedで、ローカルリポジトリは最新ではないことが確認出来きた場合のみ実行
    action: core.local
    input:
      cmd: sudo git merge origin/<% ctx().branch %>
      cwd: <% ctx().working_dir %>
      timeout: <% ctx().timeout %>
    next:
      - when: <% succeeded() %>
        do: git_status_after_merged
        publish:
          - expected: 'up_to_date'
      - when: <% failed() %>
        do: post_msg
        publish:
          - failed: True
          - action_result: |-
              [result]
              <% result() %>
  git_status_after_merged: # ローカルリポジトリは最新であるとしてgit_status_after_mergedを実行
    action: mydemo_pack.git_status_python
    input:
      working_dir: <% ctx().working_dir %>
      branch: <% ctx().branch %>
      expected: 'up_to_date'
      timeout: <% ctx().timeout %>
    next:
      - when: <% succeeded() and str(result().result.bool) = "true" %>
        do: rebuild_app
        publish:
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [expected] <% result().result.expected %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
      - when: <% failed() %>
        do: post_msg
        publish:
          - failed: True
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [expected] <% result().result.expected %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
      

  rebuild_app: # コンテナが既に立ち上がっていれば、停止/削除し、再立ち上げを図る
    action: mydemo_pack.rebuild_app_python
    input:
      working_dir: <% ctx().working_dir %>
      ptns: <% ctx().ptns %>
      timeout: <% ctx().timeout %>
    next:
      - when: <% succeeded() and str(result().result.bool) = "true" %>
        do: post_msg
        publish:
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
      - when: <% failed() or str(result().result.bool) = "false" %>
        do: post_msg
        publish:
          - failed: True
          - action_result: |-
              [bool] <% result().result.bool %>
              [commnad] <% result().result.command %>
              [stdout] <% result().result.stdout %>
              [stderr] <% result().result.stderr %>
  
  post_msg:
    action: slack.post_message
    input:
      message: |-
        [action_name] 
        <% ctx().action_name %>
        [failed]
        <% ctx().failed %>
        [action_result]
        <% ctx().action_result %>
    next:
      - do: last
  
  last:
    action: core.noop
    next:
      - when: <% ctx().failed %>
        do: fail

参考

Actions StackStorm 3.1.0 documentation
orqeustaのサンプルコード
- [github.com/StackStorm/st2/blob/master/contrib/examples/actions/workflows/orquesta-streaming-demo.yaml]
  (https://github.com/StackStorm/st2/blob/master/contrib/examples/actions/workflows/orquesta-streaming-demo.yaml)
- github.com/StackStorm/st2/blob/master/contrib/examples/actions/local_command_runner_print_to_stdout_and_stderr.yaml
テストコード
- Pack Testing — StackStorm 3.1.0 documentation
本記事で扱ったStackstormのサンプルコード
- gkzz/st2-docker-gkz
Stackstormの管理対象のアプリコンテナ
- gkzz/flask-docker
ActionとWorkflowのサンプルコード
- gkzz/mydemo_pack

P.S. Twitterもやってるのでフォローしていただけると泣いて喜びます:)

@gkzvoice

P.P.S. StackstormでLTしました！

StackstormというIFTTT的なツールをDockerコンテナに載せた小話