概要

Pythonのwatchtowerとboto3を使って、AWS Cloudwatchlogsにデバッグ内容を出力するコードを実装しましたので紹介します。フレームワークはFastapiを利用しましたが、Pythonであればどれも似た形だと思います。

コード解説

config.py

from pydantic import BaseSettings
from time import time
import os
from datetime import datetime
import pytz
import boto3
from botocore.session import Session, get_session
from botocore.credentials import RefreshableCredentials
import watchtower
import logging
import logging.config


class Settings(BaseSettings):

    # 環境変数をここに #
    region_name: str
    arn_role_name: str
    session_ttl: int = 900

    def get_session_credentials(self):
        def __get_session_credentials():
            sts_client = boto3.client("sts", region_name=self.region_name)
            params = {
                "RoleArn": self.arn_role_name,
                "RoleSessionName": "AssumeRoleSession1",
                "DurationSeconds": self.session_ttl,
            }
            response = sts_client.assume_role(**params).get("Credentials")
            credentials = {
                "access_key": response.get("AccessKeyId"),
                "secret_key": response.get("SecretAccessKey"),
                "token": response.get("SessionToken"),
                "expiry_time": datetime.fromtimestamp(time() + self.session_ttl).replace(tzinfo=pytz.utc).isoformat(),
            }
            print(f"credentials: {credentials}")
            return credentials
            
        session_credentials = RefreshableCredentials.create_from_metadata(
            metadata=__get_session_credentials(),
            refresh_using=__get_session_credentials,
            method="sts-assume-role",
        )
    
        return session_credentials
    
    def refreshable_session(self, session_credentials):
        session = get_session()
        session._credentials = session_credentials
        session.set_config_variable('region', self.region_name)
        autorefresh_session = boto3.Session(botocore_session=session)
        return autorefresh_session

    @property
    def get_logger(self):
        session_credentials = self.get_session_credentials()
        autorefresh_session = self.refreshable_session(session_credentials)
        
        LOGGING = {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "aws": {
                     "()": LocalTimeFormatter,
                    "format": "%(asctime)s [%(levelname)-8s] %(message)s [%(pathname)s:%(lineno)d]",
                    "datefmt": "%Y-%m-%d %H:%M:%S",
                },
            },
            "handlers": {
                "watchtower": {
                    "level": "DEBUG",
                    "class": "watchtower.CloudWatchLogHandler",
                    "boto3_client": autorefresh_session.client('logs'),
                    "log_group": "log-sample-app",
                    "stream_name": "sample-app",
                    "formatter": "aws",
                },
                "console": {"class": "logging.StreamHandler", "formatter": "aws",},
            },
            "loggers": {
                "watchtower": {"level": "DEBUG", "handlers": ["watchtower"], "propogate": False,}
            },
        }
        
        logging.config.dictConfig(LOGGING)
        logger = logging.getLogger("watchtower")
        return logger

解説

簡単にいうとそれぞれ以下の通り。

・get_session_credentials：AWSのSecurity Token Service (STS) を使用して一時的な認証情報を取得（AWSリソースへのアクセスを一時的に許可するためのもの）
・refreshable_session：取得した一時的な認証情報を使用して、新しいboto3セッションを作成
・get_logger：CloudWatchにログを送信するためのロガーを設定

２つ注意点があるなと思ったので、コード解説と共に紹介します。

注意点1 （他のモジュールで使うとき）

上記のように実装した場合、他のモジュールでloggerを使ってログ出力を行うことになると思いますが、その時にひとつ注意点があります。
上記の場合、loggerをグローバルスコープで定義するとエラーが発生します。

例えば、上記config.pyでグローバル変数として以下のようにするとします。

config.py

settings = Settings().
logger = settings.get_logger

そして他のモジュールで以下のようにインポート。

from config import logger

def xxxxxxx
    logger.info("test")

本来、RefreshableCredentials.create_from_metadata関数により、セッション認証情報が期限切れになると自動的にリフレッシュの関数が呼び出され、新しい認証情報が生成されます。

しかしこの場合、loggerをグローバル変数で定義してしまうとクレデンシャルのオートリフレッシュがされません。
せっかく上記関数でクレデンシャルを自動更新しているのに、from config import loggerとするとloggerは一度だけ初期化され、その後は同じインスタンスが再利用されるためです。

したがって、get_loggerを毎回呼び出して新たなクレデンシャルを取得する必要がある場合は、Settingsクラスのインスタンスを作成＆そのget_loggerメソッドを直接呼び出す必要があります。

以下のようにコードを変更すればOKです(config.pyでは上記の関数のみ。グローバル変数は利用しない)。

from config import logger

settings = Settings()

def xxxxxxx
    logger = settings.get_logger
    logger.info("test")

関数の内部で定義された変数は、関数が呼び出されるたびに新たに初期化されるので、毎回クレデンシャルを更新してくれるというわけです。

`DurationSeconds`とは

ちなみに、DurationSecondsパラメータは、一時的なセキュリティクレデンシャルが有効である期間を秒単位で指定します。
グローバル変数で設定していると初期化が最初の1回以降行われないので、15分後に以下のエラーになります。
（ExpiredTokenExceptionは、セキュリティトークンが期限切れになると発生するエラー）

WatchtowerWarning: Failed to deliver logs: An error occurred (ExpiredTokenException) when calling the PutLogEvents operation: The security token included in the request is expired

セキュリティクレデンシャルが必要なタスクを完了するのに適切な時間をDurationSecondsに設定することが推奨されています。上記では900（15分）としています。

注意点2 （AWS UTC時間で過去日付の有効期限になる）

AWSではUTC時間によりクレデンシャルの有効期限が過去日付になってしまうエラーがありました...。
永遠に過去のクレデンシャル有効期限が発行されては意味がありません。

credentialsの中身、以下のようにしているとエラーになりました。

"expiry_time": response.get("Expiration").isoformat(),

AWS ECSタスクログに出ているエラー内容は以下。

WatchtowerWarning: Failed to deliver logs: An error occurred (ExpiredTokenException) when calling the PutLogEvents operation: The security token included in the request is expired

セキュリティクレデンシャルの有効期限が切れてしまっていると出ました。
これについては、以下のサイトを参考にしたら解決しました。

expiry_timeを以下のようにします↓

"expiry_time": datetime.fromtimestamp(time() + self.session_ttl).replace(tzinfo=pytz.utc).isoformat(),

ここはアプリ側でUTCにするんだ、DBでUTCにするんだ、とか色々と話が広がっていき複雑になりがちなので、私は上記で対応としました。