0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@guupys

【コツコツAWS】KiroのUser activityを集計・通知する簡単な仕組みを作成しました

0
Last updated at Posted at 2026-09-09

KiroのEnterprise版を利用すると、利用状況(クレジット消費量・チャット回数など)を、S3に日次で出力してくれます。

ただ、毎日出力されるcsvファイルだけで利用状況を把握するのは難しく、Kiroのコンソールではユーザーごとの集計ができない…
ということで、csvファイルを集計して利用状況を把握するための仕組みを作りました。

今回作ったのは、週次のレポートメールと、月間クレジットの90%に到達した際のアラートメールの2つです。
なるべく少ないリソースで最低限の情報さえ得られればいいや、という方針で作っているので、ダッシュボードを作るとかまではしていません。

1. 今回の構成

全体構成は以下の通りです。

S3 (Kiro user_report CSV)
        │ 
        ├─────────────────────────────────────┐
        ▼                                     ▼
EventBridge Scheduler (日次)          EventBridge Scheduler (毎週月曜)
        │                                     │
        ▼                                     ▼
Lambda: クレジット閾値チェック         Lambda: 週次レポート
   │              │                           │
   │              │                           │
   ▼              └───────────┐         ┌─────┘
DynamoDB                      ▼         ▼
(累積値記録)            SNS: kiro-notifications
                     (90%到達アラート・週次レポート)

前提として、Kiroのコンソールでは確認できないけど、把握しておきたかった点は以下2つ。

  • ユーザーごとの利用量
  • 月間クレジットの上限に近づいているユーザーの有無

上記の項目を集計する仕組みをサーバーレスで構成する場合、Amazon AthenaやAmazon Quickを使うことも候補になりそうですが、
なるべく少ないリソースで完結したかったので、EventBridgeとLambdaで処理しています。
DynamoDBは月ごとのクレジット累積値を記録するために使用しました。

KiroのレポートCSVにあるCredits_Used列はその日1日分の消費量で、月内の累積値ではありません。ユーザーごとの月間のクレジット使用量を見る場合は、日々の使用量を自前で合算する必要がありました。

また、Lambda関数は役割ごとに2つ(クレジット閾値チェック用・週次レポート用)。責務が違うのでIAMロールは分けてますが、通知先は同じSNSトピックに統合しています。

SNSの配信だとプレーンテキストのみでHTML装飾はできませんが、今回は数名のユーザーの利用状況を知りたいだけなのでSNSを採用しました。

2. DynamoDBテーブル

「月の累積クレジット使用量」だけを持つテーブルです。以下の内容で作成しました。

PK: user_id (String)
SK: year_month (String, 例: "2026-08")
属性:
  - credits_used_month_to_date (Number)   # 直近チェック時点での当月累積Credits_Used
  - last_checked_date (String)            # 最終更新日
TTL属性: expire_at                        # データ削除までの期間をお好みで設定
課金モード: オンデマンド

アラートを送るのは「初めて閾値を超えた時」だけでいいので、前回チェック時点の累積値を保存し、前回値 < 閾値 <= 今回値という判定を行うようにしています。
year_monthをソートキーにしているので、月が変わるとキー自体が変わり、新しい月にはまだアイテムが無い状態から始まります。

3. Lambdaの実装

主な部分だけを記事には載せます。
詳細にご興味のある方はリポジトリを確認ください。

S3のキー組み立て・CSVパース・集計は2つのLambdaで共通なので、Lambda Layerに切り出しました。

kiro_report.py(抜粋)
def build_report_key(
    prefix: str, account_id: str, region: str, client_type: str, target_date: date
) -> str:
    """指定日・指定クライアントタイプのuser_report CSVのS3キー(オブジェクトキー)を組み立てる。
    Kiroの実ファイル名の日時部分は yyyyMMddHHmm 形式で、時分は常に "0000" 固定。
    """
    yyyymmddhhmm = target_date.strftime("%Y%m%d") + "0000"
    path = (
        f"AWSLogs/{account_id}/KiroLogs/user_report/{region}/"
        f"{target_date:%Y}/{target_date:%m}/{target_date:%d}/00/"
        f"{client_type}_{account_id}_user_report_{yyyymmddhhmm}.csv"
    )
    if prefix:
        return f"{prefix.rstrip('/')}/{path}"
    return path


def load_usage_for_period(
    s3_client,
    bucket: str,
    prefix: str,
    account_id: str,
    region: str,
    client_types: list[str],
    start_date: date,
    end_date: date,
) -> list[UserDailyUsage]:
    """指定期間・複数クライアントタイプ分のCSVをS3から読み込み、日次利用データのリストを返す。
    その日のレポートがまだ生成されていない等でオブジェクトが存在しない場合はスキップする。
    """
    usages: list[UserDailyUsage] = []
    for target_date in daterange(start_date, end_date):
        for client_type in client_types:
            key = build_report_key(prefix, account_id, region, client_type, target_date)
            try:
                obj = s3_client.get_object(Bucket=bucket, Key=key)
            except s3_client.exceptions.NoSuchKey:
                continue
            csv_text = obj["Body"].read().decode("utf-8")
            usages.extend(extract_daily_usage(parse_csv_rows(csv_text)))
    return usages

クレジット閾値チェックLambdaのメイン処理は以下の通りです。

閾値チェックメイン部分
def handler(event, context):
    today = datetime.now(JST).date()
    year_month = today.strftime("%Y-%m")
    month_start = today.replace(day=1)

    usages = load_usage_for_period(
        s3, BUCKET_NAME, BUCKET_PREFIX, ACCOUNT_ID, REGION, CLIENT_TYPES, month_start, today
    )
    current_by_user = sum_credits_by_user(usages)

    alerted_users = []
    for user_id, info in current_by_user.items():
        tier = info["tier"]
        email = info["email"]
        current = info["credits_used"]

        threshold = TIER_ALERT_THRESHOLDS.get(tier)
        if threshold is None:
            # 未知のTierは判定スキップ
            continue

        previous = _get_previous_cumulative(user_id, year_month)

        if previous < threshold <= current:
            _publish_alert(email, tier, current, threshold)
            alerted_users.append(user_id)

        _put_cumulative(user_id, year_month, current, today)

    return {"checked_users": len(current_by_user), "alerted_users": alerted_users}

週次レポートLambdaは、直近7日間と、さらにその前の7日間のcsvを読んで前週比を算出。
月累計は自前で再集計せず、閾値チェックLambdaが更新したDynamoDBの値をそのまま読みます。

週次レポートメイン部分
def handler(event, context):
    today = datetime.now(JST).date()
    report_week_end = today - timedelta(days=1)
    report_week_start = report_week_end - timedelta(days=6)
    comparison_week_end = report_week_start - timedelta(days=1)
    comparison_week_start = comparison_week_end - timedelta(days=6)

    # 実行日(today)ではなく、レポート対象週の最終日を基準に当月を決める。
    year_month = report_week_end.strftime("%Y-%m")

    report_week_usages = load_usage_for_period(
        s3, BUCKET_NAME, BUCKET_PREFIX, ACCOUNT_ID, REGION, CLIENT_TYPES, report_week_start, report_week_end
    )
    comparison_week_usages = load_usage_for_period(
        s3, BUCKET_NAME, BUCKET_PREFIX, ACCOUNT_ID, REGION, CLIENT_TYPES, comparison_week_start, comparison_week_end
    )

    report_week_by_user = sum_credits_by_user(report_week_usages)
    comparison_week_by_user = sum_credits_by_user(comparison_week_usages)

    lines = []
    for user_id, info in sorted(report_week_by_user.items(), key=lambda kv: kv[1]["email"]):
        email = info["email"]
        tier = info["tier"]
        report_week_credits = info["credits_used"]
        comparison_week_credits = comparison_week_by_user.get(user_id, {}).get("credits_used", 0.0)
        change_pct = calc_change_pct(comparison_week_credits, report_week_credits)
        month_to_date = _get_month_to_date(user_id, year_month, fallback=report_week_credits)

        lines.append(_format_user_line(email, report_week_credits, change_pct, month_to_date, tier))

    message = "\n\n".join(lines) if lines else "対象期間中の利用はありませんでした。"
    subject = f"[Kiro] 週次利用レポート ({report_week_start.isoformat()} ~ {report_week_end.isoformat()})"
    sns.publish(TopicArn=NOTIFICATION_TOPIC_ARN, Subject=subject[:100], Message=message)

    return {"reported_users": len(report_week_by_user)}

4. EventBridge Schedulerでのスケジュール設定

EventBridge SchedulerはScheduleExpressionTimezoneでcron式の解釈タイムゾーンを指定できるので、日本時間の昼12時に両方のLambdaを実行する形にしました。

# クレジット閾値チェック(日次)
ScheduleExpression: cron(0 12 * * ? *)
ScheduleExpressionTimezone: Asia/Tokyo

# 週次レポート
ScheduleExpression: cron(0 12 ? * MON *)
ScheduleExpressionTimezone: Asia/Tokyo

5. 通知される内容

SNSの設定など、細かい部分は記事では省略します。
結果として受け取れるメールの通知は、以下のような形式にしました。

週次レポートのメール本文
user01@example.com
  利用量: ** credits (前週比: -**%) 月利用率: **%

user02@example.com
  利用量: ** credits (前週比: -**%) 月利用率: **%
アラートメールの概要
タイトル: 
[Kiro]user01@example.comが月間クレジットの90%に到達

本文: 
user01@example.com (PRO)
当月累計: ** / ** credits (90%ライン: **)

まとめ

KiroのUser activityのcsvファイルを集計し、簡単な使用状況を通知する仕組みを作りました。
CDKで実装して以下のリポジトリに置いています。

最低限のものをという方針で作りましたが、実際に使ってみると物足りなくなるかもしれないので、必要に応じて改良したいと思います。(AthenaやQuickを使った場合との比較もしてみたい)

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?