📝

Lambdaのログ設定を使いこなしてエンジニアとお財布に優しいログに

2023/12/15に公開

AWS

Lambda

cloudwatchlogs

tech

はじめに:

初めまして、DELTAの馬場です。普段はクラウドエンジニアをしています。
この記事は AWS（Amazon Web Services） Advent Calendar 2023 の 15日目です。

概要

以前からAWS Lambdaを利用するサービスを見ていて、ログが分かりづらいなと感じることや
CloudWatchの料金が安くなくなってきたなぁと思うことがありました。

この課題に対して、工夫して改善できるようにしていたのですが少し前にログに関する新機能が発表されました。
これを活用することによって、自前で用意していたものが機能で実現できたので
どう使って、どのような結果を得られるのかをまとめます。

従来のやり方

Lambdaの実行ロールが持っている権限を制限し、任意の命名規則のログストリームにのみ操作が行われるように制限します。

具体例を用いて紹介

例えばヘルスチェックのAPIとして用意しているLambdaがあったと仮定します。

ヘルスチェックはエラーが起きたときのみ確認が必要だが、呼び出し頻度が高くなりがちです。
そのため通常だと不要なログが大量に出てきてしまい、コスト高くついてしまったり、ログを確認する業務が煩雑になってしまう問題があります。

以前はそれを後述の手法で解決していました。

サンプルコード

何かをチェックしたと仮定して、正常であればレスポンスを返すだけのものです。
ログは環境変数からロググループ名とログストリーム名を取得し、エラーがあった場合に環境変数で指定された場所に出力されます。

ロググループについてはあらかじめ作成しておいてください。

import boto3
import os
import time
import json

def lambda_handler(event, context):
    #try:
        ###
        ### 本当はここにヘルスチェックの処理が入っている。
        ###
        #raise Exception('Something went wrong')
    #except Exception as e:
    #    put_error_log(event, e)
    
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

# エラーをログ出力する関数
def put_error_log(event, e):
    logs_client = boto3.client('logs')
  
    # Lambdaの環境変数からロググループとログストリームの名前を取得する
    log_group_name = os.environ['LOG_GROUP_NAME']
    # ログストリームの名前を「YYYY/MM/DD」形式で作成する
    log_stream_name = time.strftime('%Y/%m/%d') + "-" + os.environ['LOG_STREAM_SUFFIX']
  
    log_event = {
        'logGroupName': log_group_name,
        'logStreamName': log_stream_name,
        'logEvents': [
            {
                'timestamp': int(time.time()) * 1000,
                'message': f"Error: {e}, Event: {json.dumps(event)}"
            }
        ]
    }

    # ログストリームが存在しない場合は作成する
    if not stream_exists(logs_client, log_group_name, log_stream_name):
        logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)

    # シーケンストークンがある場合追加する
    sequence_token = get_sequence_token(logs_client, log_group_name, log_stream_name)
    if sequence_token:
        log_event['sequenceToken'] = sequence_token

    # ログを出力する
    response = logs_client.put_log_events(**log_event)

# ログストリームが存在するかどうかを確認する関数
def stream_exists(client, log_group_name, log_stream_name):
    try:
        response = client.describe_log_streams(
            logGroupName=log_group_name,
            logStreamNamePrefix=log_stream_name
        )
        return len(response.get('logStreams', [])) > 0
    except client.exceptions.ResourceNotFoundException:
        return False

# シーケンストークンが存在するかどうかを確認する関数
def get_sequence_token(client, log_group_name, log_stream_name):
    try:
        response = client.describe_log_streams(
            logGroupName=log_group_name,
            logStreamNamePrefix=log_stream_name
        )
        if 'uploadSequenceToken' in response['logStreams'][0]:
            return response['logStreams'][0]['uploadSequenceToken']
    except client.exceptions.ResourceNotFoundException:
        pass
    return None

IAM Policyの変更

実際に手を加える部分はIAMのPolicy側です。
CreateLogStreamとPutLogEventsのActionに対してResourceを任意のlog group及びlogstreamのみ許可するように制限します。
後述の例では、利便性の向上を狙ってログストリームを日付ごとに出せるようにしています。

        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:#{AWS ACCOUNT}:log-group:/aws/lambda/${ロググループ名}:log-stream:*${ログストリーム名}"
            ]
        }

※本記事のソースコードで動かす場合は以下2つの権限が追加で必要になってしまいます。

"logs:DescribeLogGroups",
"logs:DescribeLogStreams"

動作確認

用意したコードのままLambdaのテスト実行を行います。
環境変数は以下を設定します。

実行します。

何も問題が無かったケースの処理になるため、ログ出力されず
ログストリームも作成されません。

コードを少し変更し、エラーのログが出力されるようにして実行してみます。

今度はストリームが作成され、ログが出力されました。

以前の手法では

ここまでの方法を用いることで、必要なログだけを出力しコストと業務効率の改善を実現していました。

新しい機能でできるようになったこと

ようやく本題です。
これまではIAMやソースコードに手を加えて必要なログを出力できるようにしていましたが、ログレベルごとのログ出力やシステムログの出力設定を標準機能で行うことができていませんでした。

2023/11/16 に発表されたこの機能によってこれらが可能になり、コードの変更なしに
必要なログを必要な環境ごとに必要なタイミングで取得できるようになりました。

実際に設定を有効にして動作を見ていきます。

サンプルコード

サンプルコードは以下の通りです。

import json
import logging

logger = logging.getLogger()

def lambda_handler(event, context):
    logger.debug("debug log")
    logger.info("info log")
    logger.warn("warn log")
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

アプリケーションログ

まずはアプリケーションログレベルを設定し、動作を検証していきます。
設定手順は以下の通りです。

設定 > モニタリングおよび運用ツール > ロギング設定 > 「編集」ボタン

アプリケーションログレベルをINFOに設定します。

検証観点

コードに記載したinfo,warnのログのみが出力されること。
debugのログは出力されないこと。

結果

info,warnのログは出力された。
debugのログは出力されなかった。

システムログ

続いてLambdaのシステムログの方についても挙動を見ていきます。
システムログの方は意図してみる場面が無いので、基本は出力されないような設定を探していきます。

再度ロギング設定の画面を開いてみると、 "システムログレベル" という項目があり

システムログは、Lambda が生成するログであり、プラットフォームログとも呼ばれます。より詳細なログレベルを選択すると、CloudWatch ログから追加のコストが発生することがあります。

と記載されています。

まさしく意図していたものっぽいので、これをWARNにして動作を検証していきます。
（ノリでアプリ側もWARNにしてみています。）

検証観点

コードに記載したwarnのログのみ出力される。
Lambdaのシステムログが出力されない。

結果

コードに記載したwarnのログのみ出力された。
Lambdaのシステムログが出力されなかった。

まとめ

Lambdaに追加されたロギング設定を活用することで、意図したログのみを出力させることが可能になりました。
これによりログ監視やログ確認が容易になり、ログ調査の時間が短縮されるだけでなく、運用中にログレベルの変更を設定のみで行うことが可能になります。
コスト面では、ログ出力量を制限することができ、ログ監視ツール等の処理量も減るためコスト削減にもつながることが考えられます。

参考記事：

We're hiring!

最後までお読みいただきありがとうございます。
現在DELTA では一緒に働いてくださる仲間を大募集中です！
ご興味をお持ちいただけましたら、お気軽にフォームからご連絡ください。

はじめに:

概要

従来のやり方

具体例を用いて紹介

サンプルコード

IAM Policyの変更

動作確認

以前の手法では

新しい機能でできるようになったこと

サンプルコード

アプリケーションログ

検証観点

結果

システムログ

検証観点

結果

まとめ

参考記事：

We're hiring!

Discussion