AWS CloudWatch Logs Agentに埋め込みメトリクスを（力技で）導入する

# TL; DR

CloudWatch Agentが使えたら、そちらを使いましょう。



# はじめに

CloudWatch Logsには、埋め込みメトリクス (Embedded Metric) という機能が存在します。
こちらを利用すると、CloudWatch Logsに含まれる情報を、自動でCloudWatch Metricsへ転送してくれます。
ログファイルの構造とメトリクス抽出の設定を二重管理する必要がなくなるため、非常に便利な機能です。

> [高カーディナリティログの取り込みと CloudWatch 埋め込みメトリクスフォーマットによるメトリクスの生成 - Amazon CloudWatch](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format.html)

現在ではCloudWatch Agentを使用することが一般的ですが、かつては、CloudWatch Logs Agentというツールを使ってログファイルをCloudWatch Logsに送信していました。
名前が似ているので間違いやすいのですが、CloudWatch Logs Agentは旧ツールとなっており、使えなくなることが予想されます（と言われて数年経っているわけですが）。

> このリファレンスは、廃止が予定されている古い CloudWatch Logs エージェント用です。代わりに統合 CloudWatch エージェントを使用することを強くお勧めします。エージェント詳細については、CloudWatch エージェントを使用して Amazon EC2 インスタンスとオンプレミスサーバーからメトリクスとログを収集するを参照してください。
>
> [CloudWatch Logs エージェントのリファレンス - Amazon CloudWatch Logs](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/AgentReference.html)

CloudWatch Logs Agentの開発は、埋め込みメトリクスが実装される前には既に終了しており、今後も埋め込みメトリクスが実装されるのは望み薄でしょう。
しかし、CloudWatch Agentが利用できないプラットフォームでは、現在でもCloudWatch Logs Agentを使わざるをえない、そういった状況も存在するかと思います。
そのような状況下、埋め込みメトリクスが使えたら便利だなぁと思うこともありますよね。

それでも、どうしても埋め込みメトリクスを使いたい！
そんな時に、力技で利用可能にする方法を説明します。



# CloudWatch Logs Agentに埋め込みメトリクスを導入する方法

以降、CloudWatch Logs Agentに埋め込みメトリクスを導入する方法を、補足を交えつつ説明します。


## 事前準備

CloudWatch Logs AgentはPython2.6以上Python3.6未満での実行を想定して書かれています。
また、事前にAWS-CLIはインストール済みである必要があります。
利用する際は、以下のコマンドでサーバー等にセットアップします。

```bash:CloudWatch-Logs-Agentをセットアップするスクリプト
#!/usr/bin/bash

function restart_cloudwatch_logs_agent() {
  if [ -f /etc/system-release ] && [ $(cat /etc/system-release | grep "Amazon Linux release 2" | wc -l) -eq 1 ]; then
    # OSがAmazon Linux 2の場合
    sudo service awslogsd restart
  else
    # それ以外の場合
    sudo service awslogs restart
  fi
}

# セットアップツールの導入
curl https://s3.amazonaws.com/aws-cloudwatch/downloads/latest/awslogs-agent-setup.py -O

# セットアップの実行
sudo python awslogs-agent-setup.py --region ${REGION_NAME}

# サービスのリスタート
restart_cloudwatch_logs_agent()
```

> [AWS Command Line Interface とは - AWS Command Line Interface](https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-chap-welcome.html)
> [AWS CLI のインストール、更新、アンインストール - AWS Command Line Interface](https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-chap-install.html)
> [CloudWatch Logs エージェントのリファレンス - Amazon CloudWatch Logs](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/AgentReference.html)
> [AmazonLinux2の判定方法 | ハックノート](https://hacknote.jp/archives/43036/)

本記事では、ログの送信は既にできているものとして話を進めます。
できていない場合は、AWSのドキュメントを参考に設定ファイルおよびIAMロールを設定してください。

## そのままでは導入できない埋め込みメトリクス

本節は確認のための節であるため、埋め込みメトリクスの導入方法だけ知りたい場合は読み飛ばしても大丈夫です。

CloudWatch Logs Agentで送信する対象ログファイルに、以下のようなJSON形式の文字列を出力してみてください。
下記が正常に埋め込みメトリクスとして認識されていれば、 `time` が100のメトリクスとして登録されるはずです。

> ```json:「埋め込みメトリクスフォーマットの例とJSONスキーマ」節に記載されているJSON例
> {
>   "_aws": {
>     "Timestamp": 1574109732004,
>     "CloudWatchMetrics": [
>       {
>         "Namespace": "lambda-function-metrics",
>         "Dimensions": [["functionVersion"]],
>         "Metrics": [
>           {
>             "Name": "time",
>             "Unit": "Milliseconds"
>           }
>         ]
>       }
>     ]
>   },
>   "functionVersion": "$LATEST",
>   "time": 100,
>   "requestId": "989ffbf8-9ace-4817-a57c-e4dd734019ee"
> }
> ```
>
> [仕様: 埋め込みメトリクスフォーマット - Amazon CloudWatch](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html)

しかし実際には、CloudWatch Logsにログが送信されていることは確認できるものの、メトリクスは生成されません。


## 仕様と実装の確認そして導入

### 埋め込みメトリクスの仕様

まず、埋め込みメトリクスの仕様を確認してみましょう。
どうやら、APIのリクエストヘッダーに `x-amzn-logs-format: json/emf` を設定する必要があるみたいです。

> [仕様: 埋め込みメトリクスフォーマット - Amazon CloudWatch](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html)

実際、boto3ライブラリ（Pythonで利用可能なAWS-SDK）で埋め込みメトリクスを導入するには、以下のようなソースコードを書けば良いことがわかっています。

> ```py:boto3ライブラリを利用して埋め込みメトリクスを送信するPython実装例
> def add_emf_header(request, **kwargs):
>     request.headers.add_header('x-amzn-logs-format', 'json/emf')
>     print(request.headers) # Remove this after testing of course.
>
> # You can even do this with watchtower! Use this instead, client = watchtower_handler.cwl_client
> client = boto3.client('logs')
> client.meta.events.register_first('before-sign.cloudwatch-logs.PutLogEvents', add_emf_header)
> ```
>
> [Injecting custom http headers in boto3 · Issue #2251 · boto/boto3](https://github.com/boto/boto3/issues/2251#issuecomment-613022228)

### CloudWatch Logs Agentの実装

次に、CloudWatch Logs Agentの実装を確認してみましょう。
先駆者様方により、 `/var/awslogs/lib/python/site-packages/cwlogs/push.py` のソースコードが実行されていることが把握されています。
libディレクトリ名がlib64だったり、pythonディレクトリ名がpython2.6やpython3.5だったりするため、そこはセットアップした環境に合わせて適宜読み替えてください。
以降、 `/var/awslogs/lib/python/site-packages` ディレクトリまでのパスを `${PYTHON_PACKAGES}` と書き替えて説明します。

クラス図およびシーケンス図で表現すると、以下の通りです。
説明に不要な要素（クラス、メソッド、メソッドの引数など）は省略しています。

```plantuml
@startuml
title CloudWatch Logs Agentの主要な構造を表すクラス図

skinparam classAttributeIconSize 0
skinparam linetype ortho

package threading {
  class Thread {
    start()
  }
}

package awscli/customizations/commands {
  abstract BasicCommand {
    __call__()
    {abstract} _run_main()
  }
}

package cwlogs {
  package push {
    class LogsPushCommand {
      NAME = 'push'
      logs : logs
      _run_main()
      _call_push_file()
    }
    class Watcher {
      logs : logs
      __init__(logs : logs)
      register()
      _run()
    }
    class Stream {
      logs : logs
      logs(logs : logs)
      refresh()
    }
    class EventBatchPublisher {
      logs_service : logs
      __init__(logs_service : logs)
      _run()
    }
    abstract Publisher {
      logs_service : logs
      __init__(logs_service : logs)
      _publish_event_batch()
      _put_log_events()
    }
    class FileEventsReader {
      _run()
      _push_event()
    }
    class FileEventBatchReader {
      _push_event()
    }
  }
  package threads {
    abstract BaseThread {
      run()
      {abstract} _run()
    }
  }
}


Thread <|-- BaseThread
BaseThread <|-- Publisher
BaseThread <|-- Watcher
BaseThread <|-- FileEventsReader
Publisher <|-- EventBatchPublisher
FileEventsReader <|--- FileEventBatchReader

BasicCommand <|--- LogsPushCommand

Watcher "1" <- "1" LogsPushCommand
Watcher "*" *--> "1" Stream
EventBatchPublisher "1" <- "1" Stream
Stream "1" --> "1" FileEventBatchReader
@enduml
```

```plantuml
@startuml
title CloudWatch Logs Agentの主要な振舞いを表すシーケンス図

skinparam style strictuml
autoactivate on
autonumber

participant ": LogsPushCommand" as LogsPushCommand
participant ": Watcher" as Watcher
participant ": Stream" as Stream
participant ": EventBatchPublisher" as EventBatchPublisher
participant ": FileEventBatchReader" as FileEventBatchReader
participant "logs : logs" as logs
participant "logs_service : logs" as logs_service

-> LogsPushCommand: BasicCommand::~__call__()
LogsPushCommand -> LogsPushCommand: _run_main()
  LogsPushCommand -->> logs **
  LogsPushCommand -> LogsPushCommand: _call_push_file()
    LogsPushCommand -->> Watcher **
    LogsPushCommand -> Watcher: ~__init__(logs = logs)
      note right
        LogsPushCommandクラスがフロー3で生成したlogsオブジェクトを
        Watcherクラスのlogsオブジェクトに代入する。
      end note
      deactivate Watcher
    LogsPushCommand -> Watcher: register()
      Watcher -->> Stream **
      Watcher -> Stream: logs(logs = logs)
        note right
          LogsPushCommandクラスがフロー3で生成したlogsオブジェクトを
          Streamクラスのlogsオブジェクトに代入する。
        end note
        deactivate Stream
      deactivate Watcher
    LogsPushCommand -> Watcher: Thread::start()
      note right
        ※1 Thread::start()メソッドは、本来であれば非同期処理であるBaseThread::run()メソッドを走らせるだけのメソッドであるため表記としては正確ではないが、
        わかりやすさのため、また、この箇所は本質ではないため、Thread::start()メソッド内でBaseThread::run()メソッドを呼出した後も活性区間が存在するものとして表記する。
      end note
      Watcher ->> Watcher: BaseThread::run()
        Watcher -> Watcher: _run()
          Watcher -> Stream: refresh()
            Stream -->> EventBatchPublisher **
            Stream -> EventBatchPublisher: ~__init__(logs_service = logs)
              note right
                LogsPushCommandクラスがフロー3で生成したlogsオブジェクトを
                EventBatchPublisherクラスのlogs_serviceオブジェクトに代入する。
                なぜかオブジェクト名が変わる。
              end note
              EventBatchPublisher -> EventBatchPublisher: Publisher::~__init__(logs_service = logs_service)
                deactivate EventBatchPublisher
              deactivate EventBatchPublisher
            Stream -->> FileEventBatchReader **
            par
              Stream -> EventBatchPublisher: Thread::start()
                note right: ※1 と同様
                EventBatchPublisher ->> EventBatchPublisher: BaseThread::run()
                  EventBatchPublisher -> EventBatchPublisher: _run()
                    EventBatchPublisher -> EventBatchPublisher: Publisher::_publish_event_batch()
                      EventBatchPublisher -> EventBatchPublisher: Publisher::_put_log_events()
                        EventBatchPublisher -> logs_service: put_log_events()
                          note right: ログファイルを\nCloudWatch Logsに送信する。
                          deactivate logs_service
                        deactivate EventBatchPublisher
                      deactivate EventBatchPublisher
                    deactivate EventBatchPublisher
                  deactivate EventBatchPublisher
                deactivate EventBatchPublisher
            else
              Stream -> FileEventBatchReader: Thread::start()
                note right: ※1 と同様
                FileEventBatchReader ->> FileEventBatchReader: BaseThread::run()
                  FileEventBatchReader -> FileEventBatchReader: _run()
                    note right: ログファイルの変更を読込む。
                    deactivate FileEventBatchReader
                  deactivate FileEventBatchReader
                deactivate FileEventBatchReader
            end
            deactivate Stream
          deactivate Watcher
        deactivate Watcher
      deactivate Watcher
@enduml
```

`${PYTHON_PACKAGES}/cwlogs/push.py` のLogsPushCommandクラス、および、継承元クラスである `${PYTHON_PACKAGES}/awscli/customizations/commands.py` のBasicCommandクラスを確認すると、 `BasicCommand::__call__()` メソッド（フロー1）で呼出している `${PYTHON_PACKAGES}/cwlogs/push.py` 183行目の `LogsPushCommand::_run_main()` メソッド（フロー2）で、pushコマンドを動かしていることが確認できます。
また、 `${PYTHON_PACKAGES}/cwlogs/push.py` のWatcherクラス、Watcherクラスが呼出しているStreamクラス、Streamクラスが呼出しているEventBatchPublisherクラス、EventBatchPublisherクラスが継承しているPublisherクラス、Publisherクラスが継承している `${PYTHON_PACKAGES}/cwlogs/threads.py` のBaseThreadクラスを確認すると、 `BaseThread::run()` メソッド（フロー19）で呼出している `EventBatchPublisher::_run()` メソッド（フロー20）で呼出している `Publisher::_publish_event_batch()` メソッド（フロー21）で呼出している `Publisher::_put_log_events()` メソッド（フロー22）内の1247行目 `self.logs_service.put_log_events(**params)` メソッド（フロー23）で、実際にログファイルをCloudWatch Logsに送信していることが確認できます。
同様に、Streamクラス、Streamクラスが呼出しているFileEventBatchReaderクラス、FileEventBatchReaderクラスが継承しているFileEventsReaderクラスを確認すると、 `FileEventsReader::_run()` メソッド（フロー26）でログファイルの変更を読込んでいることが確認できます。

複雑ですね。

> [CloudWatch Logsエージェントの中身を（途中まで）調べてみた | mooapp](https://moomindani.wordpress.com/2014/07/11/cloudwatch-logs-agent-inside/)
> [CloudWatch Logs Agentの中身を探してみた - Qiita](https://qiita.com/tt2004d/items/fb36a2e751e3dd8985fb)
> [CloudWatch Logs Agent の挙動について調べたことのまとめ - Qiita](https://qiita.com/szk3/items/dba1071fb9d344bcd48c)

### CloudWatch Logs Agentへの埋め込みメトリクスの導入方法

上記2点をまとめると、CloudWatch Logs送信時にヘッダーとして `x-amzn-logs-format: json/emf` を設定すれば良いですね。

`${PYTHON_PACKAGES}/cwlogs/push.py` ファイルの `LogsPushCommand::_run_main()`メソッドは、以下のような実装になっているはずです。
非常にわかりづらいですが、 `self.logs = self._session.create_client('logs', **client_args)` で生成されている `self.logs` オブジェクトが、boto3（実際にはbotocore）で生成されたlogsクライアントです。

```diff_python:push.pyにおけるLogsPushCommandクラスの_run_mainメソッド(L183-L221)
    def _run_main(self, args, parsed_globals):
        # enable basic logging initially. This will be overriden if a python logging config
        # file is provided in the agent config.
        logging.basicConfig(
            level=logging.INFO,
            format=('%(asctime)s - %(name)s - %(levelname)s - '
                    '%(process)d - %(threadName)s - %(message)s'))
        for handler in logging.root.handlers:
            handler.addFilter(logging.Filter('cwlogs'))

        # Parse a dummy string to bypass a bug before using strptime in thread
        # https://bugs.launchpad.net/openobject-server/+bug/947231
        datetime.strptime('2012-01-01', '%Y-%m-%d')
        client_args = {
            'region_name': None,
            'verify': None
        }
        if parsed_globals.region is not None:
            client_args['region_name'] = parsed_globals.region
        if parsed_globals.verify_ssl is not None:
            client_args['verify'] = parsed_globals.verify_ssl
        if parsed_globals.endpoint_url is not None:
            client_args['endpoint_url'] = parsed_globals.endpoint_url
        # Initialize services and append cwlogs version to user agent
        self._session.user_agent_extra += 'cwlogs/' + cwlogs.__version__
        self.logs = self._session.create_client('logs', **client_args)
        # This unregister call will go away once the client switchover
        # is done, but for now we're relying on Logs catching a ClientError
        # when we check if a stream exists, so we need to ensure the
        # botocore ClientError is raised instead of the CLI's error handler.
        self.logs.meta.events.unregister('after-call', unique_id='awscli-error-handler')
        self._validate_arguments(args)
        # Run the command and report success
        if args.config_file:
            self._call_push_file(args, parsed_globals)
        else:
            self._call_push_stdin(args, parsed_globals)

        return 0
```

つまり、下記スクリプトを走らせて、**push.pyファイルを書換えちゃえば使える**ことになります！

```bash:push.pyファイルを書換えるスクリプト
# ディレクトリパスを設定する
PYTHON_PACKAGES="/var/awslogs/lib/python/site-packages"

# 一応、push.pyのファイルが想定した文字列になっているかMD5で判定する
echo "8efb83571499e29142adfa5e129f67da  ${PYTHON_PACKAGES}/cwlogs/push.py" | md5sum -c
if [ $? == 0 ]; then
  # 214行目に実行コードを書込む
  sed -i "214i \        self.logs.meta.events.register_first('before-sign.cloudwatch-logs.PutLogEvents', lambda request, **kwargs: request.headers.add_header('x-amzn-logs-format', 'json/emf'))" ${PYTHON_PACKAGES}/cwlogs/push.py
else
  echo "ERROR: push.py file is invalid!"
fi

# awslogs/awslogsdのリスタートも忘れずに
restart_cloudwatch_logs_agent()
```

```diff_python:上記スクリプト実行後のpush.pyにおけるLogsPushCommandクラスの_run_mainメソッド(L213-L215)
        self.logs.meta.events.unregister('after-call', unique_id='awscli-error-handler')
+       self.logs.meta.events.register_first('before-sign.cloudwatch-logs.PutLogEvents', lambda request, **kwargs: request.headers.add_header('x-amzn-logs-format', 'json/emf'))
        self._validate_arguments(args)
```

動作確認として、先ほどと同様のJSON形式でログ出力してみてください。
CloudWatch Metricsにメトリクスが出現しているはずです。



# おわりに

開発が終了しているライブラリだからこそできる、かなり無理矢理な方法でした。
CloudWatch Logs Agentは、いつ利用できなくなるかもわからないので、気をつけながら使いましょう。
CloudWatch Agentが利用できるにも関わらずCloudWatch Logs Agentを利用している場合は、今すぐ移行することをオススメします。