Amazon CloudWatch Logs の S3 テーブル統合機能により Lambda 関数ログを Athena でクエリ可能とする構成を CDK で実装してみた

Lambda 関数
  ↓ 自動（JSON 形式）
CloudWatch Logs（/aws/lambda/xxx）
  ↓ S3 Tables Integration（自動連携）
S3 Table Bucket（aws-cloudwatch / AWS マネージド）
  ↓ Apache Iceberg
Athena（s3tablescatalog 経由）

lib/
├── constructs/
│   ├── lambda/
│   │   ├── index.ts                       # LambdaFunctionConstruct
│   │   └── log-group/index.ts             # LogGroupConstruct
│   ├── s3-tables-integration/
│   │   ├── index.ts                       # S3TablesIntegrationConstruct
│   │   └── service-role/index.ts          # ServiceRoleConstruct
│   ├── s3-tables-catalog/index.ts         # S3TablesCatalogConstruct
│   └── athena-workgroup/
│       ├── index.ts                       # AthenaWorkGroupConstruct
│       └── result-bucket/index.ts         # ResultBucketConstruct
└── lambda-logs-s3-tables-stack.ts         # スタック（4つの construct を組み合わせる）

npm install aws-cdk-lib@latest constructs@latest

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";

import { AthenaWorkGroupConstruct } from "./constructs/athena-workgroup/index";
import { LambdaFunctionConstruct } from "./constructs/lambda/index";
import { S3TablesCatalogConstruct } from "./constructs/s3-tables-catalog/index";
import { S3TablesIntegrationConstruct } from "./constructs/s3-tables-integration/index";

export class LambdaLogsS3TablesStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    /** ログ発生源となる Lambda 関数 */
    const lambdaFunction = new LambdaFunctionConstruct(this, "Lambda");

    /** S3 Tables Integration */
    new S3TablesIntegrationConstruct(this, "S3TablesIntegration", {
      lambdaFunction: lambdaFunction.function,
    });

    /** Glue s3tablescatalog フェデレーテッドカタログ（データパイプライン自体には不要） */
    new S3TablesCatalogConstruct(this, "S3TablesCatalog");

    /** Athena ワークグループ（データパイプライン自体には不要） */
    new AthenaWorkGroupConstruct(this, "AthenaWorkGroup");
  }
}

import * as cdk from "aws-cdk-lib";
import * as lambda from "aws-cdk-lib/aws-lambda";
import { Construct } from "constructs";

import { LogGroupConstruct } from "./log-group";

/** Lambda 関数とそのロググループを管理するコンストラクト。 */
export class LambdaFunctionConstruct extends Construct {
  readonly function: lambda.Function;

  constructor(scope: Construct, id: string) {
    super(scope, id);

    /**
     * 関数名を construct パスから自動導出。
     * dataSource.name（= cw:datasource:name タグ値）は小文字・数字・アンダースコアのみ有効。
     * CDK のデフォルト自動生成名（ハイフン・大文字を含む）はそのままでは使用不可。
     * @see https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/data-source-discovery-management.html#application-logs
     */
    const functionName = cdk.Names.uniqueId(this)
      .toLowerCase()
      .replace(/[^a-z0-9]/g, "_");

    const { logGroup } = new LogGroupConstruct(this, "LogGroup", {
      functionName,
    });

    this.function = new lambda.Function(this, "Default", {
      functionName,
      runtime: lambda.Runtime.NODEJS_24_X,
      handler: "index.handler",
      code: lambda.Code.fromInline(`
exports.handler = async (event) => {
  console.log(JSON.stringify({ level: 'INFO', message: 'invoked' }));
  return { statusCode: 200 };
};`),
      logGroup,
      loggingFormat: lambda.LoggingFormat.JSON, // cwl__message を JSON 形式にするために必要
    });
  }
}

import * as cdk from "aws-cdk-lib";
import * as logs from "aws-cdk-lib/aws-logs";
import { Construct } from "constructs";

/**
 * Lambda 関数のロググループと CloudWatch Data Sources タグを管理するコンストラクト。
 */
export class LogGroupConstruct extends Construct {
  readonly logGroup: logs.LogGroup;

  constructor(scope: Construct, id: string, props: { functionName: string }) {
    super(scope, id);

    const { functionName } = props;

    this.logGroup = new logs.LogGroup(this, "Default", {
      logGroupName: `/aws/lambda/${functionName}`,
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 検証用スタックのため。本番では RETAIN を推奨
    });

    /**
     * CloudWatch Data Sources として認識させるためのタグ。
     * S3 Tables Integration がログを正しくルーティングするために必要。
     * @see https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/data-source-discovery-management.html#application-logs
     */
    cdk.Tags.of(this.logGroup).add("cw:datasource:name", functionName);
    cdk.Tags.of(this.logGroup).add("cw:datasource:type", "lambda");
  }
}

import * as lambda from "aws-cdk-lib/aws-lambda";
import * as observabilityadmin from "aws-cdk-lib/aws-observabilityadmin";
import { Construct } from "constructs";

import { ServiceRoleConstruct } from "./service-role";

interface S3TablesIntegrationProps {
  lambdaFunction: lambda.Function;
}

/**
 * CloudWatch Logs の S3 Tables Integration を管理するコンストラクト。
 *
 * - S3 Tables サービスロール（CloudWatch Logs が assume して S3 Tables に書き込む）
 * - S3 Tables Integration（CfnS3TableIntegration）
 */
export class S3TablesIntegrationConstruct extends Construct {
  constructor(scope: Construct, id: string, props: S3TablesIntegrationProps) {
    super(scope, id);

    const { lambdaFunction } = props;

    const { serviceRole } = new ServiceRoleConstruct(this, "ServiceRole", {
      lambdaFunction,
    });

    /** S3 Tables Integration */
    new observabilityadmin.CfnS3TableIntegration(
      this,
      "Default",
      {
        roleArn: serviceRole.roleArn,
        encryption: {
          sseAlgorithm: "AES256",
        },
        logSources: [
          {
            name: lambdaFunction.functionName,
            type: "lambda",
          },
        ],
      },
    );
  }
}

import * as cdk from "aws-cdk-lib";
import * as iam from "aws-cdk-lib/aws-iam";
import * as lambda from "aws-cdk-lib/aws-lambda";
import { Construct } from "constructs";

/**
 * CloudWatch Logs が S3 Tables に書き込む際に assume するサービスロール。
 * @see https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/s3-tables-integration.html#service-role-iam-policies
 */
export class ServiceRoleConstruct extends Construct {
  readonly serviceRole: iam.Role;

  constructor(scope: Construct, id: string, props: { lambdaFunction: lambda.Function }) {
    super(scope, id);

    const { lambdaFunction } = props;

    const stack = cdk.Stack.of(this);
    const accountId = stack.account;

    /**
     * ロググループ ARN の組み立て
     *
     * MEMO: lambdaFunction.logGroup.logGroupArn（末尾に :* 付き）は使えない。
     *       CloudWatch が STS に送る aws:SourceArn は :* なしのため、
     *       ArnLike 条件がミスマッチして S3TableIntegrationUnauthorized が発生する。
     *       logGroupName から cdk.Arn.format で組み立てることで :* を除去する。
     */
    const logGroupArn = cdk.Arn.format(
      {
        service: "logs",
        resource: "log-group",
        resourceName: lambdaFunction.logGroup.logGroupName,
        arnFormat: cdk.ArnFormat.COLON_RESOURCE_NAME, // log-group ARN はコロン区切り（arn:...:log-group:name）
      },
      stack,
    );

    this.serviceRole = new iam.Role(this, "Default", {
      assumedBy: new iam.ServicePrincipal("logs.amazonaws.com", {
        conditions: {
          StringEquals: { "aws:SourceAccount": accountId },
          ArnLike: { "aws:SourceArn": logGroupArn },
        },
      }),
    });
    this.serviceRole.addToPolicy(
      new iam.PolicyStatement({
        actions: ["logs:integrateWithS3Table"],
        resources: [logGroupArn],
        conditions: {
          StringEquals: { "aws:ResourceAccount": accountId },
        },
      }),
    );
  }
}

import * as cdk from "aws-cdk-lib";
import * as glue from "aws-cdk-lib/aws-glue";
import { Construct } from "constructs";

/**
 * Glue s3tablescatalog フェデレーテッドカタログ。
 * @see https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-aws.html#table-integration-procedures
 */
export class S3TablesCatalogConstruct extends Construct {
  constructor(scope: Construct, id: string) {
    super(scope, id);

    /** IAM アクセス制御モードで統合する設定 */
    const iamAllowedPrincipalsAll: glue.CfnCatalog.PrincipalPermissionsProperty = {
      principal: { dataLakePrincipalIdentifier: "IAM_ALLOWED_PRINCIPALS" },
      permissions: ["ALL"],
    };

    const stack = cdk.Stack.of(this);

    new glue.CfnCatalog(this, "Default", {
      name: "s3tablescatalog",
      federatedCatalog: {
        identifier: cdk.Arn.format(
          {
            service: "s3tables",
            resource: "bucket",
            resourceName: "*",
          },
          stack,
        ),
        connectionName: "aws:s3tables",
      },
      createDatabaseDefaultPermissions: [iamAllowedPrincipalsAll], // IAM アクセス制御モード：新規 DB に IAM_ALLOWED_PRINCIPALS:ALL を自動付与
      createTableDefaultPermissions: [iamAllowedPrincipalsAll],    // IAM アクセス制御モード：新規テーブルに IAM_ALLOWED_PRINCIPALS:ALL を自動付与
      allowFullTableExternalDataAccess: "True",                    // 外部クエリエンジン（Athena 等）からのフルアクセスを許可
    });
  }
}

import * as athena from "aws-cdk-lib/aws-athena";
import { Construct } from "constructs";

import { ResultBucketConstruct } from "./result-bucket";

/**
 * Athena ワークグループとクエリ結果バケットを管理するコンストラクト。
 */
export class AthenaWorkGroupConstruct extends Construct {
  constructor(scope: Construct, id: string) {
    super(scope, id);

    const { bucket } = new ResultBucketConstruct(this, "ResultBucket");

    new athena.CfnWorkGroup(this, "Default", {
      name: "lambda-logs-s3tables",
      workGroupConfiguration: {
        resultConfiguration: {
          outputLocation: `s3://${bucket.bucketName}/results/`,
        },
      },
      recursiveDeleteOption: true, // スタック削除時にクエリ結果バケットも削除する（検証用スタックのため。本番では false を推奨）
    });
  }
}

import * as cdk from "aws-cdk-lib";
import * as s3 from "aws-cdk-lib/aws-s3";
import { Construct } from "constructs";

/**
 * Athena クエリ結果を格納する S3 バケット。
 */
export class ResultBucketConstruct extends Construct {
  readonly bucket: s3.Bucket;

  constructor(scope: Construct, id: string) {
    super(scope, id);

    this.bucket = new s3.Bucket(this, "Default", {
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 検証用スタックのため。本番では RETAIN を推奨
      autoDeleteObjects: true,
    });
  }
}

npx cdk deploy LambdaLogsS3Tables

aws lambda invoke \
  --function-name <関数名> \
  --payload '{}' \
  --cli-binary-format raw-in-base64-out \
  /dev/null

EXEC_ID=$(aws athena start-query-execution \
  --query-string 'SELECT * FROM "s3tablescatalog/aws-cloudwatch"."logs"."<テーブル名>" ORDER BY cwl__timestamp DESC LIMIT 1' \
  --work-group "lambda-logs-s3tables" \
  --result-configuration "OutputLocation=s3://<結果バケット>/results/" \
  --query 'QueryExecutionId' --output text)

aws athena get-query-results \
  --query-execution-id "$EXEC_ID" \
  --output json

cwl__aws_account:      XXXXXXXXXXXX
cwl__ingestiontime:    2026-05-04 15:49:27.964000 UTC
cwl__logstreamid:      99022dc4-...
cwl__log:              XXXXXXXXXXXX:/aws/lambda/lambdalogss3tableslambdad926ab1c
cwl__aws_region:       ap-northeast-1
cwl__data_source_name: lambdalogss3tableslambdad926ab1c
cwl__logstream:        2026/05/04/lambdalogss3tableslambdad926ab1c[$LATEST]...
cwl__data_format:      Custom
cwl__source_log:
cwl__message:          {"timestamp":"2026-05-04T15:49:27.860Z","level":"INFO","requestId":"ebc052ad-...","message":"{\"level\":\"INFO\",\"message\":\"invoked\"}"}
cwl__timestamp:        2026-05-04 15:49:27.860000 UTC
cwl__data_source_type: lambda
cwl__loggroupid:       99022dc4-4c7a-4349-9a27-57cc3b3d1704

{
  "timestamp": "2026-05-04T15:49:27.860Z",
  "level": "INFO",
  "requestId": "ebc052ad-5b4e-4d43-b90e-a530bb4b06fa",
  "message": "{\"level\":\"INFO\",\"message\":\"invoked\"}"
}

// ❌ CDK のデフォルト自動生成名（ハイフン・大文字を含む）
//    例: LambdaLogsS3Tables-LambdaDefault-u7KEHtRqDI4S
//    → AssociateSourceToS3TableIntegration の name に使うと ValidationException

// ✅ cdk.Names.uniqueId から小文字・アンダースコアのみに変換して使用
this.functionName = cdk.Names.uniqueId(this)
  .toLowerCase()
  .replace(/[^a-z0-9]/g, "_");
// 例: lambdalogss3tableslambdad926ab1c

// ❌ 間違い：末尾 :* 付き
assumedBy: new iam.ServicePrincipal("logs.amazonaws.com", {
  conditions: {
    ArnLike: { "aws:SourceArn": fn.logGroup.logGroupArn }, // :* 付き → ミスマッチ
  },
}),

// ✅ 正しい：cdk.Arn.format で :* なし ARN を組み立て
const logGroupArn = cdk.Arn.format({
  service: "logs",
  resource: "log-group",
  resourceName: lambdaFunction.logGroup.logGroupName,
  arnFormat: cdk.ArnFormat.COLON_RESOURCE_NAME,
});

assumedBy: new iam.ServicePrincipal("logs.amazonaws.com", {
  conditions: {
    ArnLike: { "aws:SourceArn": logGroupArn }, // :* なし → 正しくマッチ
  },
}),

cdk.Tags.of(this.logGroup).add("cw:datasource:name", functionName);
cdk.Tags.of(this.logGroup).add("cw:datasource:type", "lambda");

2026-05-04T15:03:34.946Z\t6929ead1-...\tINFO\t{"level":"INFO","message":"invoked"}

this.function = new lambda.Function(this, "Default", {
  // ...
  loggingFormat: lambda.LoggingFormat.JSON, // cwl__message を JSON 形式にするために必要
});