はじめに
こんばんは、mirukyです。
CloudWatch LogsをS3へエクスポートすると、開始要求を送った後に完了を確認する必要があります。今回は、その待ち時間をAWS Lambda Durable Functionsで扱ってみます。
AWSマネジメントコンソールから1GBのログを使い、待機から結果確認まで進めます。では、始めましょう。
目次
1. Durable Functionsの待機と再開
Durable Functionsは、処理の進行状況をチェックポイントへ保存し、複数回のLambda呼び出しをまたいで実行を続ける機能です。context.wait()で計算を停止し、指定時間が過ぎると新しい呼び出しで再開します。
再開時にはコードが最初から再生されます。完了したstepは保存済みの結果を返すため、AWS API呼び出しや現在時刻、乱数を使う処理は@durable_stepの中で実行します。
今回は1回の呼び出しのタイムアウトを1分に設定し、エクスポート開始後に16分待ちます（15分超えた想定のためですね）。わざわざこうするのは、CloudwatchのS3エクスポートはGB数によっては15分を超える場合があるためです。
2. 今回の構成と検証条件
Lambdaはエクスポートの開始と状態確認を担当し、ログ本体をS3へ書き込むのはCloudWatch Logsです。Lambdaの実行ロールに加えて、CloudWatch Logsからの書き込みを許可するS3バケットポリシーが必要になります。
すべて東京リージョンで作成します。S3バケットとロググループを同じリージョンに置き、次の条件で進めます。
| 項目 | 設定 |
|---|---|
| ランタイム・アーキテクチャ | Python 3.14・x86_64 |
| メモリ | 1024MB |
| 1回の呼び出しのタイムアウト | 1分 |
| 永続実行のタイムアウト・保持期間 | 13時間・14日 |
| 呼び出し | 発行済みバージョンを非同期で実行 |
疑似ログはランダムバイトをBase64へ変換して生成します。1GBは10進数の1,000,000,000バイトとし、PutLogEventsのイベントごとの26バイトも含めて計算します。
東京のCloudWatch Logsのスタンダードクラスの取り込み単価は1GBあたり0.76USDです。今回の1GBでは取り込み費用が大部分を占め、Lambda、Durable Functions、S3の短時間利用を含む概算は無料枠適用前で1USD未満です。待機中のLambdaコンピューティング料金は発生しませんが、Durable Functionsのオペレーションや保存データには料金がかかります。
3. コンソールで準備する
3-1. ロググループを作成する
エクスポート元になるロググループを用意します。
ロググループ名を入力し、保持期間を1 日、ログクラスをスタンダードにして作成します。ログの投入は4章の関数が行います。
3-2. S3バケットとバケットポリシーを設定する
書き出し先のバケットを作成します。
汎用のバケットに固有名を入力します。パブリックアクセスは全項目をブロックし、ACL無効、暗号化はSSE-S3の設定で作成します。
作成後はアクセス許可へ進み、バケットに次のポリシーを設定します。<ACCOUNT_ID>、バケット名、ロググループ名を自分の値へ置き換えてください。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowCloudWatchLogsGetBucketAcl",
"Effect": "Allow",
"Principal": {"Service": "logs.ap-northeast-1.amazonaws.com"},
"Action": "s3:GetBucketAcl",
"Resource": "arn:aws:s3:::<YOUR_EXPORT_BUCKET>",
"Condition": {
"StringEquals": {"aws:SourceAccount": "<ACCOUNT_ID>"},
"ArnLike": {
"aws:SourceArn": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:<LOG_GROUP_NAME>"
}
}
},
{
"Sid": "AllowCloudWatchLogsPutObject",
"Effect": "Allow",
"Principal": {"Service": "logs.ap-northeast-1.amazonaws.com"},
"Action": "s3:PutObject",
"Resource": "arn:aws:s3:::<YOUR_EXPORT_BUCKET>/durable-export/*",
"Condition": {
"StringEquals": {
"s3:x-amz-acl": "bucket-owner-full-control",
"aws:SourceAccount": "<ACCOUNT_ID>"
},
"ArnLike": {
"aws:SourceArn": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:<LOG_GROUP_NAME>"
}
}
}
]
}
GetBucketAclは所有者確認、PutObjectはログの書き込みに使います。SourceAccountとSourceArnで、対象アカウントのロググループだけを許可します。
3-3. Lambdaの実行ロールを用意する
Lambdaを信頼する実行ロールを作成し、AWSLambdaBasicExecutionRoleと次のインラインポリシーを追加します。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "DurableExecutionCheckpoint",
"Effect": "Allow",
"Action": [
"lambda:CheckpointDurableExecution",
"lambda:GetDurableExecutionState"
],
"Resource": "arn:aws:lambda:ap-northeast-1:<ACCOUNT_ID>:function:<FUNCTION_NAME>:*"
},
{
"Sid": "CreateExportTask",
"Effect": "Allow",
"Action": "logs:CreateExportTask",
"Resource": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:<LOG_GROUP_NAME>"
},
{
"Sid": "WriteBenchmarkLogs",
"Effect": "Allow",
"Action": ["logs:CreateLogStream", "logs:PutLogEvents"],
"Resource": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:<LOG_GROUP_NAME>:log-stream:*"
},
{
"Sid": "ReadExportStatusAndMetrics",
"Effect": "Allow",
"Action": ["logs:DescribeExportTasks", "cloudwatch:GetMetricStatistics"],
"Resource": "*"
},
{
"Sid": "ListExportObjects",
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::<YOUR_EXPORT_BUCKET>",
"Condition": {"StringLike": {"s3:prefix": "durable-export/*"}}
},
{
"Sid": "WriteBenchmarkResult",
"Effect": "Allow",
"Action": "s3:PutObject",
"Resource": "arn:aws:s3:::<YOUR_EXPORT_BUCKET>/durable-export/*/benchmark-result.json"
}
]
}
s3:ListBucketは出力の集計、s3:PutObjectは結果JSONの保存に使います。エクスポート本体の書き込み権限は、3-2で設定したバケットポリシーが担当します。
内容を確認してポリシーを作成します。関数の作成時には、この実行ロールを指定します。
4. 関数を作成して実行する
4-1. Durable Functionsを有効にする
Durable Functionsは関数の作成時に有効にする必要があります。
関数の作成から新規関数を用意します。関数名はポリシーの<FUNCTION_NAME>と一致させ、アーキテクチャをx86_64にし、3章の実行ロールを指定します。
一から作成、Python 3.14を選び、永続実行を設定を有効にします。実行タイムアウトを13 時、保持期間を14 日にして作成します。
メモリを1024MB、1回の呼び出しのタイムアウトを1分へ変更します。
4-2. コードを配置する
処理の中心は、エクスポート開始、16分待機、同じtaskIdでの状態確認です。以下は完全版からの抜粋です。
# エクスポートの開始結果を保存します
export = context.step(
start_export_task(
log_group_name,
destination_bucket,
destination_prefix,
from_ms,
to_ms,
export_task_name,
),
config=NO_RETRY_CONFIG,
)
# 計算を停止して16分待ち、保存したtaskIdで状態確認へ戻ります
context.wait(
duration=Duration.from_minutes(16),
name="wait-beyond-fifteen-minutes",
)
result = context.step(check_export_status(export["taskId"]))
完全版では、先に1GBの疑似ログを投入し、100MBごとにチェックポイントを作成します。エクスポートの完了後にS3のオブジェクト数と容量を集計し、結果をJSONへ保存します。
lambda_function.pyの完全版
import base64
import json
import math
import os
import time
from datetime import datetime, timedelta, timezone
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
from aws_durable_execution_sdk_python import (
DurableContext,
durable_execution,
durable_step,
)
from aws_durable_execution_sdk_python.config import (
Duration,
StepConfig,
StepSemantics,
)
from aws_durable_execution_sdk_python.retries import RetryDecision
AWS_CONFIG = Config(retries={"max_attempts": 10, "mode": "adaptive"})
logs_client = boto3.client("logs", config=AWS_CONFIG)
cloudwatch_client = boto3.client("cloudwatch", config=AWS_CONFIG)
s3_client = boto3.client("s3", config=AWS_CONFIG)
PUT_LOG_EVENTS_LIMIT_BYTES = 1_048_576
LOG_EVENT_OVERHEAD_BYTES = 26
MAX_MESSAGE_BYTES = PUT_LOG_EVENTS_LIMIT_BYTES - LOG_EVENT_OVERHEAD_BYTES
GENERATION_PART_BYTES = 100_000_000
@durable_step
def create_benchmark_stream(step_context, log_group_name, stream_name):
# ベンチマーク用のログストリームを一度だけ作成します
logs_client.create_log_stream(
logGroupName=log_group_name,
logStreamName=stream_name,
)
return {"streamName": stream_name}
@durable_step
def write_benchmark_part(
step_context,
log_group_name,
stream_name,
part_number,
target_api_bytes,
):
# 外部APIへの書き込みは、再生されないstep内に閉じ込めます
remaining = target_api_bytes
event_count = 0
message_bytes_total = 0
first_timestamp = None
last_timestamp = None
while remaining > 0:
if remaining <= LOG_EVENT_OVERHEAD_BYTES:
raise ValueError("A generation part ended with fewer than 27 bytes")
message_bytes = min(
MAX_MESSAGE_BYTES,
remaining - LOG_EVENT_OVERHEAD_BYTES,
)
raw_bytes = math.ceil(message_bytes * 3 / 4)
message = base64.b64encode(os.urandom(raw_bytes)).decode("ascii")[:message_bytes]
event_timestamp = int(time.time() * 1000)
response = logs_client.put_log_events(
logGroupName=log_group_name,
logStreamName=stream_name,
logEvents=[{"timestamp": event_timestamp, "message": message}],
)
if response.get("rejectedLogEventsInfo"):
raise RuntimeError(str(response["rejectedLogEventsInfo"]))
first_timestamp = first_timestamp or event_timestamp
last_timestamp = event_timestamp
event_count += 1
message_bytes_total += message_bytes
remaining -= message_bytes + LOG_EVENT_OVERHEAD_BYTES
return {
"partNumber": part_number,
"apiBytes": target_api_bytes,
"messageBytes": message_bytes_total,
"eventCount": event_count,
"firstTimestamp": first_timestamp,
"lastTimestamp": last_timestamp,
}
@durable_step
def start_export_task(
step_context,
log_group_name,
destination_bucket,
destination_prefix,
from_ms,
to_ms,
task_name,
):
# 外部API呼び出しはstep内で実行し、返されたタスクIDを保存します
try:
response = logs_client.create_export_task(
taskName=task_name,
logGroupName=log_group_name,
fromTime=from_ms,
to=to_ms,
destination=destination_bucket,
destinationPrefix=destination_prefix,
)
except ClientError as error:
error_code = error.response.get("Error", {}).get("Code", "Unknown")
step_context.logger.error("CreateExportTask failed: %s", error_code)
raise
return {
"taskId": response["taskId"],
"taskName": task_name,
"from": from_ms,
"to": to_ms,
}
@durable_step
def check_export_status(step_context, task_id):
# taskIdで対象を一件に絞り、CloudWatch Logs側の状態を返します
response = logs_client.describe_export_tasks(taskId=task_id)
tasks = response.get("exportTasks", [])
if not tasks:
return {"status": "UNKNOWN", "message": "Export task was not found"}
task = tasks[0]
status = task["status"]
execution_info = task.get("executionInfo", {})
return {
"status": status["code"],
"message": status.get("message", ""),
"creationTime": execution_info.get("creationTime"),
"completionTime": execution_info.get("completionTime"),
}
@durable_step
def read_incoming_bytes(step_context, log_group_name, first_timestamp):
# CloudWatchのIncomingBytesをロググループ単位で合計します
start_time = datetime.fromtimestamp(
(first_timestamp - 120_000) / 1000,
tz=timezone.utc,
)
end_time = datetime.now(timezone.utc) + timedelta(minutes=1)
response = cloudwatch_client.get_metric_statistics(
Namespace="AWS/Logs",
MetricName="IncomingBytes",
Dimensions=[{"Name": "LogGroupName", "Value": log_group_name}],
StartTime=start_time,
EndTime=end_time,
Period=60,
Statistics=["Sum"],
)
observed_bytes = int(sum(point["Sum"] for point in response.get("Datapoints", [])))
return {"incomingBytes": observed_bytes}
@durable_step
def summarize_s3_objects(step_context, bucket_name, prefix):
# エクスポート先プレフィックス配下の個数と容量を合計します
object_count = 0
compressed_bytes = 0
paginator = s3_client.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
for item in page.get("Contents", []):
object_count += 1
compressed_bytes += item["Size"]
return {
"objectCount": object_count,
"s3CompressedBytes": compressed_bytes,
}
@durable_step
def write_benchmark_result(step_context, bucket_name, prefix, result):
# 集計値を後から比較できるよう、サイズ別プレフィックスへJSONで残します
s3_client.put_object(
Bucket=bucket_name,
Key=f"{prefix}/benchmark-result.json",
Body=json.dumps(result, ensure_ascii=False, indent=2).encode("utf-8"),
ContentType="application/json",
)
return {"resultKey": f"{prefix}/benchmark-result.json"}
NO_RETRY_CONFIG = StepConfig(
step_semantics=StepSemantics.AT_MOST_ONCE_PER_RETRY,
retry_strategy=lambda error, attempt: RetryDecision.no_retry(),
)
@durable_execution
def lambda_handler(event, context: DurableContext):
# 入力値だけを使う処理は、再生されても同じ結果になるように保ちます
log_group_name = event["logGroupName"]
destination_bucket = event["destinationBucket"]
destination_prefix = event.get("destinationPrefix", "durable-export")
target_gb = int(event.get("targetGb", 0))
export_task_name = event["exportTaskName"]
configured_api_bytes = 0
configured_message_bytes = 0
generated_event_count = 0
if target_gb > 0:
stream_name = event["streamName"]
target_bytes = target_gb * 1_000_000_000
context.step(
create_benchmark_stream(log_group_name, stream_name),
config=NO_RETRY_CONFIG,
)
first_timestamp = None
last_timestamp = None
part_count = math.ceil(target_bytes / GENERATION_PART_BYTES)
for part_number in range(1, part_count + 1):
bytes_this_part = min(
GENERATION_PART_BYTES,
target_bytes - configured_api_bytes,
)
part = context.step(
write_benchmark_part(
log_group_name,
stream_name,
part_number,
bytes_this_part,
),
config=NO_RETRY_CONFIG,
)
configured_api_bytes += part["apiBytes"]
configured_message_bytes += part["messageBytes"]
generated_event_count += part["eventCount"]
first_timestamp = first_timestamp or part["firstTimestamp"]
last_timestamp = part["lastTimestamp"]
context.wait(
duration=Duration.from_seconds(1),
name=f"checkpoint-after-generation-part-{part_number}",
)
# IncomingBytesがメトリクスへ現れるまで少し待ちます
context.wait(
duration=Duration.from_minutes(3),
name="wait-for-incoming-bytes-metric",
)
from_ms = first_timestamp - 1_000
to_ms = last_timestamp + 1_000
else:
lookback_minutes = int(event.get("lookbackMinutes", 60))
to_ms = int(event["toMs"])
from_ms = to_ms - (lookback_minutes * 60 * 1000)
first_timestamp = from_ms
export = context.step(
start_export_task(
log_group_name,
destination_bucket,
destination_prefix,
from_ms,
to_ms,
export_task_name,
),
config=NO_RETRY_CONFIG,
)
context.logger.info("Export task started: %s", export["taskId"])
# 15分の単一呼び出し上限を確実に超えるため、まず16分サスペンドします
context.wait(
duration=Duration.from_minutes(16),
name="wait-beyond-fifteen-minutes",
)
for poll_number in range(1, 721):
result = context.step(check_export_status(export["taskId"]))
status = result["status"]
context.logger.info("Poll %d: %s", poll_number, status)
if status == "COMPLETED":
s3_summary = context.step(
summarize_s3_objects(destination_bucket, destination_prefix)
)
incoming = context.step(
read_incoming_bytes(log_group_name, first_timestamp)
)
creation_time = result.get("creationTime")
completion_time = result.get("completionTime")
export_duration_ms = None
if creation_time is not None and completion_time is not None:
export_duration_ms = completion_time - creation_time
final_result = {
**export,
"status": status,
"creationTime": creation_time,
"completionTime": completion_time,
"exportDurationMs": export_duration_ms,
"pollCount": poll_number,
"targetGb": target_gb,
"configuredApiBytes": configured_api_bytes,
"configuredMessageBytes": configured_message_bytes,
"generatedEventCount": generated_event_count,
"incomingBytes": incoming["incomingBytes"],
**s3_summary,
}
result_location = context.step(
write_benchmark_result(
destination_bucket,
destination_prefix,
final_result,
)
)
return {**final_result, **result_location}
if status in {"CANCELLED", "FAILED"}:
raise RuntimeError(
f"Export task ended with {status}: {result.get('message', '')}"
)
# CloudWatch Logs側の完了時刻はミリ秒で取得できるため、確認間隔は1分にします
context.wait(
duration=Duration.from_minutes(1),
name=f"wait-before-poll-{poll_number + 1}",
)
raise TimeoutError("Export task did not finish within the polling window")
コードエディタへ完全版を貼り付け、Deployを選びます。PythonのLambdaランタイムには開発用のDurable Execution SDKが含まれています。本番のデプロイパッケージではSDKのバージョンを固定してください。
CreateExportTaskにはクライアント側の冪等性トークンがないため、開始ステップのDurable SDK側の自動再試行を無効にしています。開始直後に応答を失った場合は、既存タスクを確認してから対応します。
4-3. バージョンを発行して非同期で呼び出す
コードを固定するため、バージョンを発行してからイベントを設定します。
発行済みのバージョン: 1で、呼び出しタイプを非同期にします。同期呼び出しは実行全体が15分以内に制限されるため、今回の16分待機には使いません。イベントJSONは次の内容にします。
{
"logGroupName": "<LOG_GROUP_NAME>",
"streamName": "<LOG_STREAM_NAME>",
"destinationBucket": "<YOUR_EXPORT_BUCKET>",
"destinationPrefix": "durable-export/01gb",
"exportTaskName": "<EXPORT_TASK_NAME>",
"targetGb": 1
}
targetGbは生成する容量、streamNameは新しく作るログストリーム、exportTaskNameはエクスポートタスクの固有名です。ロググループ名とバケット名が、3章の両ポリシーと一致することを確認して保存し、非同期で呼び出します。
5. 16分後の再開とS3出力を確認する
疑似ログの生成後、エクスポートの開始ステップが完了すると、wait-beyond-fifteen-minutesで待機します。16分後の再開後にタスクの状態を確認し、未完了なら1分待って再確認します。
1GBの実行記録では、エクスポートは58.318秒、関数の開始から結果までが21分19.056秒でした。全体にはログ生成やメトリクス反映の待機も含まれます。
結果から確認に使う値を抜粋します。exportDurationMsはCloudWatch Logsが返したcompletionTimeとcreationTimeの差です。
{
"status": "COMPLETED",
"targetGb": 1,
"exportDurationMs": 58318,
"objectCount": 10,
"s3CompressedBytes": 753183698
}
S3では、指定したdurable-export/01gb配下のgzipファイルとbenchmark-result.jsonを確認します。上のオブジェクト数と容量は、結果JSONを書き込む前にプレフィックス配下を集計した値です。
1分の呼び出しタイムアウトで、16分の固定待機を挟んだ後も結果を取得できました。チェックポイントへ保存したtaskIdを使って、後続の状態確認とS3集計へ戻れたことが確認できます。
6. 1〜30GBの追加測定
同じ処理で容量を増やしたときのエクスポート時間も測定しました。測定日は2026年8月14日、条件は2章と同じです。
8回とも15分未満でした。30GBの再測定は初回より2分40.929秒短く、20GBの2回より短い結果です。この記録から、容量に比例する固定の処理速度は求められません。
20GBの初回はエクスポート完了後のS3一覧取得で失敗し、再実行では結果JSONの保存に失敗しました。再実行もエクスポート本体は完了しています。再実行は既存20GBを再利用し、30GBは既存20GBへ10GBを追加したデータを2回書き出しました。
エクスポート時間と実行全体の記録
| 容量と回 | 開始から結果／エラーまで | エクスポート時間 | S3集計 |
|---|---|---|---|
| 1GB | 21分19.056秒 | 58.318秒 | 10個、753,183,698バイト |
| 3GB | 25分42.382秒 | 2分24.975秒 | 35個、2,259,552,374バイト |
| 5GB | 30分04.467秒 | 3分25.245秒 | 68個、3,765,920,682バイト |
| 10GB | 40分51.032秒 | 9分26.079秒 | 137個、7,531,839,241バイト |
| 20GB、初回 | 1時間05分27.055秒 | 13分07.302秒 | 集計前に停止 |
| 20GB、復旧実行 | 17分28.896秒 | 14分26.862秒 | 301個、15,063,677,936バイト |
| 30GB、初回 | 16分04.249秒 | 12分50.980秒 | 432個、22,595,516,921バイト |
| 30GB、再測定 | 16分03.803秒 | 10分10.051秒 | 432個、22,595,516,921バイト |
20GB初回の実行全体には、疑似ログ生成、100MBごとのチェックポイント、3分のメトリクス待機、固定16分待機が含まれます。エクスポート本体は13分07.302秒でした。
測定全体で投入したログは合計49GBで、取り込み費用は無料枠適用前の単価計算で約37.24USDです。疑似データによる一時点の観測値です。
7. 運用時の待機方法と制約
運用では当たり前ですが、固定16分を使わず、短いDurable待機を挟んで状態を確認します。context.wait_for_condition()を使えば、条件確認と待機、バックオフをまとめて記述できます。
CloudWatch Logsの実行中または保留中のエクスポートタスクは、1アカウント・1リージョンにつき1件です。ログがエクスポート可能になるまで最大12時間かかる場合があり、タスクは24時間でタイムアウトします。複数のロググループを扱う場合は直列で実行し、最大待機時間を決めます。
継続的だったりリアルタイム性が必要なログアーカイブには、CloudWatch Logsのサブスクリプションフィルターを検討することになるかと思います。
おわりに
ここまでお付き合いいただきありがとうございます。
Durable FunctionsとCloudWatch LogsのS3エクスポートを組み合わせた試みでした。
ではまた、お会いしましょう。
参考リンク
Lambda Durable Functions
- Durable Functionsの作成 - AWS Lambda
- Durable Functionsの基本概念 - AWS Lambda
- Durable Functionsの設定 - AWS Lambda
- Durable Functionsの呼び出し - AWS Lambda
- Durable Functionsのベストプラクティス - AWS Lambda
- Durable Functionsのアクセス許可 - AWS Lambda
- Durable Functionsで使用できるランタイム - AWS Lambda
- 待機オペレーション - AWS Durable Execution SDK
- 条件を確認しながら待機する - AWS Durable Execution SDK
- AWS Lambdaの料金 - AWS
CloudWatch Logs、IAM、料金
- CreateExportTask - Amazon CloudWatch Logs
- DescribeExportTasks - Amazon CloudWatch Logs
- Amazon S3へのログデータのエクスポート - AWS
- CloudWatch Logsのクォータ - AWS
- PutLogEvents - Amazon CloudWatch Logs
- CloudWatch Logsのアクション、リソース、条件キー - AWS
- Amazon CloudWatchの料金 - AWS