将 Model Armor 日志与 Gemini Enterprise 日志相关联

本文档介绍了如何在 Cloud Logging 中将 Model Armor 清理日志与 Gemini Enterprise 平台日志和跟踪记录 span 相关联。它解释了客户端到代理（入站流量）和代理到任意目的地（出站流量）流量的相关机制，概述了生成跟踪记录的前提条件，并提供了分步说明和代码示例，以便在日志处理流水线中联接这些日志条目。

在调查 Model Armor 跟踪记录 span 或日志条目时，您可能需要在 Cloud Logging 中找到相应记录，以获取请求的完整上下文。例如：

• 如果您从跟踪记录 span 开始，可能需要确定最终用户的身份或检查详细的清理结果。
• 如果您从 Model Armor 清理日志条目开始，可能需要将其与用户身份或跟踪记录信息相关联。

日志关联的工作原理

Model Armor 可以在 Gemini Enterprise 中的以下通信点过滤提示和回答：

• 客户端到代理（入站流量）：当用户向 Gemini Enterprise 助理提交提示时，Gemini Enterprise 会直接 调用 Model Armor API。生成的 Model Armor 平台日志 (SanitizeOperation) 不直接包含 OpenTelemetry trace 或 spanId 字段。如需将这些日志与用户身份和跟踪记录 span 相关联，您可以使用会话令牌在安全信息和事件管理 (SIEM) 或日志处理流水线中执行日志联接。

• 智能体到任意目的地（出站流量）：当智能体调用外部工具、Model Context Protocol (MCP) 服务器或外部大语言模型 (LLM) 时，流量会通过 Agent Gateway 和 Secure Web Proxy 路由。对于出站流量调用，当启用 OpenTelemetry 插桩时，Model Armor SanitizeOperation 日志会直接包含 trace 和 spanId 字段。您可以直接过滤日志并在 Cloud Trace 或 Agent Registry中查看跟踪记录 span。

关联机制摘要

Flow	路径和路由	Model Armor 日志中的跟踪记录	关联方法
客户端到代理（入站流量）	从 Gemini Enterprise 到 Model Armor 的直接 API 调用	trace 和 spanId 未填充。	使用 client_correlation_id 和 assistToken 中的会话令牌进行日志联接
代理到任意目的地（出站流量）	通过 Agent Gateway 和 Secure Web Proxy 路由	trace 和 spanId 已填充。	直接匹配 trace ID 和跟踪记录 span 检查

准备工作

在开始将 Model Armor 日志与 Gemini Enterprise 日志相关联之前，请按以下步骤操作：

1. 在 Gemini Enterprise中启用 Model Armor。
2. 如需生成跟踪记录上下文并在 Gemini Enterprise 和 Model Armor 日志中查看跟踪记录详情，请在可观测性设置中开启启用 OpenTelemetry 跟踪记录和日志的插桩 ，并根据需要开启启用提示输入和回答输出的日志记录 。如需查看相关说明，请参阅开启可观测性 设置
3. 对于出站流量过滤，请在 Agent Gateway上配置 Model Armor。

所需的角色

如需获得查看和关联日志和跟踪记录 span 所需的权限，请让管理员为您授予项目的以下 IAM 角色：

• Cloud Trace User (roles/cloudtrace.user)
• Logs Viewer (roles/logging.viewer)

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您也可以通过自定义 角色或其他预定义 角色来获取所需的权限。

如需了解您可能需要的其他角色，请参阅 跟踪记录 访问权限控制 和 Cloud Logging 访问权限 控制。

关联客户端到代理（入站流量）日志

单个 StreamAssist 互动会在 Cloud Logging 中生成三个不同的日志条目：

• Model Armor 清理日志 (SanitizeOperation)：
  • 受监控的资源： modelarmor.googleapis.com/SanitizeOperation
  • 属性： 包含详细的清理结果和安全结果（例如 PII 密文、Responsible AI 过滤器匹配项或提示注入检测），但不包含跟踪记录上下文或最终用户身份。
  • 关联键： labels."modelarmor.googleapis.com/client_correlation_id"
• Gemini Enterprise StreamAssist 日志 (consumed_api)：
  • 受监控的资源： consumed_api
  • 属性： 包含最终用户身份 (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 标签，该标签具有以竖线分隔的结构。此标签的第三段 是 base64url 编码的会话令牌，与记录在 consumed_api 日志中为 StreamAssist的 assistToken字段匹配。

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 编码的会话令牌。

2. 从 StreamAssist 条目中提取 assistToken 值：对于 每个候选 StreamAssist consumed_api 日志条目，请按以下步骤操作：

   1. 找到 jsonPayload 对象。
   2. 从 response.assistToken 字段中提取令牌值。

3. 标准化并比较令牌：如需比较令牌，请标准化两个 令牌字符串：

   1. 将所有连字符 (-) 替换为加号 (+)。
   2. 将所有下划线 (_) 替换为正斜杠 (/)。
   3. 去除所有末尾的等号 (=)。
   4. 如果标准化后的令牌匹配，则关联日志条目。

4. 提取关联的数据：如果找到匹配项，请从匹配的条目中提取以下字段 ：

   • 用户 IAM 身份：userIamPrincipal 字段，来自 StreamAssist 条目
   • 跟踪记录 ID：trace 条目中的 StreamAssist 字段
   • Span ID：StreamAssist 条目中的 spanId 字段
   • 清理结果：Model Armor 条目中 jsonPayload.sanitizationResult下的sanitizationVerdict字段

Python 关联示例

以下 Python 脚本演示了如何查询 Cloud Logging 中的 Model Armor 和 Gemini Enterprise 日志、执行令牌标准化和匹配，以及输出关联的记录：

#!/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()

关联代理到任意目的地（出站流量）日志和跟踪记录 span

当代理执行受 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。

如需了解详情，请参阅查看和分析日志 条目。

查看跟踪记录 span

在 Trace 或 Agent Registry 中，您可以查看代理互动的执行图和时间轴。Model Armor 会生成以下 span：

• 父 span: apply_guardrail "Google Cloud Model Armor"
• 子 span: Request Path 和 Response Path

每个 span 都包含政策 ID、安全决策和匹配的过滤器违规等属性。如需了解详情，请参阅查看 Model Armor 跟踪记录 span。

后续步骤

• 在 Gemini Enterprise 中启用 Model Armor
• 查看 Model Armor 跟踪记录 span
• 在 Model Armor 中配置日志记录
• 管理 Gemini Enterprise 中的可观测性设置