Model Armor 로그와 Gemini Enterprise 로그의 상관관계

이 문서에서는 Cloud Logging에서 Model Armor 정리 로그를 Gemini Enterprise 플랫폼 로그 및 trace 범위와 상호 연결하는 방법을 설명합니다. 클라이언트에서 에이전트로(인그레스) 및 에이전트에서 모든 위치로 (이그레스) 트래픽 흐름의 상호 연결 메커니즘을 설명하고, trace 생성의 사전 요구사항을 간략하게 설명하며, 로그 처리 파이프라인에서 이러한 로그 항목을 조인하기 위한 단계별 안내 및 코드 샘플을 제공합니다.

Model Armor trace 스팬 또는 로그 항목을 조사할 때 요청의 전체 컨텍스트를 가져오려면 Cloud Logging에서 해당 레코드를 찾아야 할 수 있습니다. 예를 들면 다음과 같습니다.

• trace 스팬에서 시작하는 경우 최종 사용자의 ID를 확인하거나 자세한 정리 발견 사항을 검사해야 할 수 있습니다.
• Model Armor 정리 로그 항목에서 시작하는 경우 사용자 ID 또는 trace 정보와 상호 연결해야 할 수 있습니다.

로그 상호 연결 작동 방식

Model Armor는 Gemini Enterprise의 다음 통신 지점에서 프롬프트와 응답을 검사할 수 있습니다.

• 클라이언트에서 에이전트로 (인그레스) 트래픽: 사용자가 Gemini Enterprise 어시스턴트에 프롬프트를 제출하면 Gemini Enterprise는 Model Armor API를 직접 호출합니다. 결과 Model Armor 플랫폼 로그 (SanitizeOperation) 에는 OpenTelemetry trace 또는 spanId 필드가 직접 포함되지 않습니다. 이러한 로그를 사용자 ID 및 trace 범위와 상호 연결하려면 세션 토큰을 사용하여 보안 정보 및 이벤트 관리 (SIEM) 또는 로그 처리 파이프라인에서 로그 조인 을 수행합니다.

• 에이전트에서 모든 위치로 (이그레스) 트래픽: 에이전트가 외부 도구, 모델 컨텍스트 프로토콜 (MCP) 서버 또는 외부 대규모 언어 모델 (LLM)을 호출하면 트래픽이 Agent Gateway 및 Secure Web Proxy를 통해 라우팅됩니다. 이그레스 호출의 경우 OpenTelemetry 계측이 사용 설정되면 Model Armor SanitizeOperation 로그에 trace 및 spanId 필드가 직접 포함됩니다. Cloud Trace 또는 Agent Registry에서 로그를 직접 필터링하고 trace 스팬을 볼 수 있습니다.

상호 연결 메커니즘 요약

Flow	경로 및 라우팅	Model Armor 로그의 trace	상호 연결 방법
클라이언트에서 에이전트로 (인그레스)	Gemini Enterprise에서 Model Armor로 직접 API 호출	trace 및 spanId가 채워지지 않습니다.	client_correlation_id 및 assistToken의 세션 토큰을 사용하여 로그 조인
에이전트에서 모든 위치로 (이그레스)	Agent Gateway 및 Secure Web Proxy를 통해 라우팅됨	trace 및 spanId가 채워집니다.	trace ID에 대한 직접 일치 및 trace 스팬 검사

시작하기 전에

Model Armor 로그를 Gemini Enterprise 로그와 상호 연결하기 전에 다음 단계를 따르세요.

1. Gemini Enterprise 에서 Model Armor를 사용 설정합니다.
2. trace 컨텍스트를 생성하고 Gemini Enterprise 및 Model Armor 로그에서 trace 세부정보를 보려면 관측 가능성 설정에서 OpenTelemetry trace 및 로그의 계측 사용 설정 을 사용 설정하고 필요에 따라 프롬프트 입력 및 대답 출력 로깅 사용 설정 을 사용 설정합니다. 자세한 내용은 모니터링 가능성 설정 사용 설정을 참조하세요.
3. 이그레스 트래픽 검사를 위해 에이전트 게이트웨이에서 Model Armor를 구성합니다.

필요한 역할

로그 및 trace 범위를 보고 상호 연결하는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 다음 IAM 역할을 부여해 달라고 요청하세요.

• Cloud Trace 사용자 (roles/cloudtrace.user)
• 로그 뷰어 (roles/logging.viewer)

역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

커스텀 역할이나 다른 사전 정의된 역할을 통해 필요한 권한을 얻을 수도 있습니다.

필요할 수 있는 다른 역할에 대한 자세한 내용은 Trace 액세스 제어 및 Cloud Logging 액세스 제어를 참조하세요.

클라이언트에서 에이전트로 (인그레스) 로그 상호 연결

단일 StreamAssist 상호작용은 Cloud Logging에서 세 가지 고유한 로그 항목을 생성합니다.

• Model Armor 정리 로그 (SanitizeOperation):
  • 모니터링 리소스: modelarmor.googleapis.com/SanitizeOperation
  • 속성: 자세한 정리 결과 및 안전 결과 (예: PII 수정, 책임감 있는 AI 필터 일치 또는 프롬프트 인젝션 감지)를 포함하지만 trace 컨텍스트 또는 최종 사용자 ID는 포함하지 않습니다.
  • 상호 연결 키: labels."modelarmor.googleapis.com/client_correlation_id"
• Gemini Enterprise StreamAssist 로그 (consumed_api):
  • 모니터링 리소스: consumed_api
  • 속성: 최종 사용자 ID (userIamPrincipal), trace 세부정보 (trace 및 spanId), 세션 토큰(response.assistToken)을 포함합니다.
  • 상호 연결 키: jsonPayload.response.assistToken
• Gemini Enterprise ModelArmorAudit 로그 (Agent):
  • 모니터링 리소스: jsonPayload.logMetadata.methodName이 ModelArmorAudit인 discoveryengine.googleapis.com/Agent
  • 속성: 상위 수준 정리 결과를 반영하고 trace 컨텍스트 (trace 및 spanId)를 포함하지만 자세한 결과 또는 상호 연결 ID는 포함하지 않습니다.
  • 상호 연결 키: trace

상호 연결 조인 키

Model Armor 정리 로그에는 파이프로 구분된 구조가 있는 client_correlation_id 라벨이 포함됩니다. 이 라벨의 세 번째 세그먼트 는 assistToken 필드와 일치하는 base64url로 인코딩된 세션 토큰입니다. consumed_api 로그에 기록된 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로 인코딩된 세션 토큰을 나타내는 세 번째 세그먼트를 추출합니다.

2. StreamAssist 항목에서 assistToken 값 추출: 각 후보 StreamAssist consumed_api 로그 항목에 대해 다음 단계를 따르세요.

   1. jsonPayload 객체를 찾습니다.
   2. response.assistToken 필드에서 토큰 값을 추출합니다.

3. 토큰 정규화 및 비교: 토큰을 비교하려면 두 토큰 문자열을 모두 정규화합니다.

   1. 모든 하이픈 (-)을 더하기 기호 (+)로 바꿉니다.
   2. 모든 밑줄 (_)을 슬래시 (/)로 바꿉니다.
   3. 후행 등호 (=)를 삭제합니다.
   4. 정규화된 토큰이 일치하면 로그 항목을 상호 연결합니다.

4. 상호 연결된 데이터 추출: 일치하는 항목이 있으면 일치하는 항목에서 다음 필드 를 추출합니다.

   • 사용자 IAM ID: userIamPrincipal 필드 StreamAssist 항목의
   • trace ID: StreamAssist 항목의 trace 필드
   • 스팬 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()

에이전트에서 모든 위치로 (이그레스) 로그 및 trace 범위 상호 연결

에이전트가 Agent Gateway 및 Model Armor로 보호되는 도구 호출 (예: MCP 서버 또는 외부 API와 상호작용)을 실행하면 요청이 에이전트에서 모든 위치로 트래픽의 일부가 됩니다.

앱에서 OpenTelemetry 계측이 사용 설정되면 결과 SanitizeOperation 로그 항목에 trace 및 spanId 필드가 자동으로 포함됩니다.

Cloud Logging에서 이그레스 로그 필터링

Cloud Logging에서 특정 trace와 연결된 모든 Model Armor 정리 로그를 찾으려면 다음 쿼리 필터를 사용합니다.

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

TRACE_ID를 에이전트 상호작용의 trace ID로 바꿉니다.

자세한 내용은 로그 항목 보기 및 분석 을 참조하세요.

trace 범위 보기

Trace 또는 Agent Registry에서 에이전트 상호작용의 실행 그래프와 타임라인을 볼 수 있습니다. Model Armor는 다음 범위를 생성합니다.

• 상위 스팬: apply_guardrail "Google Cloud Model Armor"
• 하위 범위: Request Path 및 Response Path

각 스팬에는 정책 ID, 보안 결정, 일치하는 필터 위반과 같은 속성이 포함됩니다. 자세한 내용은 Model Armor trace 범위 보기를 참조하세요.

다음 단계

• Gemini Enterprise에서 Model Armor 사용 설정
• Model Armor trace 범위 보기
• Model Armor에서 로깅 구성
• Gemini Enterprise에서 관측 가능성 설정 관리