AWS ECS Exec のアウトプットを取得する

これまで何回か ECS Exec Command について触れてきましたが、コマンドを実行するまでに留まっていました。

では、そのコマンドの出力内容を得たい場合は、というのが今回です。

ECS Execute Command の過去記事

この辺です。

参考文献

前も今も、自分の見落としなだけかもですが、AWSドキュメントにこの辺って載ってないんです。そこで、ググったらすぐに救世主がいたのでリンクしておきます。

amazon web services – How can I get output from boto3 ecs execute_command? – Stack Overflow

ほぼ、これを流用させてもらいつつ、軽くコードを書いて動作確認していきます。

コード

自分の実際のコードから、あえてペライチに抜き出した内容になっています。

インストール

pip3 で入れるモノがあります。

pip3 install boto3 websocket-client construct

1	pip3 install boto3 websocket-client construct

クラス

コマンド実行と、その出力を得るメソッドを書きます。executeCommand は、タスク起動直後は数秒～十数秒は受け付けないので、リトライを仕込んでいます。

import boto3
import time
import json
import uuid
import websocket
import construct

class EcsExample:

    def __init__(self):
        self.session = boto3.session.Session()
        self.region  = self.session.region_name
        self.ecs     = boto3.client('ecs', region_name=self.region)

    def executeCommand(self, command, cluster, task_arn, container=None, interactive=True, retry=10):
        params = {
            "cluster"    : cluster,
            "task"       : task_arn,
            "command"    : command,
            "interactive": interactive,
        }
        if container: params["container"] = container

        sleep = 5
        for i in range(retry):
            try:
                res = self.ecs.execute_command(**params)
            except Exception as e:
                if "Wait and try again" in str(e):
                    logger.log(f"Not ready task = {task_arn} in {cluster}")
                    logger.log(f"Try again ... wait {sleep} seconds ...")
                    time.sleep(sleep)
                    continue

                logger.log(f"Failed execute command. task = {task_arn} in {cluster}")
                logger.log(f"Error: {e}")
                return False
            break

        return res

    def getExecuteCommandOutput(self, res):
        session    = res['session']
        connection = websocket.create_connection(session['streamUrl'])
        try:
            init_payload = {
                "MessageSchemaVersion": "1.0",
                "RequestId"           : str(uuid.uuid4()),
                "TokenValue"          : session['tokenValue']
            }
            connection.send(json.dumps(init_payload))

            AgentMessageHeader = construct.Struct(
                'HeaderLength' / construct.Int32ub,
                'MessageType' / construct.PaddedString(32, 'ascii'),
            )
            AgentMessagePayload = construct.Struct(
                'PayloadLength' / construct.Int32ub,
                'Payload' / construct.PaddedString(construct.this.PayloadLength, 'ascii')
            )
            while True:
                response = connection.recv()
                message  = AgentMessageHeader.parse(response)
                if 'channel_closed' in message.MessageType:
                    raise Exception('Channel closed before command output was received')

                if 'output_stream_data' in message.MessageType:
                    break

        finally:
            connection.close()

        payload_message = AgentMessagePayload.parse(response[message.HeaderLength:])
        return payload_message.Payload

import boto3

import time

import json

import uuid

import websocket

import construct

class EcsExample:

def __init__(self):

self.session = boto3.session.Session()

self.region = self.session.region_name

self.ecs = boto3.client('ecs', region_name=self.region)

def executeCommand(self, command, cluster, task_arn, container=None, interactive=True, retry=10):

params = {

"cluster" : cluster,

"task" : task_arn,

"command" : command,

"interactive": interactive,

}

if container: params["container"] = container

sleep = 5

for i in range(retry):

try:

res = self.ecs.execute_command(**params)

except Exception as e:

if "Wait and try again" in str(e):

logger.log(f"Not ready task = {task_arn} in {cluster}")

logger.log(f"Try again ... wait {sleep} seconds ...")

time.sleep(sleep)

continue

logger.log(f"Failed execute command. task = {task_arn} in {cluster}")

logger.log(f"Error: {e}")

return False

break

return res

def getExecuteCommandOutput(self, res):

session = res['session']

connection = websocket.create_connection(session['streamUrl'])

try:

init_payload = {

"MessageSchemaVersion": "1.0",

"RequestId" : str(uuid.uuid4()),

"TokenValue" : session['tokenValue']

}

connection.send(json.dumps(init_payload))

AgentMessageHeader = construct.Struct(

'HeaderLength' / construct.Int32ub,

'MessageType' / construct.PaddedString(32, 'ascii'),

)

AgentMessagePayload = construct.Struct(

'PayloadLength' / construct.Int32ub,

'Payload' / construct.PaddedString(construct.this.PayloadLength, 'ascii')

)

while True:

response = connection.recv()

message = AgentMessageHeader.parse(response)

if 'channel_closed' in message.MessageType:

raise Exception('Channel closed before command output was received')

if 'output_stream_data' in message.MessageType:

break

finally:

connection.close()

payload_message = AgentMessagePayload.parse(response[message.HeaderLength:])

return payload_message.Payload

実行

コマンドを実行し、直後に結果を得て、print します。

cluster  = "example"
task_arn = "arn:aws:ecs:ap-northeast-1:1234567890:task/example/1234567890abcdefg"
command  = "cat /etc/debian_version"

example  = EcsExample()
res      = example.executeCommand(command, cluster, task_arn)
output   = example.getExecuteCommandOutput(res)

print(output)

cluster = "example"

task_arn = "arn:aws:ecs:ap-northeast-1:1234567890:task/example/1234567890abcdefg"

command = "cat /etc/debian_version"

example = EcsExample()

res = example.executeCommand(command, cluster, task_arn)

output = example.getExecuteCommandOutput(res)

print(output)

結果はこんな感じ。Debianバージョンをゲッツ。

$ python3 execute_command_output.py
10.10

$ python3 execute_command_output.py

10.10

このファイルだと余計な改行が入っちゃうけど、echo -n YES > /tmp/test.txt とかで書き込んだファイルなら、ちゃんと改行もナシの文字列になります。

標準出力とエラー出力

結果を得ているコードはだいたいこの辺なんだけど、

response = connection.recv()
message  = AgentMessageHeader.parse(response)
...
payload_message = AgentMessagePayload.parse(response[message.HeaderLength:])

response = connection.recv()

message = AgentMessageHeader.parse(response)

...

payload_message = AgentMessagePayload.parse(response[message.HeaderLength:])

コマンドには標準出力・エラー出力・戻り値があるので、どうなっているのか確認してみます。

response の中身を print するのですが、それぞれ標準(0)・エラー(1)・両方(1) となるようにしています。

# cat /etc/debian_version
b'\x00\x00\x00toutput_stream_data              \x00\x00\x00\x01\x00\x00\x01\x80\x17\x86\xe9:\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xbdmL\x0fV\x82\xa8\x12\xcc2R`\xb8vB\x1b_%M\xad&d!C\t\xf6@<\xd6\x0eP\xf5P\xa40\xb8\x10:\x97~\xab)m\x99\x9f\xd2\xd1\xa9\x00\x00\x00\x01\x00\x00\x00\x0710.10\r\n'

# cat /etc/notfound
b"\x00\x00\x00toutput_stream_data              \x00\x00\x00\x01\x00\x00\x01\x80\x17\x87W\xb6\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x8b\xc5\xf3w\x85\xc5}\x0f\xda\xb6O\xd1\xa7\xf6L\x04\xefj\xfd\xc2>0\xc5O\xb6\xb4&q\x87'\xe7\xf3c\x15BU\xd9\x9bzn\xa7;\x9f\x1c\xb7\xcb\x15C\x00\x00\x00\x01\x00\x00\x00/cat: /etc/notfound: No such file or directory\r\n"

# cat /etc/debian_version /etc/notfound
b'\x00\x00\x00toutput_stream_data              \x00\x00\x00\x01\x00\x00\x01\x80\x17\x87\xffR\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xb0T\x8e\xf0>\xc5Jk\xdf`w\xe0\xdc\tD-t\x0c\x90\x11 F[\x9a\xe0\x06\xf2\x06\xa8\xef\xae\xb2t\xfa\xc1\xca\x11H?\xbe<L\xfan\xc5\xe0\xf96\x00\x00\x00\x01\x00\x00\x00610.10\r\ncat: /etc/notfound: No such file or directory\r\n'

# cat /etc/debian_version

b'\x00\x00\x00toutput_stream_data \x00\x00\x00\x01\x00\x00\x01\x80\x17\x86\xe9:\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xbdmL\x0fV\x82\xa8\x12\xcc2R`\xb8vB\x1b_%M\xad&d!C\t\xf6@<\xd6\x0eP\xf5P\xa40\xb8\x10:\x97~\xab)m\x99\x9f\xd2\xd1\xa9\x00\x00\x00\x01\x00\x00\x00\x0710.10\r\n'

# cat /etc/notfound

b"\x00\x00\x00toutput_stream_data \x00\x00\x00\x01\x00\x00\x01\x80\x17\x87W\xb6\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x8b\xc5\xf3w\x85\xc5}\x0f\xda\xb6O\xd1\xa7\xf6L\x04\xefj\xfd\xc2>0\xc5O\xb6\xb4&q\x87'\xe7\xf3c\x15BU\xd9\x9bzn\xa7;\x9f\x1c\xb7\xcb\x15C\x00\x00\x00\x01\x00\x00\x00/cat: /etc/notfound: No such file or directory\r\n"

# cat /etc/debian_version /etc/notfound

b'\x00\x00\x00toutput_stream_data \x00\x00\x00\x01\x00\x00\x01\x80\x17\x87\xffR\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xb0T\x8e\xf0>\xc5Jk\xdf`w\xe0\xdc\tD-t\x0c\x90\x11 F[\x9a\xe0\x06\xf2\x06\xa8\xef\xae\xb2t\xfa\xc1\xca\x11H?\xbe<L\xfan\xc5\xe0\xf96\x00\x00\x00\x01\x00\x00\x00610.10\r\ncat: /etc/notfound: No such file or directory\r\n'

ヘッダから先の文字列は、標準出力とエラー出力がシンプルに連結されているようです。返り値は……ヘッダに含んでなさそうですが、得る方法がゼロかは不明としておきます。

これらの区別がハッキリしないので、それらを複雑に扱わないよう、標準出力だけで済むように構成するのが無難そうです。

今回これを調べたのは、ExecCommandで少し長めのインストール処理をさせた時に、非同期処理になってしまうので、その処理が終わるのを正確に待機したかったからです。最後の処理で作られるべきファイルをチェックして、なかったら待機みたいな。

改善の余地はまだありそうですが、材料としては揃った感があるので、自動化処理で大抵のことは実現できそうになってよかったです:-)