ECS から Slack へ送られるエラーログの集約方法を考える – サーバーワークスエンジニアブログ

import os
import json
import boto3
import gzip
import base64
import traceback

sqs = boto3.client('sqs')
SQS_URL = os.environ['SQS_URL']

def handler(event, context):
    try:
        cw_data = event['awslogs']['data']
        payload = gzip.decompress(base64.b64decode(cw_data))
        logs = json.loads(payload)
        print(f"Decoded CloudWatch Logs data. logGroup: {logs.get('logGroup')}")
    except Exception as e:
        print(f"Failed to decode CloudWatch Logs event: {e}")
        print(traceback.format_exc())
        return {'status': 'decode_error', 'error': str(e)}

    log_group = logs.get('logGroup', '')
    log_stream = logs.get('logStream', '')

    sent_count = 0
    error_count = 0

    for log_event in logs.get('logEvents', []):
        try:
            message = log_event.get('message', '')
            timestamp = log_event.get('timestamp', 0)
            
            try:
                log = json.loads(message)
            except Exception:
                log = {}
            
            log['logGroup'] = log_group
            log['logStream'] = log_stream
            log['timestamp'] = timestamp
            
            log['message'] = log.get('message', message)
            
            if 'service' not in log:
                
                if log_group.startswith('/ecs/'):
                    log['service'] = log_group.split('/ecs/')[-1]
                else:
                    log['service'] = 'unknown'

            
            sqs.send_message(
                QueueUrl=SQS_URL,
                MessageBody=json.dumps(log)
            )
            sent_count += 1
        except Exception as e:
            print(f"Failed to send log to SQS: {e}")
            print(traceback.format_exc())
            error_count += 1

    print(f"=== Lambda handler completed: sent={sent_count}, error={error_count} ===")
    return {
        'status': 'completed',
        'sent_count': sent_count,
        'error_count': error_count
    }

import os
import json
import boto3
from datetime import datetime
from collections import defaultdict

sns = boto3.client('sns')
SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN')

def handler(event, context):
    print("=== Simple SQS Batch Lambda invoked ===")
    records = event.get('Records', [])
    service_messages = defaultdict(list)

    for idx, record in enumerate(records):
        try:
            body = json.loads(record['body'])
            service = body.get('service', 'unknown')
            log_group = body.get('logGroup', 'unknown')
            log_stream = body.get('logStream', 'unknown')
            msg = body.get('message', str(body))
            ts = body.get('timestamp')
            if ts:
                if ts > 1e12:
                    ts = int(ts / 1000)
                dt = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')
            else:
                dt = "unknown"
            formatted = (
                f"サービス: {service}\n"
                f"LogGroup: {log_group}\n"
                f"コンテナ: {log_stream}\n"
                f"発生時刻: {dt}\n"
                f"メッセージ: {msg}"
            )
            service_messages[service].append(formatted)
        except Exception as e:
            print(f"Failed to parse record[{idx}] as JSON: {e}")

    for service, messages in service_messages.items():
        aggregated_message = "\n\n".join(messages)
        notification = {
            "version": "1.0",
            "source": "custom",
            "content": {
                "title": f":package: SQSバッチ集約ログ（{service}, {len(messages)}件）",
                "description": (
                    f"{aggregated_message}\n\n"
                    f"---\n"
                    f"集約件数: {len(messages)}件\n"
                    f"この通知はSQSバッチ集約（最大100件/最大60秒）で送信されています。"
                )
            }
        }
        try:
            resp = sns.publish(
                TopicArn=SNS_TOPIC_ARN,
                Message=json.dumps(notification)
            )
            print(f"SNS publish response for {service}: {resp}")
        except Exception as e:
            print(f"Failed to publish to SNS for {service}: {e}")

    return {"status": "sent", "service_count": len(service_messages)}

import os
import json
import boto3
import time
import gzip
import base64
import traceback
from datetime import datetime
from boto3.dynamodb.conditions import Key
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])
sqs = boto3.client('sqs')
sns = boto3.client('sns')
SQS_URL = os.environ['SQS_URL']
SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN')
AGGREGATION_KEY = "all"
AGGREGATION_PERIOD_SECONDS = 60  

def put_item_with_retry(table, item, max_retries=3, condition_expression=None):
    """DynamoDBへのリトライ機能付きput_item（条件付き書き込み対応）"""
    for attempt in range(max_retries):
        try:
            if condition_expression:
                table.put_item(Item=item, ConditionExpression=condition_expression)
            else:
                table.put_item(Item=item)
            print(f"Successfully put item on attempt {attempt + 1}: {item.get('aggregation_key', 'unknown')}")
            return True
        except ClientError as e:
            error_code = e.response['Error']['Code']
            print(f"DynamoDB put_item attempt {attempt + 1} failed with {error_code}: {e}")

            if error_code in ['ConditionalCheckFailedException']:
                
                raise e
            elif attempt == max_retries - 1:
                print(f"Failed to put item after {max_retries} attempts: {e}")
                return False
            else:
                
                wait_time = (2 ** attempt) + (time.time() % 1)  
                print(f"Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
        except Exception as e:
            print(f"Unexpected error on attempt {attempt + 1}: {e}")
            if attempt == max_retries - 1:
                print(f"Failed to put item after {max_retries} attempts: {e}")
                return False
            time.sleep(2 ** attempt)
    return False

def send_sns_with_retry(topic_arn, message, max_retries=3):
    """SNS送信のリトライ機能付き関数"""
    for attempt in range(max_retries):
        try:
            response = sns.publish(TopicArn=topic_arn, Message=json.dumps(message))
            print(f"SNS message sent successfully on attempt {attempt + 1}: {response['MessageId']}")
            return True
        except Exception as e:
            print(f"SNS send attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                print(f"Failed to send SNS after {max_retries} attempts: {e}")
                return False
            time.sleep(2 ** attempt)
    return False

def send_sqs_with_retry(queue_url, message_body, delay_seconds=0, max_retries=3):
    """SQS送信のリトライ機能付き関数"""
    for attempt in range(max_retries):
        try:
            response = sqs.send_message(
                QueueUrl=queue_url,
                MessageBody=message_body,
                DelaySeconds=delay_seconds
            )
            print(f"SQS message sent successfully on attempt {attempt + 1}: {response['MessageId']}")
            return True
        except Exception as e:
            print(f"SQS send attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                print(f"Failed to send SQS after {max_retries} attempts: {e}")
                return False
            time.sleep(2 ** attempt)
    return False

def handler(event, context):
    """
    CloudWatch Logsサブスクリプションフィルターからのイベントを処理
    エラーハンドリングとリトライ機能を強化
    """
    print(f"=== Lambda handler started at {datetime.now().isoformat()} ===")
    print(f"Received event: {json.dumps(event)[:500]}...")  

    try:
        
        cw_data = event['awslogs']['data']
        payload = gzip.decompress(base64.b64decode(cw_data))
        logs = json.loads(payload)
        print(f"Successfully decoded CloudWatch Logs data. Log group: {logs.get('logGroup')}")
    except Exception as e:
        error_msg = f"Failed to decode CloudWatch Logs event: {e}"
        print(error_msg)
        print(traceback.format_exc())
        return {'status': 'decode_error', 'error': str(e)}

    log_group = logs.get('logGroup', 'unknown')
    log_stream = logs.get('logStream', 'unknown')
    now = int(time.time())
    processed_count = 0
    error_count = 0

    print(f"Processing {len(logs.get('logEvents', []))} log events from {log_group}/{log_stream}")

    for log_event in logs.get('logEvents', []):
        try:
            message = log_event.get('message', '')
            timestamp = log_event.get('timestamp', now * 1000) // 1000  

            try:
                log = json.loads(message)
                print(f"Successfully parsed JSON log: {log.get('level', 'unknown')} level")
            except Exception:
                print(f"Skip non-JSON message: {message[:100]}...")
                continue  

            
            aggregation_key = log_group

            
            period_start = now - (now % AGGREGATION_PERIOD_SECONDS)
            
            notified_flag = f"notified#{aggregation_key}#{period_start}"

            try:
                
                success = put_item_with_retry(
                    table,
                    {
                        'aggregation_key': notified_flag,
                        'timestamp': 0,
                        'notified': True
                    },
                    condition_expression='attribute_not_exists(aggregation_key)'
                )

                if not success:
                    print(f"Failed to put notified flag for {aggregation_key}")
                    error_count += 1
                    continue

                
                log_item = {
                    'aggregation_key': aggregation_key,
                    'timestamp': timestamp,
                    'message': log.get('message', message),
                    'logGroup': log_group,
                    'logStream': log_stream,
                    'level': log.get('level', 'ERROR'),
                    'created_at': datetime.now().isoformat()
                }

                success = put_item_with_retry(table, log_item)
                if not success:
                    print(f"Failed to put log item for {aggregation_key}")
                    error_count += 1
                    continue

                print(f"Successfully stored 1st error: aggregation_key={aggregation_key}, timestamp={timestamp}")

                
                if SNS_TOPIC_ARN:
                    custom_notification = {
                        "version": "1.0",
                        "source": "custom",
                        "content": {
                            "title": f":rotating_light: Dynamo方式: 1件目エラー発生",
                            "description": f"""
📍 **サービス情報**
- LogGroup: {log_group}
- LogStream: {log_stream}
- 発生時刻: {datetime.fromtimestamp(timestamp).isoformat()}

🔍 **エラー詳細**
{log.get('message', message)}

🏷️ **追加情報**
- レベル: {log.get('level', 'ERROR')}
- 期間: {period_start} - {period_start + AGGREGATION_PERIOD_SECONDS}秒
""".strip()
                        }
                    }

                    if send_sns_with_retry(SNS_TOPIC_ARN, custom_notification):
                        print(f"Successfully sent immediate SNS notification for {aggregation_key}")
                    else:
                        print(f"Failed to send immediate SNS notification for {aggregation_key}")
                        error_count += 1

                
                sqs_message = {
                    'aggregation_key': aggregation_key,
                    'period_start': period_start,
                    'period_seconds': AGGREGATION_PERIOD_SECONDS,
                    'trigger_time': now
                }

                if send_sqs_with_retry(SQS_URL, json.dumps(sqs_message), AGGREGATION_PERIOD_SECONDS):
                    print(f"Successfully sent SQS trigger message for {aggregation_key}")
                else:
                    print(f"Failed to send SQS trigger message for {aggregation_key}")
                    error_count += 1

            except ClientError as e:
                if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                    
                    log_item = {
                        'aggregation_key': aggregation_key,
                        'timestamp': timestamp,
                        'message': log.get('message', message),
                        'logGroup': log_group,
                        'logStream': log_stream,
                        'level': log.get('level', 'ERROR'),
                        'created_at': datetime.now().isoformat()
                    }

                    if put_item_with_retry(table, log_item):
                        print(f"Successfully stored subsequent error: aggregation_key={aggregation_key}, timestamp={timestamp}")
                    else:
                        print(f"Failed to store subsequent error for {aggregation_key}")
                        error_count += 1
                else:
                    print(f"Unexpected DynamoDB error for {aggregation_key}: {e}")
                    error_count += 1
            except Exception as e:
                print(f"Unexpected error processing log event for {aggregation_key}: {e}")
                print(traceback.format_exc())
                error_count += 1

            processed_count += 1

        except Exception as e:
            print(f"Error processing individual log event: {e}")
            print(traceback.format_exc())
            error_count += 1

    result = {
        'status': 'completed',
        'processed_count': processed_count,
        'error_count': error_count,
        'log_group': log_group,
        'timestamp': datetime.now().isoformat()
    }

    print(f"=== Lambda handler completed: {result} ===")
    return result

import os
import json
import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])
sns = boto3.client('sns')
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']


def handler(event, context):
    print("=== Dynamo Aggregator Lambda invoked ===")
    print(f"event: {json.dumps(event)[:1000]}")  

    
    for idx, record in enumerate(event.get('Records', [])):
        print(f"Processing record[{idx}]: {record['body']}")
        body = json.loads(record['body'])
        period_start = body.get('period_start')
        period_seconds = body.get('period_seconds', 60)
        period_end = period_start + period_seconds

        
        aggregation_key = body.get('aggregation_key', 'unknown')

        
        log_group = None
        log_stream = None

        
        
        
        log_group = body.get('logGroup', '不明')
        log_stream = body.get('logStream', '不明')

        print(f"Querying DynamoDB: aggregation_key={aggregation_key}, period_start={period_start}, period_end={period_end}")
        
        resp = table.query(
            KeyConditionExpression=Key('aggregation_key').eq(aggregation_key) & Key('timestamp').between(period_start, period_end)
        )
        logs = resp.get('Items', [])
        print(f"Found {len(logs)} logs in DynamoDB for this period.")
        if logs:
            
            
            messages_with_meta = [
                f"[logGroup]: {log.get('logGroup', '不明')}\n[logStream]: {log.get('logStream', '不明')}\n---\n{log.get('message', '')}"
                for log in logs
            ]
            message_with_meta = "\n\n".join(messages_with_meta)
            custom_notification = {
                "version": "1.0",
                "source": "custom",
                "content": {
                    "title": f":cyclone:Dynamo方式集約ログ（{len(logs)}件）",
                    "description": (
                        f"{message_with_meta}\n\n"
                        f":ラベルタグ: **追加情報**\n"
                        f"- 集約件数: {len(logs)}件\n"
                        f"- 期間: {period_start} - {period_end}秒 "
                        f"(この期間内のエラーを集約しています)"
                    )
                }
            }
            print(f"Publishing to SNS: {SNS_TOPIC_ARN}")
            print(f"Aggregated message:\n{message_with_meta}")
            resp = sns.publish(
                TopicArn=SNS_TOPIC_ARN,
                Message=json.dumps(custom_notification)
            )
            print(f"SNS publish response: {resp}")
        else:
            print("No logs to aggregate for this period.")
    print("=== Dynamo Aggregator Lambda finished ===")
    return {'status': 'aggregated'}