はじめに

AWS Lambdaで発生したエラーのログをAmazon Bedrockに一次調査してもらい、Slackに通知するところまで試してみました。

構成図は以下です。

実際の通知1

処理

  1. Lambda(sample-app)でエラー発生
  2. CloudWatch Alarmで発火
  3. SNSトピックからSlackへのエラー通知 & エラーログ調査用Lambda(error-invasitigator)をトリガー
  4. エラーログ調査用LambdaでBedrockを呼び出し、Lambda(sample-app)のCloudWatch Logsを調査した結果をSlackへ通知

プロンプトは分析の上、根本原因、修正方法、優先度を結果として返してもらうようにしています。

ソースコード

※ 雑ですがご容赦ください。

sample-app

import json

def lambda_handler(event, context):
    """
    テスト用エラー発生関数
    """
    print(f"Received event: {json.dumps(event)}")

    # エラーを発生させる
    raise Exception("Test error for CloudWatch alarm monitoring system")

    # 以下は実行されない
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'OK'})
    }

error-invasitigator

import json
import boto3
import requests
import os
from datetime import datetime, timedelta, timezone

logs_client = boto3.client('logs')
bedrock_client = boto3.client('bedrock-runtime')

def lambda_handler(event, context):
    try:
        for record in event['Records']:
            sns_message = json.loads(record['Sns']['Message'])

            # CloudWatch Logsからエラーログを取得
            error_logs = get_error_logs(sns_message)

            # Bedrockで調査
            investigation = investigate_error(error_logs, sns_message)

            # Slackに通知
            send_to_slack(investigation)

        return {'statusCode': 200}
    except Exception as e:
        print(f"Error: {e}")
        return {'statusCode': 500}

def get_error_logs(alarm_data):
    try:
        # 関数名を取得
        dimensions = alarm_data.get('Trigger', {}).get('Dimensions', [])
        function_name = None
        for d in dimensions:
            if d.get('name') == 'FunctionName':
                function_name = d['value']
                break

        if not function_name:
            return []

        # 過去5分間のエラーログを取得（timezone-aware）
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(minutes=5)

        response = logs_client.filter_log_events(
            logGroupName=f'/aws/lambda/{function_name}',
            startTime=int(start_time.timestamp() * 1000),
            endTime=int(end_time.timestamp() * 1000),
            filterPattern='ERROR'
        )

        return [event['message'] for event in response.get('events', [])][:5]
    except:
        return []

def investigate_error(error_logs, alarm_data):
    log_messages = '\n'.join(error_logs) if error_logs else 'No error logs found'

    prompt = f"""
以下のLambda関数のエラーログを分析してください:

関数名: {alarm_data.get('Trigger', {}).get('Dimensions', [{}])[0].get('value', 'Unknown')}
エラーログ:
{log_messages}

以下の形式で分析結果を返してください:
根本原因: 
修正方法: 
優先度: (HIGH/MEDIUM/LOW)
"""

    try:
        response = bedrock_client.invoke_model(
            modelId='apac.anthropic.claude-3-5-sonnet-20241022-v2:0',
            body=json.dumps({
                'anthropic_version': 'bedrock-2023-05-31',
                'max_tokens': 500,
                'messages': [{'role': 'user', 'content': prompt}]
            })
        )

        result = json.loads(response['body'].read())
        return result['content'][0]['text']
    except Exception as e:
        return f"Investigation failed: {e}"

def send_to_slack(investigation):
    try:
        requests.post(os.environ['SLACK_WEBHOOK_URL'], json={
            "text": f"🔍 AI Error Investigation Complete\n```{investigation}```"
        })
    except:
        pass

他のパターン

Lambda(sample-app)の内容を変えてエラーパターンも試してみます。

sample-app

import json

def lambda_handler(event, context):
    """
    2つのキーワードを受け取って結合して返す
    """
    print(f"Received event: {json.dumps(event)}")

    keyword1 = event['keyword1']
    keyword2 = event['keyword2']
    result = keyword1 + keyword2

    print(f"Combined result: {result}")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'keyword1': keyword1,
            'keyword2': keyword2,
            'result': result
        })
    }

エラーを発生させてみます。

実際の通知2

感想

自分自身エラーの内容を生成AIに確認してもらうこともあり、その部分を自動化してみたかったので期待した形になってよかったです。
今回のものはあくまでもお試しで作ったものなのでかなり簡易的ではありますが、実際のアプリケーションでエラーが発生した際は原因調査とは別に影響範囲の確認も必要になるので、原因調査の方をひとまずAIに任せてざっくり把握できるのはありがたいかもしれません。

なお、エラー調査用Lambdaが大量にBedrockを呼び出してしまうとコストがかかる可能性があるので、その点は意識しておいた方が良さそうだと思いました。