AWS Lambda CloudWatch Jobs

import json
import logging
import boto3
from datetime import datetime

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    """スケジュールジョブのハンドラー"""
    logger.info(f"Job triggered at: {datetime.utcnow()}")
    logger.info(f"Event: {json.dumps(event)}")
    
    try:
        # ジョブ処理ロジック
        result = process_scheduled_task(event)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Job completed successfully',
                'result': result
            })
        }
    except Exception as e:
        logger.error(f"Job failed: {str(e)}")
        raise

def process_scheduled_task(event):
    """メインのジョブ処理ロジック"""
    # 例: データベースクリーンアップ、レポート送信など
    logger.info("Processing scheduled task...")
    
    # ビジネスロジックをここに記述
    return {"status": "success", "processed_items": 10}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:*",  # DynamoDBアクセスが必要な場合
        "s3:*"         # S3アクセスが必要な場合
      ],
      "Resource": "*"
    }
  ]
}

resource "aws_scheduler_schedule" "daily_job" {
  name       = "daily-job"
  group_name = "default"

  flexible_time_window {
    mode                      = "FLEXIBLE"
    maximum_window_in_minutes = 15
  }

  schedule_expression          = "cron(0 9 * * ? *)"
  schedule_expression_timezone = "UTC"

  target {
    arn      = aws_lambda_function.my_function.arn
    role_arn = aws_iam_role.scheduler_role.arn

    retry_policy {
      maximum_retry_attempts       = 3
      maximum_event_age_in_seconds = 3600
    }

    dead_letter_config {
      arn = aws_sqs_queue.dlq.arn
    }
  }
}

aws events put-rule \
  --name daily-lambda-rule \
  --schedule-expression 'cron(0 9 * * ? *)'

aws events put-targets \
  --rule daily-lambda-rule \
  --targets '{
    "Id": "1",
    "Arn": "arn:aws:lambda:us-east-1:123456789012:function:my-function"
  }'

aws lambda add-permission \
  --function-name my-function \
  --statement-id daily-lambda-rule \
  --action 'lambda:InvokeFunction' \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:us-east-1:123456789012:rule/daily-lambda-rule

# 毎日9:00 AM UTC
cron(0 9 * * ? *)

# 毎週月曜日9:00 AM
cron(0 9 ? * MON *)

# 毎月1日00:00
cron(0 0 1 * ? *)

# 15分ごと
rate(15 minutes)

# 1時間ごと
rate(1 hour)

# 月曜日から金曜日9:00 AM
cron(0 9 ? * MON-FRI *)

# 月の最終日
cron(0 0 L * ? *)

# 毎月の最初の月曜日
cron(0 0 ? * MON#1 *)

import boto3
from datetime import datetime, timedelta

dynamodb = boto3.resource('dynamodb')
lock_table = dynamodb.Table('job-locks')

def acquire_lock(job_id):
    """重複を防ぐためにロックを取得"""
    try:
        lock_table.put_item(
            Item={
                'job_id': job_id,
                'locked_at': datetime.utcnow().isoformat(),
                'expires_at': (datetime.utcnow() + timedelta(minutes=10)).isoformat()
            },
            ConditionExpression='attribute_not_exists(job_id)'
        )
        return True
    except Exception:
        return False  # ロックが既に存在

def release_lock(job_id):
    """完了後にロックを解放"""
    lock_table.delete_item(Key={'job_id': job_id})

import time
import logging

logger = logging.getLogger()

def lambda_handler(event, context):
    start_time = time.time()
    
    try:
        # カスタムメトリクス
        logger.info("JOB_STARTED")
        
        result = process_job(event)
        
        # 実行時間をログ
        duration = time.time() - start_time
        logger.info(f"JOB_COMPLETED duration={duration}")
        
        return result
    except Exception as e:
        logger.error(f"JOB_FAILED error={str(e)}")
        raise

def lambda_handler(event, context):
    try:
        result = process_job(event)
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
    except ValueError as e:
        logger.error(f"Validation error: {str(e)}")
        return {
            'statusCode': 400,
            'body': json.dumps({'error': str(e)})
        }
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        # DLQに送信またはアラート
        send_alert(str(e))
        raise

import boto3
import logging
from datetime import datetime, timedelta

logger = logging.getLogger()
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):
    """30日以上前のレコードをクリーンアップ"""
    table = dynamodb.Table('user-activity')
    
    cutoff_date = (datetime.utcnow() - timedelta(days=30)).isoformat()
    
    try:
        # 古いレコードをスキャンして削除
        response = table.scan(
            FilterExpression='created_at < :cutoff',
            ExpressionAttributeValues={':cutoff': cutoff_date}
        )
        
        deleted_count = 0
        with table.batch_writer() as batch:
            for item in response['Items']:
                batch.delete_item(Key={'id': item['id']})
                deleted_count += 1
        
        logger.info(f"Deleted {deleted_count} old records")
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'deleted_count': deleted_count,
                'cutoff_date': cutoff_date
            })
        }
    except Exception as e:
        logger.error(f"Cleanup failed: {str(e)}")
        raise