Cloud Logging シンク作成前のログ転送は?ーー過去ログの扱いを検証してみた
はじめに
こんにちは。クラウドエース株式会社のきょくです。
最近、業務の中で Google Cloud Logging のログデータを BigQuery に転送して分析する作業を行っております。
その中で、「ログシンクを作成する前のログはどう転送するのか」という点で少しハマったので、本記事はその検証内容と解決方法を共有します。
ログシンクとは?
ログシンクは Cloud Logging の「ログルーター」に対して、どのログをどこに転送するかを定義する仕組みです。
少しわかりづらいかもしれないですが、順を追って説明していきます。
Cloud Logging では、ログの収集、受信、表示だけでなく、さまざまなサポートされている宛先へルーティング・転送することができます。
サポートされている転送先は下記の通りです。
- 別の Google Cloud プロジェクト
- ログバケット
- BigQuery データセット
- Cloud Storage バケット
- Pub/Sub トピック
参考:
ログルーターとログシンクの関係
Cloud Logging の内部には「ログルーター(Log Router)」というログルーティングの仕組みがあります。
ログルーターは受信したログエントリを一時的に保存し、「どのログをどこに転送するか」を判断して転送先にログを転送します。
この転送ルールを定義するのが「ログシンク(Log Sink)」です。
つまり、
- ログルーター:実際に転送を行う仕組み
- ログシンク:転送ルールを設定する仕組み
例えば、BigQuery に対する操作を記録する監査ログを BigQuery のデータセットに転送したい場合、シンクで「BigQuery の監査ログ」をフィルタし、転送先として BigQuery データセットを指定します。
では、早速シンクを作成していきましょう!
ログシンクにて BigQuery の監査ログを BigQuery に転送する
繰り返しになりますが、今回の要件は下記の通りです。
- 対象ログ:BigQuery の監査ログ
- 転送先:BigQuery データセット
Google Cloud コンソールからシンクを作成する手順は以下の通りです。
- 「ロギング」>「ログルーター」画面にアクセス

左側のナビゲーションメニューに「ロギング」が表示されない場合は、画面上部の検索バーで「ロギング」を検索してアクセスできます。ロギング画面に遷移後、左側ナビゲーションメニューの「ログルーター」をクリックしてください。 - 画面上方の「シンクを作成」をクリック
- 「シンクの詳細」セクションに名前と説明を入力
- 「シンクの宛先」セクションに転送先「BigQuery データセット」を選択し、転送したいデータセットを選択
- 「シンクに含めるログの選択」セクションにフィルタを入力
例えば、リソース BigQuery の監査ログを転送する場合※logName="projects/<PROJECT_ID>/logs/cloudaudit.googleapis.com%2Fdata_access" protoPayload.serviceName="bigquery.googleapis.com"<PROJECT_ID>は Google Cloud プロジェクト ID に置き換えてください。 - 「シンクを作成」をクリック
参考:
作成後、指定した BigQuery のデータセットに cloudaudit_googleapis_com_data_accessというテーブルが自動生成されていれば OK です。
※ 初期転送には数分かかります。
ログシンク作成前のログはどう扱う?
ここが本題です。
生成された cloudaudit_googleapis_com_data_access テーブルのログデータを確認すると、シンク作成時点以後のログしか存在しません。
公式ドキュメントにも以下の記載があります。
ログシンクでは、過去に遡ってログエントリを転送できません。
つまり、ログシンクは、シンクが作成される前に受信したログエントリを転送できません。
参考:
つまり、ログシンク作成前の過去ログを扱うには別の方法が必要ですね。
調べたところ、過去ログを手動で Cloud Storage へコピーする方法がありました。
参考:
手順は下記のとおりです。
-
Cloud Storage に過去ログ保存用バケットを作成
-
gcloudコマンドでログをコピーgcloud logging copy <BUCKET_ID> storage.googleapis.com/<CLOUD_STORAGE_BUCKET_NAME> \ --location=global \ --log-filter='<FILTER>' \ --project=<PROJECT_ID>各パラメータの意味:
-
<BUCKET_ID>
コピー元ログバケット(データアクセス監査ログはデフォルトで_Defaultログバケットに保存されます)。 -
<CLOUD_STORAGE_BUCKET_NAME>
コピー先 Cloud Storage バケット名(手順 1. で作成された Cloud Storage バケットの名前に置き換えます)。 -
<FILTER>
例えば、2025 年 10 月 1 日のみの BigQuery の監査ログを取得したい場合、ログを定義するフィルタは下記の通りです。※ 日本標準時 JST と協定世界時 UTC の時差に注意してください。logName="projects/<PROJECT_ID>/logs/cloudaudit.googleapis.com%2Fdata_access" protoPayload.serviceName="bigquery.googleapis.com" timestamp >= "2025-09-30T15:00:00Z" AND timestamp <= "2025-10-01T15:00:00Z"
(例)
2025-10-01 00:00:00 JST = 2025-09-30 15:00:00 UTC
2025-10-02 00:00:00 JST = 2025-10-01 15:00:00 UTC -
<PROJECT_ID>
Google Cloud プロジェクト ID に置き換えます。
このコマンドは長時間実行されるオペレーションを作成し、完了まで数時間かかります。
-
コマンド実行後、Cloud Storage 上に以下のような構造でオブジェクトが作成されます。
CLOUD_STORAGE_BUCKET_NAME/
└ cloudaudit.googleapis.com/
└ data_access/
└ YYYY/
└ MM/
└ DD/
└ *.json
Cloud Storage 上の過去ログを BigQuery に統合する
ここまでの状況を整理します。
| 種類 | 保存場所 | 形式 |
|---|---|---|
| シンク作成前ログ | Cloud Storage | JSON |
| シンク作成後ログ | BigQuery | テーブル形式 |
ここで目指すのは、過去ログを BigQuery にロードして、既存テーブルに統合することです。
ハマりポイント
過去ログを BigQuery に取り込もうとした際に、いくつかの問題に遭遇しました。
試したことと失敗内容は以下のとおりです。
-
Cloud Storage の JSON ログを BigQuery の外部テーブルとして読み込む
- 操作:BigQuery で新規テーブルを作成し、外部ソースとして Cloud Storage の JSON ログを指定し、スキーマを「自動検出」で指定
- 結果:下記エラーが発生
Invalid field name "@type". Fields must contain the allowed characters, and be at most 300 characters long. - 原因:BigQuery では、
@typeフィールド名が使用できない
参考:
-
Cloud Storage の JSON ログを直接 BigQuery にロード
- 操作:
bqコマンドで Cloud Storage の JSON ログを直接 BigQuery のシンク転送用テーブルにロード - 結果:一部フィールドのログが欠損することを発見
- 原因:Cloud Logging から BigQuery へ転送時に特殊な命名規則が使われている
Cloud Storage にある JSON ログと、BigQuery にエクスポートされた監査ログの構造が微妙に違います。
参考: - 操作:
解決方法
-
既存のシンク転送用 BigQuery テーブルのスキーマをコピーする

-
スキーマ内のフィールド名を Cloud Storage 側ログ形式に合わせて修正する
BigQuery 側 Cloud Storage 側 protopayload_auditlogprotoPayloadprotopayload_auditlog_servicedata_v1_bigqueryprotoPayload.serviceData -
修正済みスキーマで、Cloud Storage 上の過去ログを格納するための BigQuery テーブルを新規作成する
- ソース
- Google Cloud Storage
- バケット:
<CLOUD_STORAGE_BUCKET_NAME>/cloudaudit.googleapis.com/data_access/YYYY/MM/*.json - ファイル形式:JSONL(改行区切りJSON)
- テーブル名:
cloudaudit_googleapis_com_data_access_before_sink - スキーマ:修正済みスキーマ
- 詳細オプション:「不明な値」をチェック
- 「テーブルを作成」を押す
- ソース
-
既存のシンク転送テーブルに対して以下のクエリで統合する
INSERT INTO
`<BIGQUERY_DATASET_NAME>.cloudaudit_googleapis_com_data_access` (
logName,
resource,
protopayload_auditlog,
textPayload,
timestamp,
receiveTimestamp,
severity,
insertId,
httpRequest,
operation,
trace,
spanId,
traceSampled,
sourceLocation,
split,
errorGroups,
apphub,
apphubDestination
)
SELECT
logName,
resource,
STRUCT(
protoPayload.serviceName AS serviceName,
protoPayload.methodName AS methodName,
protoPayload.resourceName AS resourceName,
protoPayload.resourceLocation AS resourceLocation,
protoPayload.numResponseItems AS numResponseItems,
protoPayload.status AS status,
protoPayload.authenticationInfo AS authenticationInfo,
protoPayload.authorizationInfo AS authorizationInfo,
protoPayload.policyViolationInfo AS policyViolationInfo,
protoPayload.requestMetadata AS requestMetadata,
protoPayload.serviceData AS servicedata_v1_bigquery,
protoPayload.metadataJson AS metadataJson
) AS protopayload_auditlog,
textPayload,
timestamp,
receiveTimestamp,
severity,
insertId,
httpRequest,
operation,
trace,
spanId,
traceSampled,
sourceLocation,
split,
errorGroups,
apphub,
apphubDestination
FROM
`<BIGQUERY_DATASET_NAME>.cloudaudit_googleapis_com_data_access_before_sink`
まとめ
- Cloud Logging のシンクでは、作成前のログは転送されません。
- 過去ログは
gcloud logging copyコマンドで Cloud Storage にコピーが可能です。 - Cloud Storage に保存されたログは、スキーマ調整後に BigQuery に手動ロードすれば統合できます。
Discussion