Model Armor ログと Gemini Enterprise ログを関連付ける

このドキュメントでは、Model Armor のサニタイズ ログを Cloud Logging の Gemini Enterprise プラットフォーム ログとトレーススパンに関連付ける方法について説明します。クライアントからエージェントへ（内向き）とエージェントから任意の宛先へ（外向き）のトラフィック フローの関連付けメカニズムについて説明し、トレース生成の前提条件の概要を示します。また、ログ処理パイプラインでこれらのログエントリを結合するための手順とコードサンプルを示します。

Model Armor のトレーススパンまたはログエントリを調査する場合は、リクエストの完全なコンテキストを取得するために、Cloud Logging で対応するレコードを見つける必要がある場合があります。次に例を示します。

• トレーススパンから開始する場合は、エンドユーザーの ID を特定したり、サニタイズの詳細な結果を検査したりする必要がある場合があります。
• Model Armor のサニタイズ ログエントリから開始する場合は、ユーザー ID またはトレース情報に関連付ける必要がある場合があります。

ログの関連付けの仕組み

Model Armor は、Gemini Enterprise の次の通信ポイントでプロンプトとレスポンスをスクリーニングできます。

• クライアントからエージェントへ（内向き）のトラフィック: ユーザーが Gemini Enterprise アシスタントにプロンプトを送信すると、Gemini Enterprise は Model Armor API を直接呼び出します。結果として得られる Model Armor プラットフォーム ログ （SanitizeOperation） には、OpenTelemetry trace または spanId フィールドが直接含まれていません。これらのログをユーザー ID とトレーススパンに関連付けるには、セッション トークンを使用して、セキュリティ情報およびイベント管理（SIEM）またはログ処理パイプラインでログ結合を実行します。

• エージェントから任意の宛先へ（外向き）のトラフィック: エージェントが外部ツール、 Model Context Protocol（MCP）サーバー、または外部の大規模言語モデル（LLM）を呼び出すと、 トラフィックは Agent Gateway と Secure Web Proxy を介してルーティングされます。外向きの呼び出しの場合、OpenTelemetry 計測が有効になっていると、Model Armor の SanitizeOperation ログに trace フィールドと spanId フィールドが直接含まれます。ログを直接フィルタして、トレース スパンをCloud Trace または Agent Registryで表示できます。

関連付けメカニズムの概要

フロー	パスとルーティング	Model Armor ログのトレース	関連付け方法
クライアントからエージェントへ（内向き）	Gemini Enterprise から Model Armor への直接 API 呼び出し	trace と spanId は入力されません。	client_correlation_id と assistToken のセッション トークンを使用したログ結合
エージェントから任意の宛先へ（外向き）	Agent Gateway と Secure Web Proxy を介してルーティング	trace と spanId が入力されます。	trace ID の直接一致とトレーススパンの検査

始める前に

Model Armor ログを Gemini Enterprise ログに関連付ける前に、次の操作を行います。

1. Gemini Enterprise で Model Armor を有効にします。
2. トレース コンテキストを生成し、Gemini Enterprise ログと Model Armor ログでトレースの詳細を表示するには、オブザーバビリティ設定で [OpenTelemetry のトレースとログの計測を有効にする] をオンにします。必要に応じて、[プロンプト入力と回答出力のロギングを有効にする] をオンにします。手順については、オブザーバビリティ 設定を有効にするをご覧ください。
3. 外向きトラフィックのスクリーニングを行う場合は、Agent Gateway で Model Armor を構成します。

必要なロール

ログとトレーススパンの表示と関連付けに必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼します。

• Cloud トレース ユーザー （roles/cloudtrace.user）
• ログビューア （roles/logging.viewer）

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

必要な権限は、カスタム ロールや他の事前定義 ロールから取得することもできます。

必要なその他のロールについては、トレース のアクセス制御と Cloud Logging のアクセス 制御をご覧ください。

クライアントからエージェントへ（内向き）のログを関連付ける

1 回の StreamAssist のやり取りで、Cloud Logging に 3 つの異なるログエントリが生成されます。

• Model Armor のサニタイズ ログ（SanitizeOperation）:
  • モニタリング対象リソース: modelarmor.googleapis.com/SanitizeOperation
  • プロパティ: サニタイズの詳細な判定と安全性の結果（PII の編集、責任ある AI フィルタの一致、プロンプト インジェクションの検出など）が含まれますが、トレース コンテキストやエンドユーザー ID は含まれません。
  • 関連付けキー: labels."modelarmor.googleapis.com/client_correlation_id"
• Gemini Enterprise StreamAssist ログ（consumed_api）:
  • モニタリング対象リソース: consumed_api
  • プロパティ: エンドユーザー ID（userIamPrincipal）、トレースの詳細（trace と spanId）、セッション トークン（response.assistToken）が含まれます。
  • 関連付けキー: jsonPayload.response.assistToken
• Gemini Enterprise ModelArmorAudit ログ（Agent）:
  • モニタリング対象リソース: discoveryengine.googleapis.com/Agent（jsonPayload.logMetadata.methodName が ModelArmorAudit の場合）。
  • プロパティ: サニタイズの概要判定を反映し、トレース コンテキスト（trace と spanId）が含まれますが、詳細な結果や関連付け ID は含まれません。
  • 関連付けキー: trace

関連付け結合キー

Model Armor のサニタイズ ログには、パイプ区切りの構造を持つ client_correlation_id ラベルが含まれています。このラベルの 3 番目のセグメント は、 assistToken ログに記録された consumed_api フィールドと一致する base64url エンコードされたセッション トークンです StreamAssist。

client_correlation_id ラベルの形式は次のとおりです。

AS|ASSISTANT_RESOURCE|SESSION_TOKEN

関連付け ID には次の値が含まれます。

• ASSISTANT_RESOURCE：次の形式の Gemini Enterprise Assistant リソースの完全なリソース名。

  projects/PROJECT/locations/LOCATION/collections/COLLECTION/engines/ENGINE/assistants/ASSISTANT

• SESSION_TOKEN: base64url パディングが正規化された後、consumed_api ログの assistToken と一致する一意のセッション トークン。

マッチングのロジック

Model Armor のサニタイズ ログエントリを Gemini Enterprise StreamAssist ログに関連付けるには、ログ処理パイプラインに次のマッチング ロジックを実装します。

1. Model Armor エントリからセッション トークンを抽出する:

   1. Model Armor エントリで labels オブジェクトを見つけます。
   2. modelarmor.googleapis.com/client_correlation_id ラベルの値を取得します。
   3. パイプ文字（|）を使用して、このラベルの値を分割します。
   4. base64url エンコードされたセッション トークンを表す 3 番目のセグメントを抽出します。

2. StreamAssist エントリから assistToken 値を抽出する : 候補の StreamAssist consumed_api ログエントリごとに、次の操作を行います。

   1. jsonPayload オブジェクトを見つけます。
   2. response.assistToken フィールドからトークン値を抽出します。

3. トークンを正規化して比較する: トークンを比較するには、両方の トークン文字列を正規化します。

   1. すべてのハイフン（-）をプラス記号（+）に置き換えます。
   2. すべてのアンダースコア（_）をスラッシュ（/）に置き換えます。
   3. 末尾の等号（=）を削除します。
   4. 正規化されたトークンが一致する場合は、ログエントリを関連付けます。

4. 関連付けられたデータを抽出する: 一致するものが見つかった場合は、一致するエントリから次のフィールドを抽出します。

   • ユーザー IAM ID: userIamPrincipal フィールド StreamAssist エントリ
   • トレース ID: trace エントリの StreamAssist フィールド
   • スパン ID: spanId エントリの StreamAssist フィールド
   • サニタイズの判定: Model Armor エントリの jsonPayload.sanitizationResultのsanitizationVerdictフィールド

Python の関連付けサンプル

次の Python スクリプトは、Model Armor ログと Gemini Enterprise ログの両方を Cloud Logging にクエリし、トークンの正規化と照合を行い、関連付けられたレコードを出力する方法を示しています。

#!/usr/bin/env python3
from datetime import datetime, timedelta, timezone
from google.cloud import logging

# Google Cloud project ID
PROJECT_ID = "YOUR_PROJECT_ID"


def correlate_logs(ma_entry, de_consumed_entries):
  """Correlates a Model Armor log entry with StreamAssist logs."""
  # 1. Extract client_correlation_id from Model Armor log labels
  labels = ma_entry.get("labels", {})
  client_corr_id = labels.get(
      "modelarmor.googleapis.com/client_correlation_id", ""
  )
  if not client_corr_id:
    return None

  # 2. Extract session token (3rd pipe-delimited segment)
  parts = client_corr_id.split("|")
  if len(parts) < 3:
    return None
  ma_token = parts[2]

  # 3. Normalize base64url padding for comparison
  ma_token_normalized = ma_token.replace("-", "+").replace("_", "/").rstrip("=")

  # 4. Search for matching assistToken in StreamAssist logs
  for de in de_consumed_entries:
    payload = de.get("jsonPayload", {})
    de_token = payload.get("response", {}).get("assistToken", "")
    de_token_normalized = (
        de_token.replace("-", "+").replace("_", "/").rstrip("=")
    )

    if ma_token_normalized == de_token_normalized:
      return {
          "user": payload.get("userIamPrincipal"),
          "trace": de.get("trace"),
          "span_id": de.get("spanId"),
          "verdict": (
              ma_entry.get("jsonPayload", {})
              .get("sanitizationResult", {})
              .get("sanitizationVerdict")
          ),
      }
  return None


def main():
  # Initialize Google Cloud Logging Client
  print(f"Connecting to Google Cloud Logging (Project: {PROJECT_ID})...")
  client = logging.Client(project=PROJECT_ID)

  # Calculate ISO timestamp for 1 hour ago
  one_hour_ago = (
      datetime.now(timezone.utc) - timedelta(hours=1)
  ).strftime("%Y-%m-%dT%H:%M:%SZ")
  print(f"Filtering logs starting from: {one_hour_ago}")

  # Build log query filters
  ma_filter = f"""
    resource.type="modelarmor.googleapis.com/SanitizeOperation"
    AND timestamp >= "{one_hour_ago}"
    """

  de_filter = f"""
    resource.type="consumed_api"
    AND jsonPayload.response.assistToken:*
    AND timestamp >= "{one_hour_ago}"
    """

  # Fetch Model Armor log entries
  print("Fetching Model Armor log entries...")
  ma_entries = [
      entry.to_api_repr()
      for entry in client.list_entries(filter_=ma_filter, max_results=100)
  ]
  print(f"Found {len(ma_entries)} Model Armor entries.")

  # Fetch Gemini Enterprise log entries
  print("Fetching Gemini Enterprise StreamAssist log entries...")
  de_entries = [
      entry.to_api_repr()
      for entry in client.list_entries(filter_=de_filter, max_results=500)
  ]
  print(f"Found {len(de_entries)} Gemini Enterprise entries.")

  # Perform Correlation
  print("\n================ Correlating Logs ================")
  correlated_results = []
  for ma in ma_entries:
    match = correlate_logs(ma, de_entries)
    if match:
      correlated_results.append(match)
      print(f"  User IAM Principal  : {match['user']}")
      print(f"  Sanitization Verdict: {match['verdict']}")
      print(f"  Trace ID            : {match['trace']}")
      print(f"  Span ID             : {match['span_id']}")
      print("-" * 50)

  print(f"\nDone. Total Correlated Records: {len(correlated_results)}")


if __name__ == "__main__":
  main()

エージェントから任意の宛先へ（外向き）のログとトレーススパンを関連付ける

エージェントが Agent Gateway と Model Armor で保護されているツール呼び出し（MCP サーバーや外部 API とのやり取りなど）を実行すると、リクエストはエージェントから任意の宛先へのトラフィックの一部になります。

アプリで OpenTelemetry 計測が有効になっている場合、結果の SanitizeOperation ログエントリには trace フィールドと spanId フィールドが自動的に含まれます。

Cloud Logging で外向きログをフィルタする

Cloud Logging で特定のトレースに関連付けられているすべての Model Armor のサニタイズ ログを見つけるには、次のクエリ フィルタを使用します。

resource.type="modelarmor.googleapis.com/SanitizeOperation"
trace="TRACE_ID"

TRACE_ID は、エージェントのやり取りのトレース ID に置き換えます。

詳細については、ログエントリを表示して分析するをご覧ください。

トレーススパンを表示する

Trace または Agent Registry で、エージェントのやり取りの実行グラフとタイムラインを表示できます。Model Armor は次のスパンを生成します。

• 親スパン: apply_guardrail "Google Cloud Model Armor"
• 子スパン: Request Path と Response Path

各スパンには、ポリシー ID、セキュリティ上の決定、一致したフィルタ違反などの属性が含まれます。詳細については、Model Armor のトレ 101}ーススパンを表示するをご覧ください。

次のステップ

• Gemini Enterprise で Model Armor を有効にする
• Model Armor のトレーススパンを表示する
• Model Armor でロギングを構成する
• Gemini Enterprise でオブザーバビリティ設定を管理する