Amazon RDS for Db2 の監査ログをもっと使いこなす — CloudFormationでの自動化とDELファイルの読み方

Last updated at 2026-07-23Posted at 2026-07-23

こんにちは、インサイトテクノロジーの松尾です！

前回の投稿「エンタープライズで使われ続ける Db2 —— Amazon RDS for Db2 Community Edition で監査ログ出力を検証する」では、DB2_AUDITオプションを使って監査ログをS3に出力し、自分の操作(テーブル作成やSQL実行)がきちんと記録されていることまで確認しました。

本稿は、ボリュームの都合で本編に入りきらなかった2つのテーマを扱う続編です。

db2-ceインスタンス本体と監査ログ基盤(S3バケット・IAMロール・オプショングループ)をCloudFormationで一括構築する
EXECUTE WITH DATAで記録される実際のSQLテキストをPythonで読み解く

1. db2-ceインスタンスと監査ログ基盤をCloudFormationで構築する

本編ではコンソールでの構築手順を中心に紹介しましたが、コンソールでポチポチ設定していった内容は、CloudFormationで再現できます。ここではDBインスタンス本体と、監査ログ基盤一式(S3・IAM・オプショングループ)の両方をテンプレート化してみます。

1.1. コンソールとの違い

CFn化するにあたって、コンソールとの違いが2点ありました。

1つ目は、AWS License Manager 設定です。コンソールでは名前入力が必須でしたが、公式アナウンスブログのCLI例にはLicense Manager関連のパラメータは一切出てきません。実際、AWS License Managerによるライセンス使用状況の追跡はdb2-se/db2-ae向けの機能として案内されており(Amazon RDS for Db2 licensing options)、ライセンス費用が発生しないdb2-ceではCLI/CloudFormation経由の作成時にLicense Manager設定は不要でした。コンソールの入力必須項目が、実はAPIレベルでは必須ではないという、これも実機検証ならではの発見です。

2つ目は、本編4.3節で見つかったオプショングループの新規作成がConsoleでdb2-ceを選べないという制限です。RDSコンソールの「オプショングループの作成」画面では、エンジンの選択肢にdb2-ceが存在せず(db2-ae/db2-seのみ)、db2-ce用のオプショングループは作れませんでした。一方、AWS CLIのaws rds create-option-group --engine-name db2-ce ...は問題なく成功し、その後の「オプションの追加」はConsole上で完結できることも確認済みです。CloudFormationも裏側では同じRDS APIを叩いているだけなので、AWS::RDS::OptionGroupリソースでdb2-ce用のオプショングループを定義しても、同様に問題なく作成できるはずです。

ここで一点、混同しないように補足しておきます。本編4.3節の制限はRDSコンソールの「オプショングループの作成」画面固有の話で、これから使うCloudFormationコンソールは別物です。CloudFormationコンソールからスタックを作成する場合、裏側ではCLIと同じRDS APIが呼ばれるので、RDSコンソールの制限を回避できます。

DBインスタンス本体のテンプレート(1.2節)は、CloudFormationコンソールから実際にスタックを作成し、DBeaverからの接続確認まで済ませています。 監査ログ基盤(特にオプショングループ)の部分(1.3節)については、本編で検証したのは同等のAWS CLIコマンド(create-option-group)であり、CloudFormation経由での動作はまだ検証できていません。CLIで通ることは確認済みなので動く可能性は高いと考えていますが、試される場合はその点を踏まえた上で検証してみてください。

1.2. DBインスタンス本体のテンプレート

パスワードの扱いは、コンソールでの「セルフマネージド + 自動生成」から一歩進めて、CloudFormationではManageMasterUserPassword: trueを指定してAWS Secrets Managerにパスワード管理を任せる構成にしました。テンプレートにパスワードを平文で持たせずに済み、ローテーションも任せられるので、コード化するならこちらの方が安全です。

以下がテンプレート全文です。VPCやサブネットは既存のものを利用する前提で、IDをパラメータとして受け取る形にしています。IBM Customer ID / IBM Site IDはNoEcho: trueにして、CloudFormationコンソールやCLI出力にそのまま表示されないようにしています。

また、複数のスタックを並行して作成できるように、DBインスタンス識別子をはじめとするリソース名は固定値のパラメータではなく、${AWS::StackName}という擬似パラメータから組み立てるようにしています。これはCloudFormationがスタックごとに自動的に埋めてくれる値なので、スタック名さえ変えれば同じテンプレートを使い回しても名前が衝突しません。

AWSTemplateFormatVersion: "2010-09-09"
Description: >
  Template to build an RDS for Db2 Community Edition (db2-ce) instance.
  The option group for audit logging (DB2_AUDIT) is attached separately
  via the template in section 1.3. Resource names are derived from the
  stack name (${AWS::StackName}) so multiple stacks can coexist.

Parameters:
  VpcId:
    Type: AWS::EC2::VPC::Id
    Description: VPC to place the DB instance in
  SubnetIds:
    Type: List<AWS::EC2::Subnet::Id>
    Description: Subnets for the DB subnet group (2 or more AZs)
    # Must be public subnets (route table has a route to an Internet Gateway)
    # if PubliclyAccessible is true below, otherwise the instance gets a
    # public IP but is still unreachable from outside the VPC.
  AllowedCidr:
    Type: String
    Description: CIDR allowed to access the Db2 port (50000), e.g. your working location's IP
  MasterUsername:
    Type: String
    Default: dbadmin
  DBInstanceClass:
    Type: String
    Default: db.t3.small
  EngineVersion:
    Type: String
    Default: 12.1.4.0.sb00085812.r1
    Description: >
      Latest value as of authoring. Check the latest version string with
      `aws rds describe-db-engine-versions` before deploying.
  IbmCustomerId:
    Type: String
    NoEcho: true
    Description: IBM Customer ID obtained via free registration on the IBM site
  IbmSiteId:
    Type: String
    NoEcho: true
    Description: IBM Site ID obtained via free registration on the IBM site

Resources:
  DBSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    Properties:
      DBSubnetGroupDescription: !Sub "Subnet group for ${AWS::StackName}"
      SubnetIds: !Ref SubnetIds

  DBSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub "Allow Db2 (50000) access for ${AWS::StackName}"
      VpcId: !Ref VpcId
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 50000
          ToPort: 50000
          CidrIp: !Ref AllowedCidr

  DBParameterGroup:
    Type: AWS::RDS::DBParameterGroup
    Properties:
      Description: !Sub "${AWS::StackName} parameter group (IBM ID for BYOL)"
      Family: db2-ce-12.1
      Parameters:
        rds.ibm_customer_id: !Ref IbmCustomerId
        rds.ibm_site_id: !Ref IbmSiteId

  DBInstance:
    Type: AWS::RDS::DBInstance
    DeletionPolicy: Delete
    Properties:
      DBInstanceIdentifier: !Sub "rds-db2ce-${AWS::StackName}"
      Engine: db2-ce
      EngineVersion: !Ref EngineVersion
      DBInstanceClass: !Ref DBInstanceClass
      AllocatedStorage: "20"
      StorageType: gp3
      DBName: TESTDB
      MasterUsername: !Ref MasterUsername
      ManageMasterUserPassword: true
      DBSubnetGroupName: !Ref DBSubnetGroup
      DBParameterGroupName: !Ref DBParameterGroup
      VPCSecurityGroups:
        - !GetAtt DBSecurityGroup.GroupId
      PubliclyAccessible: true
      MultiAZ: false
      BackupRetentionPeriod: 0
      StorageEncrypted: false
      # 本番運用では PubliclyAccessible: false、StorageEncrypted: true、
      # BackupRetentionPeriod を1以上にすることを推奨

Outputs:
  DBEndpointAddress:
    Value: !GetAtt DBInstance.Endpoint.Address
  DBEndpointPort:
    Value: !GetAtt DBInstance.Endpoint.Port

デプロイはaws cloudformation deployでも実行できますが、本稿では進行状況が視覚的に追いやすいCloudFormationコンソールからスタックを作成します。

CloudFormationコンソール → スタック → スタックの作成と進むと、4ステップのウィザードが始まります。

ステップ1「スタックの作成」では、まず「前提条件 - テンプレートの準備」で既存のテンプレートを選択、続く「テンプレートの指定」の「テンプレートソース」でテンプレートファイルのアップロードを選び、上記のYAMLファイル(db2-create.yml)を指定します。ファイルを選択すると、CloudFormationが裏で自動的にS3へアップロードしてくれて、そのS3 URLが画面下部に表示されます。

ここで一つ罠を踏みました。テンプレートファイルを改行コードCRLF(Windows形式)で保存した状態でアップロードすると、Cannot parse template as YAML : special characters are not allowedというエラーになります。YAMLの構文自体は誤っておらず、AWS CLIやaws cloudformation deploy経由なら同じ内容でも問題なく通るのですが、Consoleの「テンプレートファイルのアップロード」機能はアップロード時に\rを特殊文字として弾いてしまうようです。テンプレートファイルはUTF-8 + LF(Unix改行)で保存するのが無難です。

ステップ2「スタックの詳細を指定」で、スタック名(例: db2ce-demo1)とパラメータ(VpcId / SubnetIds / AllowedCidr / IbmCustomerId / IbmSiteIdなど)を入力します。ここに入力したスタック名がそのままDBインスタンス識別子などに埋め込まれるので、別の環境をもう一つ作りたくなったら、この画面でスタック名を変えるだけで済みます。

ステップ3「スタックオプションの設定」はデフォルトのままで問題ありません。ステップ4「確認して作成」で内容を確認し、送信するとスタックの作成が始まります。このテンプレートはIAMリソースを作成しないので、後述の監査ログ基盤テンプレートと違って権限に関する確認チェックボックスは出てきません。

「イベント」タブでリソースが順番に作成されていく様子を確認できます。ステータスがCREATE_COMPLETEになれば完了です。

「出力」タブにDBEndpointAddressとDBEndpointPortが表示されるので、これを使えば本編3.2節と同じ手順でDBeaverから接続できます。

ここで一点、本編と異なる点があります。本編ではコンソールの「セルフマネージド + 自動生成」でパスワードを作成したため、作成直後のバナーからそのままパスワードを確認できました。一方、このテンプレートではManageMasterUserPassword: trueを使っているため、パスワードはテンプレートに現れず、AWS Secrets Managerが自動生成・管理しています。取得するには2通りの方法があります。

コンソールからは、RDSコンソール → データベース → 対象のDBインスタンス → 「接続とセキュリティ」タブを開くと、「マスタークレデンシャルのARN」の下に 「シークレットマネージャーで表示」 というリンクがあるので、ここからSecrets Managerの画面に飛んで「シークレット値を取得する」を押せば確認できます。

CLIからは以下のように取得できます。

SECRET_ARN=$(aws rds describe-db-instances \
  --db-instance-identifier rds-db2ce-db2ce-demo1 \
  --query "DBInstances[0].MasterUserSecret.SecretArn" \
  --output text)

aws secretsmanager get-secret-value \
  --secret-id "$SECRET_ARN" \
  --query SecretString --output text | python3 -m json.tool

passwordキーの値がマスターパスワードです。DBeaverの接続設定にはユーザー名dbadminとあわせてこの値を入力すれば、本編3.2節と同じ手順で接続できます。

同じデプロイをCLIで行う場合は以下のコマンドになります。IBM Customer ID / IBM Site IDをシェル履歴にそのまま残したくない場合は、--parameter-overridesをコマンドラインに直書きせず、パラメータファイル(Git管理外にする)を使う方法もあります。

aws cloudformation deploy \
  --template-file db2-create.yml \
  --stack-name db2ce-demo1 \
  --parameter-overrides \
      VpcId=vpc-xxxxxxxx \
      SubnetIds=subnet-aaaa,subnet-bbbb \
      AllowedCidr=203.0.113.0/32 \
      IbmCustomerId=xxxxxxx \
      IbmSiteId=xxxxxxxxx \
  --capabilities CAPABILITY_NAMED_IAM

1.3. 監査ログ基盤(S3・IAM・オプショングループ)のテンプレート

続いて、監査ログ基盤一式を別テンプレートとして定義します。中身は本編4.1〜4.3の内容をそのままIaC化したものです。こちらもIAMロール・ポリシー名を${AWS::StackName}から組み立てて、スタックごとに衝突しないようにしています(S3バケット名だけはAWS全体でグローバルに一意である必要があるため、引き続きパラメータとして個別に指定する形にしています)。

AWSTemplateFormatVersion: "2010-09-09"
Description: >
  Template to build the audit logging infrastructure (S3 bucket, IAM role,
  and an option group with the DB2_AUDIT option) for RDS for Db2 Community
  Edition. See the template in section 1.2 for the DB instance itself.
  IAM role/policy names are derived from the stack name (${AWS::StackName})
  so multiple stacks can coexist.

Parameters:
  AuditBucketName:
    Type: String
    Description: S3 bucket name for audit logs (must be globally unique; use a different value per stack)

Resources:
  AuditLogBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref AuditBucketName
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true

  AuditIamPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      ManagedPolicyName: !Sub "rds-db2-audit-policy-${AWS::StackName}"
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Sid: Statement1
            Effect: Allow
            Action:
              - s3:ListBucket
              - s3:GetBucketAcl
              - s3:GetBucketLocation
            Resource:
              - !GetAtt AuditLogBucket.Arn
          - Sid: Statement2
            Effect: Allow
            Action:
              - s3:PutObject
              - s3:ListMultipartUploadParts
              - s3:AbortMultipartUpload
            Resource:
              - !Sub "${AuditLogBucket.Arn}/*"
          - Sid: Statement3
            Effect: Allow
            Action:
              - s3:ListAllMyBuckets
            Resource:
              - "*"
      # s3:ListAllMyBuckets は、RDSがS3バケットの所有者アカウントと
      # DBインスタンスの所有者アカウントが一致しているかを内部確認するために必要
      # (本編4.2節参照)

  AuditIamRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "rds-db2-audit-role-${AWS::StackName}"
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: rds.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Ref AuditIamPolicy

  AuditOptionGroup:
    Type: AWS::RDS::OptionGroup
    Properties:
      EngineName: db2-ce
      MajorEngineVersion: "12.1"
      OptionGroupDescription: !Sub "db2-ce audit option group (${AWS::StackName})"
      OptionConfigurations:
        - OptionName: DB2_AUDIT
          OptionSettings:
            - Name: IAM_ROLE_ARN
              Value: !GetAtt AuditIamRole.Arn
            - Name: S3_BUCKET_ARN
              Value: !GetAtt AuditLogBucket.Arn

Outputs:
  OptionGroupName:
    Value: !Ref AuditOptionGroup
    Description: >
      Apply this option group name to the DB instance using the steps described below.

1.4. 監査ログ基盤のデプロイとDBインスタンスへの適用

こちらもCloudFormationコンソールから作成します。手順は1.2節と同じ(スタックの作成 → テンプレートファイルのアップロード → スタック名とパラメータを入力)ですが、一点差があります。

このテンプレートはRoleName/ManagedPolicyNameを明示的に指定してIAMリソースを作成するため、「レビュー」画面で 「AWS CloudFormation によって IAM リソースがカスタム名で作成される場合があることを承認します」 のチェックボックスにチェックを入れる必要があります(CLIでの--capabilities CAPABILITY_NAMED_IAMに相当するものです)。チェックを入れずに送信しようとするとエラーになるので注意してください。

スタックがCREATE_COMPLETEになったら、「出力」タブでOptionGroupNameを確認します。

作成されたオプショングループは、このテンプレート単体では自動的にDBインスタンスへ適用されません。1.2節のDBインスタンス用テンプレートにOptionGroupNameプロパティを追加して再デプロイするか、本編4.3節と同じ要領で RDSコンソール → データベース → 対象のインスタンスを選択 → 変更 → オプショングループを先ほど確認した名前に変更 → すぐに適用 で結び付けます。

CLIから行う場合は以下のコマンドです。

aws rds modify-db-instance \
  --db-instance-identifier rds-db2ce-db2ce-demo1 \
  --option-group-name db2ce-demo1-audit-auditoptiongroup-xxxxxxxx \
  --apply-immediately

本編で確認した通り、DB2_AUDITオプションの適用にDBインスタンスの再起動は不要です。

なお、DBインスタンス用テンプレートのOptionGroupNameにこのオプショングループを!Refすれば、1つのテンプレート・1回のdeployにまとめることも技術的には可能です(依存関係は一方向なので循環参照にはなりません)。それでもあえて2つのテンプレートに分けているのは、S3バケットやIAMロールといった監査ログ基盤はDBインスタンスより長いライフサイクルで使い回したい(インスタンスを作り直しても監査ログやIAMロールは残したい)ことが多いためです。ただしこの疎結合ゆえに、後片付けの際は先にDBインスタンス側のスタックを削除し、オプショングループが「使用中」でなくなってから監査ログ基盤側のスタックを削除するという順番を守る必要があります(逆の順番だと、使用中のオプショングループは削除できずエラーになります)。

ここから先、実際に監査ポリシーを設定してS3への出力を確認する手順は、本編4.4節以降とまったく同じです。RDSADMINデータベースに接続し直してrdsadmin.configure_db_auditを呼び出し、rdsadmin.get_task_statusで成功を確認、1時間ほど待ってS3バケットに.delファイルが出力されることを確認する、という流れになるので、詳しくは本編を参照してください。

2. `EXECUTE WITH DATA`のSQLテキストをPythonで読み解く

もう一つのテーマです。EXECUTEカテゴリをWITH DATA(実行されたSQLの本文まで記録する設定)にした場合、そのSQLテキストがexecute.delにそのまま書かれているわけではなく、auditlobsという別ファイルへのポインタ参照になっている、という挙動が本編で見つかりました。ここでは、この参照を実際に解決して、SQLテキストを読み出すところまでやってみます。

2.1. `auditlobs`の正体

execute.delの中身を見ると、こんな行がありました。

"2026-07-19-10.11.17.310324","EXECUTE","DATA",77,100,"TESTDB","dbadmin","DBADMIN","DBADMIN",,,"::ffff:xxx.xxx.xxx.xxx.49193.260719091409",,,,,,,,,,,,,"      U","                    ",,,,,,,,,,,,1,"VARCHAR ","auditlobs.58720.6/",0,,,,,

末尾近くにある auditlobs.58720.6/ という文字列がポインタです。auditlobs自体をfileコマンドで確認するとdata(構造化されていない生データ)と判定される、いわばバイナリの塊でした。

$ file auditlobs
auditlobs: data

これは監査ログ専用の仕組みではなく、Db2のEXPORTユーティリティが昔からLOB列を扱う際に使っているLOB Location Specifier (LLS) という標準機構がそのまま流用されたものでした。IBM公式ドキュメントに定義があります。

The format of the LLS is lobfilename.ext.nnn.mmm/, where lobfilename.ext is the name of the file that contains the LOB, nnn is the offset of the LOB within the file (measured in bytes), and mmm is the length of the LOB (measured in bytes).

つまり auditlobs.58720.6/ は「auditlobsファイルの58720バイト目から6バイト分を読め」という位置指定です。auditlobsを先頭から素朴に読んでも意味のある区切りは分からず、.del側のポインタとオフセット・長さを突き合わせて初めて1件分のデータが復元できる、という構造でした。

2.2. Pythonでポインタを解決してCSVにする

Db2側の正規の方法はIMPORT ... LOBS FROM ... MODIFIED BY LOBSINFILEでステージングテーブルに読み込むことですが、手元にDb2環境を用意しなくてもサクッと確認したかったので、Pythonで直接ポインタを解決するスクリプトを書きました。

どうせなら、生の.delをそのまま出すのではなく、普段見慣れている監査ログらしいCSVにしたいところです。IBM公式ドキュメントに、EXECUTEカテゴリの各列が何を意味するか、DELファイルへの出力順で列挙されています。

実際にexecute.delの1行のフィールド数を数えてみたところ、このドキュメントに載っている列数(46列)とぴったり一致しました。これで各列の意味を機械的にマッピングできます。

ただ、このEXECUTEカテゴリのレコードはそのままだと少しクセがあります。ドキュメントに明記されている通り、1回のSQL実行につきSTATEMENT行が1つ、それに続くDATA行(バインド変数の値)が0件以上という形で複数行に分かれています。同じ実行に属する行はevent_correlatorという列が同じ値になり、かつファイル内で連続して並ぶので、これを1回のSQL実行=1行に畳み込めば、日時・ログイン(接続元)情報・SQLテキスト・バインド変数・処理行数が横に並んだ、いわゆる普通の監査ログらしいCSVになります。

import csv
import re
import sys

LLS_PATTERN = re.compile(r'^([A-Za-z0-9_]+)\.(\d+)\.(\d+)/$')

# EXECUTEカテゴリの監査レコードの列順(IBM公式ドキュメント準拠、実データの46列と一致確認済み)
# https://www.ibm.com/docs/en/db2/11.5.x?topic=layouts-audit-record-layout-execute-events
EXECUTE_COLUMNS = [
    "timestamp", "category", "audit_event", "event_correlator", "event_status",
    "database_name", "user_id", "authorization_id", "session_authorization_id",
    "origin_node_number", "coordinator_node_number", "application_id",
    "application_name", "client_user_id", "client_accounting_string",
    "client_workstation_name", "client_application_name", "trusted_context_name",
    "connection_trust_type", "role_inherited", "package_schema", "package_name",
    "package_section", "package_version", "local_transaction_id",
    "global_transaction_id", "uow_id", "activity_id", "statement_invocation_id",
    "statement_nesting_level", "activity_type", "statement_text",
    "statement_isolation_level", "compilation_environment_description",
    "rows_modified", "rows_returned", "savepoint_id", "statement_value_index",
    "statement_value_type", "statement_value_data",
    "statement_value_extended_indicator", "local_start_time", "original_user_id",
    "instance_name", "hostname", "tenant_name",
]

# 1回のSQL実行として畳み込んだ後の出力列
FLAT_COLUMNS = [
    "timestamp", "database", "user_id", "authid", "application_id",
    "sql_text", "bind_variables", "rows_modified", "rows_returned",
]

def resolve_lls(value, base_dir):
    """LLS形式(lobfilename.offset.length/)のポインタを実データに解決する"""
    if not value:
        return value
    m = LLS_PATTERN.match(value)
    if not m:
        return value
    lobfile, offset, length = m.group(1), int(m.group(2)), int(m.group(3))
    if length == -1:
        return None  # NULL
    if length == 0:
        return ""
    with open(f"{base_dir}/{lobfile}", "rb") as f:
        f.seek(offset)
        data = f.read(length)
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError:
        return data.hex()  # SQLテキスト以外の内部バイナリ値は16進文字列にして見えるようにする

def iter_clean_lines(del_path):
    """DELファイルには非テキスト値が混じることがあるためNUL等を除去しつつ行単位で読む"""
    with open(del_path, "rb") as f:
        for raw_line in f:
            text = raw_line.decode("utf-8", errors="replace").replace("\x00", "")
            if text.strip():
                yield text

def read_execute_del(del_path, base_dir="."):
    reader = csv.reader(iter_clean_lines(del_path), quotechar='"', doublequote=True)
    for row in reader:
        row = [resolve_lls(v, base_dir) for v in row]
        n = len(EXECUTE_COLUMNS)
        row = (row + [""] * n)[:n]
        yield dict(zip(EXECUTE_COLUMNS, row))

def flatten_executions(rows):
    """STATEMENT行と、それに続くDATA行(バインド変数)を1回の実行=1行に畳み込む。

    STATEMENTとDATAはevent_correlatorが同じでファイル内で連続して並ぶという
    ドキュメント記載の前提に基づく。ただしevent_correlatorはグローバルに一意な
    IDではなく使い回されるため、この結合はファイル内での隣接順に依存している
    (今回のような1セッションのテストなら問題ないが、大量の同時接続がある環境では
    注意が必要)。
    """
    rows = list(rows)
    i = 0
    while i < len(rows):
        r = rows[i]
        if r["audit_event"] == "STATEMENT":
            binds = []
            j = i + 1
            while (j < len(rows)
                   and rows[j]["audit_event"] == "DATA"
                   and rows[j]["event_correlator"] == r["event_correlator"]):
                if rows[j]["statement_value_data"]:
                    binds.append(rows[j]["statement_value_data"])
                j += 1
            yield {
                "timestamp": r["timestamp"],
                "database": r["database_name"],
                "user_id": r["user_id"],
                "authid": r["authorization_id"],
                "application_id": r["application_id"],
                "sql_text": r["statement_text"],
                "bind_variables": ";".join(binds),
                "rows_modified": r["rows_modified"],
                "rows_returned": r["rows_returned"],
            }
            i = j
        else:
            i += 1

if __name__ == "__main__":
    del_file = sys.argv[1]
    base = sys.argv[2] if len(sys.argv) > 2 else "."
    out_path = sys.argv[3] if len(sys.argv) > 3 else "execute_audit.csv"

    rows = list(read_execute_del(del_file, base))
    with open(out_path, "w", encoding="utf-8", newline="") as out:
        writer = csv.DictWriter(out, fieldnames=FLAT_COLUMNS)
        writer.writeheader()
        count = 0
        for exec_row in flatten_executions(rows):
            writer.writerow(exec_row)
            count += 1
    print(f"wrote {count} execution rows to {out_path}")

ポイントは3つです。

DELファイルはDb2内部の値(バイナリ)が一部混ざるため、Pythonのcsvモジュールにそのまま食わせると_csv.Error: line contains NULで落ちます。行ごとにバイト列として読み、NUL文字を除去してからcsv.readerに渡すことで回避しました。
各フィールドがLLSポインタ形式(lobfilename.offset.length/)にマッチしたら、対象ファイルを該当オフセットまでシークして指定バイト数だけ読み出し、文字列として復元します。
STATEMENT行と、直後に連続する同じevent_correlatorのDATA行を1回の実行としてまとめ、バインド変数は;区切りの1列にしています。

2.3. 実際に動かしてみる

execute.delに対して実行すると、732件のSQL実行がCSVとして書き出せました。

$ python3 resolve_lls.py execute.del . execute_audit.csv
wrote 732 execution rows to execute_audit.csv

テストで実行した3つのSQLの部分を抜き出すと、こうなっています。

timestamp	database	user_id	authid	application_id	sql_text	rows_modified	rows_returned
2026-07-19-10.11.54.043345	TESTDB	dbadmin	DBADMIN	::ffff:xxx.xxx.xxx.xxx.60838.260719101136	CREATE TABLE AUDIT_TEST (ID INT, NAME VARCHAR(50))	0	0
2026-07-19-10.11.54.127201	TESTDB	dbadmin	DBADMIN	::ffff:xxx.xxx.xxx.xxx.60838.260719101136	INSERT INTO AUDIT_TEST VALUES (1, 'test')	1	0
2026-07-19-10.11.54.150300	TESTDB	dbadmin	DBADMIN	::ffff:xxx.xxx.xxx.xxx.60838.260719101136	SELECT * FROM AUDIT_TEST	0	1

INSERT文の値(1, 'test')はSQLテキストにリテラルで書いたためbind_variablesは空です。パラメータマーカー(?)を使った実行だとここに値が入ります。実際、DBeaverが裏で発行しているシステムカタログ照会の中に、こんな行がありました。

timestamp	database	user_id	authid	application_id	sql_text	bind_variables	rows_modified	rows_returned
2026-07-19-10.09.32.415205	TESTDB	rdsdb	RDSDB	*LOCAL.rdsdb.260719101002	SELECT FILE, PATH FROM TABLE(SYSPROC.AUDIT_LIST_LOGS(CAST(? AS VARCHAR(1024)))) AS AUDIT_LIST_LOGS WHERE FILE LIKE ?	%TESTDB%	0	1

?に対応する実際の値(%TESTDB%)がbind_variables列にきちんと入っています。EXECUTE WITH DATAで記録された内容が、ポインタを解決することで完全に復元できることが確認できました。

副産物として、DBeaverが接続時に自動発行するシステムカタログ照会も同じ仕組みで大量に記録されていることが分かりました(732件のうち、自分で実行したSQLは数件で、大半はDBeaverやRDS基盤自身が発行したものです)。つまり EXECUTE WITH DATAを有効にすると、自分が意図したSQLだけでなく、GUIクライアントやRDS基盤が裏で自動発行するクエリまで丸ごと記録されるということです。監査ログの量を見積もる際は、この点も考慮に入れる必要がありそうです。

なお、上の表のapplication_id列にはもともと接続元の実IPアドレスが含まれていましたが、掲載にあたってxxx.xxx.xxx.xxxにマスキングしています。実運用でこのCSVを扱う際も、接続元IPは個人情報に該当しうる点は意識しておくとよさそうです。

2.4. ログインイベントを含めて1つのタイムラインにする

ここまではexecute.delだけを見てきましたが、「誰が・いつ・何をしたか」を追うには、ログイン(validate.del、VALIDATEカテゴリ)も一緒に見たいところです。両者は監査カテゴリが違うので別ファイルに分かれていますが、application_idがDb2のセッションID相当であることが分かっているので、これをキーにタイムラインを1本にまとめられます。

validate.delもIBM公式ドキュメントで列定義を確認し、実データの列数(32列)と一致することを確認しました。

また、execute.del側もこれまではSTATEMENT行だけを拾っていましたが、COMMIT/ROLLBACK/CONNECT/CONNECT RESETといった他のイベントもそのまま行として残すようにし、validate.delのログインイベントとタイムスタンプ順にマージします。

application_idには接続元IPも埋め込まれています(::ffff:<IPv4>.<ポート>.<タイムスタンプ>、あるいはローカル接続なら*LOCAL.<名前>.<タイムスタンプ>という形式)。これを正規表現で抜き出してclient_ip列にし、あわせてapplication_name(DBeaverやdb2bpなど、接続元クライアントの種類)もclient_application列として残すようにしました。

なお、この末尾のタイムスタンプ部分は、IBM公式ドキュメントで「approximate timestamp(おおよそのタイムスタンプ)」と明記されており、監査ログのLOGINイベント自体のtimestamp列とは数十秒程度ズレることがあります。実際、今回のテストセッションでも38秒ほどの差がありました。正確な接続時刻が欲しい場合はapplication_id内の値ではなく、監査ログ側のtimestamp列を使うべきです。

execute.delとvalidate.delをまとめて処理する最終版のスクリプト全文です。

import csv
import re
import sys

LLS_PATTERN = re.compile(r'^([A-Za-z0-9_]+)\.(\d+)\.(\d+)/$')
APPID_IPV4_PATTERN = re.compile(r'^::ffff:(\d+\.\d+\.\d+\.\d+)\.(\d+)\.(\d+)$')
APPID_LOCAL_PATTERN = re.compile(r'^\*LOCAL\.([^.]+)\.(\d+)$')

# https://www.ibm.com/docs/en/db2/11.5.x?topic=layouts-audit-record-layout-execute-events
EXECUTE_COLUMNS = [
    "timestamp", "category", "audit_event", "event_correlator", "event_status",
    "database_name", "user_id", "authorization_id", "session_authorization_id",
    "origin_node_number", "coordinator_node_number", "application_id",
    "application_name", "client_user_id", "client_accounting_string",
    "client_workstation_name", "client_application_name", "trusted_context_name",
    "connection_trust_type", "role_inherited", "package_schema", "package_name",
    "package_section", "package_version", "local_transaction_id",
    "global_transaction_id", "uow_id", "activity_id", "statement_invocation_id",
    "statement_nesting_level", "activity_type", "statement_text",
    "statement_isolation_level", "compilation_environment_description",
    "rows_modified", "rows_returned", "savepoint_id", "statement_value_index",
    "statement_value_type", "statement_value_data",
    "statement_value_extended_indicator", "local_start_time", "original_user_id",
    "instance_name", "hostname", "tenant_name",
]

# https://www.ibm.com/docs/en/db2/11.5.x?topic=layouts-audit-record-layout-validate-events
VALIDATE_COLUMNS = [
    "timestamp", "category", "audit_event", "event_correlator", "event_status",
    "database_name", "user_id", "authorization_id", "execution_id",
    "origin_node_number", "coordinator_node_number", "application_id",
    "application_name", "authentication_type", "package_schema", "package_name",
    "package_section_number", "package_version", "plugin_name",
    "local_transaction_id", "global_transaction_id", "client_user_id",
    "client_workstation_name", "client_application_name",
    "client_accounting_string", "trusted_context_name", "connection_trust_type",
    "role_inherited", "original_user_id", "instance_name", "hostname",
    "tenant_name",
]

UNIFIED_COLUMNS = [
    "timestamp", "event", "database", "user_id", "authid", "application_id",
    "client_ip", "client_application", "hostname", "status", "sql_text",
    "bind_variables", "rows_modified", "rows_returned",
]


def parse_client_ip(application_id):
    """application_id(セッションID)に埋め込まれた接続元IPを取り出す"""
    if not application_id:
        return ""
    m = APPID_IPV4_PATTERN.match(application_id)
    if m:
        return m.group(1)
    if APPID_LOCAL_PATTERN.match(application_id):
        return "(local)"
    return ""


def resolve_lls(value, base_dir):
    """LLS形式(lobfilename.offset.length/)のポインタを実データに解決する"""
    if not value:
        return value
    m = LLS_PATTERN.match(value)
    if not m:
        return value
    lobfile, offset, length = m.group(1), int(m.group(2)), int(m.group(3))
    if length in (-1, 0):
        return None if length == -1 else ""
    with open(f"{base_dir}/{lobfile}", "rb") as f:
        f.seek(offset)
        data = f.read(length)
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError:
        return data.hex()


def iter_clean_lines(del_path):
    """DELファイルには非テキスト値が混じることがあるためNUL等を除去しつつ行単位で読む"""
    with open(del_path, "rb") as f:
        for raw_line in f:
            text = raw_line.decode("utf-8", errors="replace").replace("\x00", "")
            if text.strip():
                yield text


def read_del(del_path, columns, base_dir="."):
    reader = csv.reader(iter_clean_lines(del_path), quotechar='"', doublequote=True)
    n = len(columns)
    for row in reader:
        row = [resolve_lls(v, base_dir) for v in row]
        row = (row + [""] * n)[:n]
        yield dict(zip(columns, row))


def common_fields(r):
    return {
        "client_ip": parse_client_ip(r["application_id"]),
        "client_application": r.get("application_name", ""),
    }


def flatten_execute(rows):
    """1イベント=1行。STATEMENT行は直後のDATA行をbind_variablesとして吸収し、
    COMMIT/ROLLBACK/CONNECTなどはそのまま1行として残す。"""
    rows = list(rows)
    i = 0
    while i < len(rows):
        r = rows[i]
        if r["audit_event"] == "STATEMENT":
            binds = []
            j = i + 1
            while (j < len(rows) and rows[j]["audit_event"] == "DATA"
                   and rows[j]["event_correlator"] == r["event_correlator"]):
                if rows[j]["statement_value_data"]:
                    binds.append(rows[j]["statement_value_data"])
                j += 1
            row = {
                "timestamp": r["timestamp"], "event": "SQL_EXECUTE",
                "database": r["database_name"], "user_id": r["user_id"],
                "authid": r["authorization_id"], "application_id": r["application_id"],
                "hostname": r["hostname"], "status": r["event_status"],
                "sql_text": r["statement_text"], "bind_variables": ";".join(binds),
                "rows_modified": r["rows_modified"], "rows_returned": r["rows_returned"],
            }
            row.update(common_fields(r))
            yield row
            i = j
        elif r["audit_event"] == "DATA":
            i += 1  # STATEMENT側で吸収済みなのでスキップ
        else:
            row = {
                "timestamp": r["timestamp"], "event": r["audit_event"],
                "database": r["database_name"], "user_id": r["user_id"],
                "authid": r["authorization_id"], "application_id": r["application_id"],
                "hostname": r["hostname"], "status": r["event_status"],
                "sql_text": "", "bind_variables": "", "rows_modified": "",
                "rows_returned": "",
            }
            row.update(common_fields(r))
            yield row
            i += 1


def flatten_validate(rows):
    for r in rows:
        row = {
            "timestamp": r["timestamp"], "event": "LOGIN",
            "database": r["database_name"], "user_id": r["user_id"],
            "authid": r["authorization_id"], "application_id": r["application_id"],
            "hostname": r["hostname"], "status": r["event_status"],
            "sql_text": "", "bind_variables": "", "rows_modified": "",
            "rows_returned": "",
        }
        row.update(common_fields(r))
        yield row


def main():
    execute_del, validate_del, base, out_path = sys.argv[1:5]
    execute_rows = list(flatten_execute(read_del(execute_del, EXECUTE_COLUMNS, base)))
    validate_rows = list(flatten_validate(read_del(validate_del, VALIDATE_COLUMNS, base)))
    unified = execute_rows + validate_rows
    unified.sort(key=lambda r: r["timestamp"])
    with open(out_path, "w", encoding="utf-8", newline="") as out:
        writer = csv.DictWriter(out, fieldnames=UNIFIED_COLUMNS)
        writer.writeheader()
        writer.writerows(unified)
    print(f"wrote {len(unified)} rows to {out_path}")


if __name__ == "__main__":
    main()

実行方法です。

$ python3 build_unified_audit.py execute.del validate.del . unified_audit.csv
wrote 2466 rows to unified_audit.csv

これを実行して、自分のテストで使ったセッションだけを抜き出すと、こういうタイムラインになりました。

timestamp	event	user_id	client_ip	client_application	sql_text
2026-07-19-10.10.58.540349	LOGIN	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.10.58.545388	CONNECT	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.11.23.521123	SQL_EXECUTE	dbadmin	xxx.xxx.xxx.xxx	DBeaver	SELECT 1 FROM SYSIBM.SYSDUMMY1
2026-07-19-10.11.23.541910	COMMIT	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.11.54.043345	SQL_EXECUTE	dbadmin	xxx.xxx.xxx.xxx	DBeaver	CREATE TABLE AUDIT_TEST (ID INT, NAME VARCHAR(50))
2026-07-19-10.11.54.048172	COMMIT	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.11.54.127201	SQL_EXECUTE	dbadmin	xxx.xxx.xxx.xxx	DBeaver	INSERT INTO AUDIT_TEST VALUES (1, 'test')
2026-07-19-10.11.54.128341	COMMIT	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.11.54.150300	SQL_EXECUTE	dbadmin	xxx.xxx.xxx.xxx	DBeaver	SELECT * FROM AUDIT_TEST
2026-07-19-10.11.54.168657	COMMIT	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.38.17.721661	ROLLBACK	dbadmin	xxx.xxx.xxx.xxx	DBeaver
2026-07-19-10.38.17.721720	CONNECT RESET	dbadmin	xxx.xxx.xxx.xxx	DBeaver

DBeaverでの接続テスト(SELECT 1 FROM SYSIBM.SYSDUMMY1)から始まり、ログインしてから約27分後に切断(CONNECT RESET)している様子まで、ログインからログオフまでを1本のタイムラインとして追えました。まさに「よくある監査ログ」らしい形になったと思います。(client_ipは掲載にあたってマスキングしています)

ここで一つ、想定外の発見がありました。 user_idをdbadminで絞り込むと、application_idが異なる(＝別セッションの)行が他に2つ混ざっていました。しかも、そのうち1つ(ポート49193)はSYSCAT.TABLESやSYSCAT.COLUMNSといったカタログ参照が11件も並んでいるのに、対応するLOGIN/CONNECT行がどこにもありません。

調べてみると、種明かしは前述のapplication_id末尾のタイムスタンプにありました。このセッションでは...091409(9時14分09秒相当)で、本編4.4節で監査カテゴリCONTEXT/EXECUTE/ERRORを有効化した時刻(UTC 10:09)より前です。前述の通りこの値は「approximate」ではあるものの、ズレは数十秒単位であり、1時間近く前というこの差を覆すものではありません。つまり、このセッション自体は監査ログが有効になる前からすでに接続済みだったため、LOGINイベントは記録されず、監査有効化後にそのセッション上で実行されたSQL(SYSCAT系のカタログ参照)だけが記録された、と考えられます。DBeaverはメイン接続とは別に、スキーマツリー表示用のメタデータ専用接続を裏で張ることが知られており、その接続がたまたま監査有効化のタイミングをまたいでいた、というのが実態のようです。

つまり実務上の教訓として、「セッションタイムライン」は必ずしもLOGINから始まるとは限らないということです。今回のように監査を有効化した直後に既存の接続がその境界をまたぐケースは一回限りのものですが、より一般的で日常的に起こりうるのは、監査ログが1時間ごとのバッチ単位でS3に出力される(本編5.2節参照)ことに起因するパターンです。DBeaverのように接続を長時間張りっぱなしにするクライアントや、コネクションプールを使うアプリケーションでは、LOGINは10時台のファイルに記録されていても、そのままSQL実行は11時台・12時台…と複数時間分のファイルにまたがって記録されていきます。1つの.delファイルだけを見ていると、そこにLOGINが存在しないSQL実行は日常的に起こりうる、ということです。application_idごとに集計・分析する処理を書く際は、複数時間分のファイルをまたいで集約する前提で、LOGINが見当たらないケースも起こりうるものとして書く必要があります。

3. まとめ

db2-ceインスタンス本体は、コンソール手順に対応する形でCloudFormationテンプレート化でき、実際にCloudFormationコンソールからのデプロイとDBeaverでの接続確認まで検証済み(License Manager設定はCLI/CFn経由では不要という差異あり)
監査ログ基盤(S3・IAM・オプショングループ)も、本編で見つかったConsoleの制限(db2-ce用オプショングループが新規作成できない)を踏まえ、CloudFormationで一括構築できる。実際にCloudFormationコンソールからのデプロイとDBインスタンスへの適用(再起動不要)まで検証済み
EXECUTE WITH DATAで記録されるSQLテキストや入力値は、.delファイルにインラインで入るのではなく、auditlobsというファイルへLOB Location Specifier(LLS)形式のポインタで参照される
これはDb2監査ログ専用の仕組みではなく、EXPORTコマンドがLOB列を扱う際に使う標準機構(LOBS TO/LOBFILEオプション)がそのまま流用されたもの
正規の読み出し方法はIMPORT ... LOBS FROM ... MODIFIED BY LOBSINFILEでのステージングテーブルへの取り込みだが、簡易的にはPythonでポインタを解決するスクリプトでも十分実用になる
IBM公式ドキュメントのEXECUTEカテゴリ列定義(46列)・VALIDATEカテゴリ列定義(32列)は、どちらも実データの列数と完全に一致しており、これをもとにSTATEMENT行とDATA行を1回の実行=1行に畳み込める
application_idはDb2のセッションID相当で、ログイン(VALIDATE)からSQL実行(EXECUTE)、切断まで同一セッション内は同じ値を持つ。中に接続元IPや接続タイムスタンプも埋め込まれているため、正規表現で取り出せばclient_ip列も作れる。これをキーに、ログイン〜ログオフまでを1本のタイムラインとして統合したCSVを作れる
ただし「セッションは必ずLOGINから始まる」とは限らない。監査ログは1時間ごとのバッチでファイル出力されるため、長時間張りっぱなしの接続はLOGINとSQL実行が別々の時間帯のファイルに分かれて記録される。複数ファイルをまたいで集計しないと、LOGINの見当たらないSQL実行が日常的に出てくる(監査有効化前から接続済みだったケースも実際に確認した)
EXECUTE WITH DATAは、アプリケーションのSQLだけでなくクライアントツールが自動発行するクエリまで拾うため、ログ量の見積もりには注意が必要

監査ログの「設定して終わり」ではなく、実際に出力されたファイルの中身まで追いかけてみると、こうした実装の詳細が見えてきて面白いですね。本稿が何かの参考になれば幸いです。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up