Skip to content

Commit 448abba

Browse files
vertex-sdk-botcopybara-github
vertex-sdk-bot
authored and
copybara-github
committed
feat: export ADK agent engine template logs to the Telemetry API
PiperOrigin-RevId: 999305085
1 parent 0ac315e commit 448abba

5 files changed

Lines changed: 899 additions & 259 deletions

File tree

‎agentplatform/frameworks/adk.py‎

Lines changed: 218 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
from collections.abc import Awaitable
1818
import enum
1919
import os
20-
import sys
2120
import threading
2221
from typing import (
2322
Any,
@@ -132,6 +131,17 @@
132131
# rather than inherit AuthorizedSession's 120s default.
133132
_TELEMETRY_API_CHECK_TIMEOUT_SECONDS = 5.0
134133

134+
_DEFAULT_TELEMETRY_LOGS_ENDPOINT = "https://telemetry.googleapis.com/v1/logs"
135+
_DEFAULT_MTLS_TELEMETRY_LOGS_ENDPOINT = "https://telemetry.mtls.googleapis.com/v1/logs"
136+
137+
_GCP_LOG_NAME = "gcp.log_name"
138+
_EVENT_NAME = "event.name"
139+
_GCP_RESOURCE_TYPE = "gcp.resource_type"
140+
_LOCATION = "location"
141+
_REASONING_ENGINE_ID = "reasoning_engine_id"
142+
_RESOURCE_CONTAINER = "resource_container"
143+
_SERVICE_VERSION = "service.version"
144+
135145

136146
class _MtlsEndpoint(enum.Enum):
137147
"""Enum for the mTLS endpoint setting."""
@@ -337,8 +347,6 @@ def _default_instrumentor_builder(
337347
)
338348
return None
339349

340-
import os
341-
342350
def _warn_missing_dependency(
343351
package: str,
344352
*,
@@ -356,8 +364,8 @@ def _warn_missing_dependency(
356364
)
357365
MISSING_LOGGING_IMPORT_ERROR_MESSAGE = (
358366
"proceeding with logging disabled because not all packages (i.e."
359-
" `google-cloud-logging`, `opentelemetry-sdk`,"
360-
" `opentelemetry-exporter-gcp-logging`) for tracing have been installed"
367+
" `opentelemetry-sdk`, `opentelemetry-exporter-otlp-proto-http`)"
368+
" for logging have been installed"
361369
)
362370

363371
if needed_for_tracing and enable_tracing:
@@ -366,15 +374,6 @@ def _warn_missing_dependency(
366374
_warn(MISSING_LOGGING_IMPORT_ERROR_MESSAGE)
367375
return None
368376

369-
def _detect_cloud_resource_id(project_id: str) -> Optional[str]:
370-
location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv(
371-
"GOOGLE_CLOUD_LOCATION", ""
372-
)
373-
runtime_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID")
374-
if all(v is not None for v in (location, runtime_id)):
375-
return f"//aiplatform.googleapis.com/projects/{project_id}/locations/{location}/reasoningEngines/{runtime_id}"
376-
return None
377-
378377
try:
379378
import opentelemetry
380379
import opentelemetry.trace
@@ -395,37 +394,13 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]:
395394
"opentelemetry-sdk", needed_for_tracing=True, needed_for_logging=True
396395
)
397396

398-
import uuid
399-
400-
# Provide a set of resource attributes but allow to override them with env
401-
# variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME.
402-
cloud_resource_id = _detect_cloud_resource_id(project_id)
403-
resource = opentelemetry.sdk.resources.Resource.create(
404-
attributes={
405-
"gcp.project_id": project_id,
406-
"cloud.account.id": project_id,
407-
"cloud.provider": "gcp",
408-
"cloud.platform": "gcp.agent_engine",
409-
"service.name": os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", ""),
410-
"service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}",
411-
"cloud.region": (
412-
os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "")
413-
or os.getenv("GOOGLE_CLOUD_LOCATION", "")
414-
),
415-
}
416-
| (
417-
{"cloud.resource_id": cloud_resource_id}
418-
if cloud_resource_id is not None
419-
else {}
420-
)
421-
).merge(opentelemetry.sdk.resources.OTELResourceDetector().detect())
397+
resource = _create_otel_resource(project_id)
422398

423399
if enable_tracing:
424400
try:
425401
import opentelemetry.exporter.otlp.proto.http.version
426402
import opentelemetry.exporter.otlp.proto.http.trace_exporter
427403
import google.auth.transport.requests
428-
from google.cloud.aiplatform import version as aip_version
429404
except (ImportError, AttributeError):
430405
return _warn_missing_dependency(
431406
"opentelemetry-exporter-otlp-proto-http", needed_for_tracing=True
@@ -434,12 +409,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]:
434409
import google.auth
435410

436411
credentials, _ = google.auth.default()
437-
vertex_sdk_version = aip_version.__version__
438-
otlp_http_version = opentelemetry.exporter.otlp.proto.http.version.__version__
439-
user_agent = (
440-
f"Vertex-Agent-Engine/{vertex_sdk_version}"
441-
f" OTel-OTLP-Exporter-Python/{otlp_http_version}"
442-
)
412+
user_agent = _get_user_agent()
443413

444414
session = requests_auth.AuthorizedSession(credentials=credentials)
445415

@@ -499,54 +469,45 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]:
499469

500470
if enable_logging:
501471
try:
502-
import opentelemetry.exporter.cloud_logging
472+
import opentelemetry.exporter.otlp.proto.http._log_exporter
473+
import google.auth.transport.requests
503474
except (ImportError, AttributeError):
504475
return _warn_missing_dependency(
505-
"opentelemetry-exporter-gcp-logging", needed_for_logging=True
476+
"opentelemetry-exporter-otlp-proto-http", needed_for_logging=True
506477
)
507478

508-
class _SimpleLogRecordProcessor(
509-
opentelemetry.sdk._logs.export.SimpleLogRecordProcessor
510-
):
479+
import google.auth
511480

512-
def force_flush(
513-
self, timeout_millis: int = 30000
514-
) -> bool: # pylint: disable=no-self-use
515-
sys.stdout.flush()
516-
sys.stderr.flush()
517-
return True
518-
519-
logger_provider = opentelemetry.sdk._logs.LoggerProvider(resource=resource)
520-
# Use the legacy log processor when experimental semconv is enabled.
521-
# Exporting JSON logs to stdout is bugged; Agent Engine fails to
522-
# correctly parse the `gen_ai.client.inference.operation.details`
523-
# messages.
524-
# TODO: b/480102541 - Unify both branches once the regression is fixed.
525-
if "gen_ai_latest_experimental" in os.getenv(
526-
"OTEL_SEMCONV_STABILITY_OPT_IN", ""
527-
).split(","):
528-
logger_provider.add_log_record_processor(
529-
opentelemetry.sdk._logs.export.BatchLogRecordProcessor(
530-
opentelemetry.exporter.cloud_logging.CloudLoggingExporter(
531-
project_id=project_id,
532-
default_log_name=os.getenv(
533-
"GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine"
534-
),
535-
),
536-
)
481+
credentials, _ = google.auth.default()
482+
session = requests_auth.AuthorizedSession(credentials=credentials)
483+
484+
if _use_client_cert_effective():
485+
client_cert_source = (
486+
mtls.default_client_cert_source()
487+
if mtls.has_default_client_cert_source()
488+
else None
537489
)
490+
session.configure_mtls_channel()
491+
endpoint = _get_logs_api_endpoint(client_cert_source)
538492
else:
539-
logger_provider.add_log_record_processor(
540-
_SimpleLogRecordProcessor(
541-
opentelemetry.exporter.cloud_logging.CloudLoggingExporter(
542-
project_id=project_id,
543-
default_log_name=os.getenv(
544-
"GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine"
545-
),
546-
structured_json_file=sys.stdout,
547-
),
548-
)
493+
endpoint = _DEFAULT_TELEMETRY_LOGS_ENDPOINT
494+
495+
# One processor serves stable and experimental semconv. The stdout
496+
# branch experimental records used to need is gone along with the
497+
# Cloud Logging exporter.
498+
logger_provider = opentelemetry.sdk._logs.LoggerProvider(
499+
resource=_create_otel_resource(project_id)
500+
)
501+
logger_provider.add_log_record_processor(
502+
_gcp_batch_log_record_processor(
503+
opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter(
504+
session=session,
505+
endpoint=endpoint,
506+
headers={"User-Agent": _get_user_agent()},
507+
),
508+
project_id,
549509
)
510+
)
550511

551512
opentelemetry._logs.set_logger_provider(logger_provider=logger_provider)
552513

@@ -670,6 +631,178 @@ def _warn_if_telemetry_api_disabled():
670631
_warn(_TELEMETRY_API_DISABLED_WARNING % (project, project))
671632

672633

634+
def _get_user_agent() -> str:
635+
"""Returns the User-Agent to send on OTLP exports."""
636+
from google.cloud.aiplatform import version as aip_version
637+
638+
user_agent = f"Vertex-Agent-Engine/{aip_version.__version__}"
639+
try:
640+
import opentelemetry.exporter.otlp.proto.http.version
641+
642+
user_agent += (
643+
" OTel-OTLP-Exporter-Python/"
644+
f"{opentelemetry.exporter.otlp.proto.http.version.__version__}"
645+
)
646+
except (ImportError, AttributeError):
647+
pass
648+
return user_agent
649+
650+
651+
def _get_logs_api_endpoint(client_cert_source: bytes | None = None) -> str:
652+
"""Returns the logs endpoint matching _get_api_endpoint's mTLS decision.
653+
654+
Args:
655+
client_cert_source (bytes | None): The client certificate source.
656+
657+
Returns:
658+
str: The logs API endpoint to be used.
659+
"""
660+
if _get_api_endpoint(client_cert_source) == _DEFAULT_MTLS_TELEMETRY_ENDPOINT:
661+
return _DEFAULT_MTLS_TELEMETRY_LOGS_ENDPOINT
662+
return _DEFAULT_TELEMETRY_LOGS_ENDPOINT
663+
664+
665+
def _create_otel_resource(project_id: str):
666+
"""Returns the OTel resource describing the Agent Engine deployment.
667+
668+
Args:
669+
project_id: Project to which to send telemetry.
670+
671+
Returns:
672+
The resource to set on the providers.
673+
"""
674+
import os
675+
import uuid
676+
677+
import opentelemetry.sdk.resources
678+
679+
location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv(
680+
"GOOGLE_CLOUD_LOCATION", ""
681+
)
682+
agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", "")
683+
attributes = {
684+
"gcp.project_id": project_id,
685+
"cloud.account.id": project_id,
686+
"cloud.provider": "gcp",
687+
"cloud.platform": "gcp.agent_engine",
688+
"service.name": agent_engine_id,
689+
"service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}",
690+
"cloud.region": location,
691+
}
692+
if location and agent_engine_id:
693+
attributes["cloud.resource_id"] = (
694+
f"//aiplatform.googleapis.com/projects/{project_id}"
695+
f"/locations/{location}/reasoningEngines/{agent_engine_id}"
696+
)
697+
698+
# Provide a set of resource attributes but allow to override them with env
699+
# variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME.
700+
return opentelemetry.sdk.resources.Resource.create(attributes=attributes).merge(
701+
opentelemetry.sdk.resources.OTELResourceDetector().detect()
702+
)
703+
704+
705+
def _gcp_batch_log_record_processor(exporter, project_id: str):
706+
"""Returns a batch processor that keeps log names and labels stable.
707+
708+
Args:
709+
exporter: The OTLP log exporter to wrap.
710+
project_id (str): Project to which to send telemetry.
711+
712+
Returns:
713+
The configured log record processor.
714+
"""
715+
import copy
716+
import os
717+
718+
import opentelemetry.sdk._logs.export
719+
import opentelemetry.sdk.resources
720+
721+
class GCPBatchLogRecordProcessor(
722+
opentelemetry.sdk._logs.export.BatchLogRecordProcessor
723+
):
724+
"""Keeps Cloud Logging log names and labels stable."""
725+
726+
def __init__(
727+
self,
728+
exporter: opentelemetry.sdk._logs.export.LogRecordExporter,
729+
project_id: str,
730+
):
731+
super().__init__(exporter)
732+
self._log_resource = self._agent_engine_log_resource(project_id)
733+
self._default_log_name = os.getenv(
734+
# OTel logs used to be written to stdout in Agent Engine.
735+
# Let's keep the same log name for backward compatibility.
736+
"GCP_DEFAULT_LOG_NAME",
737+
"aiplatform.googleapis.com/reasoning_engine_stdout",
738+
)
739+
# Resource attributes are dropped once ingested as a
740+
# MonitoredResource, so the version has to travel as a label to
741+
# stay queryable.
742+
self._service_version = os.getenv(
743+
"GOOGLE_CLOUD_AGENT_ENGINE_RUNTIME_REVISION_ID", ""
744+
)
745+
746+
@staticmethod
747+
def _agent_engine_log_resource(project_id: str):
748+
"""Returns the MonitoredResource hints Cloud Logging needs.
749+
750+
These are logs-only: `gcp.resource_type` also steers metric
751+
ingestion, so putting them on the resource traces and metrics
752+
share would move Agent Engine metrics off their monitored
753+
resource.
754+
755+
Args:
756+
project_id (str): Project to which to send telemetry.
757+
758+
Returns:
759+
The resource to merge onto every exported log record.
760+
"""
761+
location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv(
762+
"GOOGLE_CLOUD_LOCATION", ""
763+
)
764+
return opentelemetry.sdk.resources.Resource(
765+
attributes={
766+
# Cloud Logging otherwise detects the resource as
767+
# `generic_task`.
768+
_GCP_RESOURCE_TYPE: "aiplatform.googleapis.com/ReasoningEngine",
769+
_LOCATION: location,
770+
_REASONING_ENGINE_ID: os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", ""),
771+
# Without `projects/`, telemetry.googleapis.com returns a 4xx.
772+
_RESOURCE_CONTAINER: f"projects/{project_id}",
773+
}
774+
)
775+
776+
def on_emit(
777+
self, log_record: opentelemetry.sdk._logs.ReadWriteLogRecord
778+
) -> None:
779+
# The provider hands the same record to every registered
780+
# processor, so the rewrites below go on a copy of our own.
781+
# Shallow is enough: nothing here mutates the body, the
782+
# attributes or the resource in place.
783+
emitted = copy.copy(log_record)
784+
record = emitted.log_record = copy.copy(log_record.log_record)
785+
786+
attributes = dict(record.attributes or {})
787+
if record.event_name:
788+
attributes.setdefault(_EVENT_NAME, record.event_name)
789+
# Cloud Logging derives the log name from `event_name` in
790+
# preference to `gcp.log_name`, which would scatter records
791+
# over one log per event type. The name survives as the
792+
# `event.name` label set above.
793+
record.event_name = None
794+
attributes.setdefault(_GCP_LOG_NAME, self._default_log_name)
795+
emitted.resource = (
796+
log_record.resource or opentelemetry.sdk.resources.Resource.get_empty()
797+
).merge(self._log_resource)
798+
if self._service_version:
799+
attributes.setdefault(_SERVICE_VERSION, self._service_version)
800+
record.attributes = attributes
801+
super().on_emit(emitted)
802+
803+
return GCPBatchLogRecordProcessor(exporter, project_id)
804+
805+
673806
def _get_api_endpoint(client_cert_source: bytes | None = None) -> str:
674807
"""Returns API endpoint based on mTLS configuration and cert availability.
675808

‎setup.py‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,6 @@
151151
reasoning_engine_extra_require = [
152152
"cloudpickle >= 3.0, < 4.0",
153153
"opentelemetry-sdk < 2",
154-
"opentelemetry-exporter-gcp-logging >= 1.11.0a0, < 2.0.0",
155154
"opentelemetry-exporter-otlp-proto-http < 2",
156155
"opentelemetry-instrumentation-google-genai>=0.3b0, <1.0.0",
157156
# TODO(b/538550724): update to stable version of
@@ -165,9 +164,7 @@
165164
agent_engines_extra_require = [
166165
"packaging >= 24.0",
167166
"cloudpickle >= 3.0, < 4.0",
168-
"google-cloud-logging < 4",
169167
"opentelemetry-sdk < 2",
170-
"opentelemetry-exporter-gcp-logging >= 1.11.0a0, < 2.0.0",
171168
"opentelemetry-exporter-otlp-proto-http < 2",
172169
"pydantic >= 2.11.1, < 3",
173170
"typing_extensions",

0 commit comments

Comments
 (0)