From affb6bba28fa6fe875078207daa209d11cf5bdec Mon Sep 17 00:00:00 2001 From: A Vertex SDK engineer Date: Tue, 18 Aug 2026 08:12:59 -0700 Subject: [PATCH] feat: export ADK agent engine template logs to the Telemetry API CloudLoggingExporter is deprecated, so the ADK agent engine templates now send logs over OTLP to telemetry.googleapis.com, alongside the traces they already export there. Applied to all four copies of the template (vertexai/agent_engines, agentplatform/agent_engines, vertexai/preview/reasoning_engines, agentplatform/private/frameworks) so their telemetry does not diverge. This collapses the semconv split. Experimental semconv used a batching processor writing through the Cloud Logging API, while stable semconv wrote structured JSON to stdout, because Agent Engine mis-parsed `gen_ai.client.inference.operation.details` records off stdout. Neither branch survives the move to OTLP -- there is one path now, and it behaves the same under both settings. Because the OTLP mapping is done server side, two behaviours had to be reproduced explicitly: - `_create_otel_resource(project_id, "logs")` pins `gcp.resource_type` and the location/reasoning_engine_id labels, so entries keep landing on `aiplatform.googleapis.com/ReasoningEngine` instead of being detected as `generic_task`. It returns a logs-only resource: the metrics pipeline reads `gcp.resource_type` too, and would move Agent Engine metrics off `prometheus_target` if it saw it. - `_named_batch_log_record_processor` keeps GCP_DEFAULT_LOG_NAME (and the `adk-on-agent-engine` default) working, and re-publishes the record's event name as an `event.name` attribute so it still lands as a log entry label. The User-Agent string is now built by `_get_user_agent` rather than inline in the tracing branch, so both signals report the same one. Nothing imports `opentelemetry-exporter-gcp-logging` or `google-cloud-logging` any more -- the ADK templates were their only users, and the other templates never set up logging -- so both are dropped from the `agent_engines` and `reasoning_engine` extras, and the missing-dependency warning now names the packages logging actually needs. PiperOrigin-RevId: 966602473 --- agentplatform/agent_engines/templates/adk.py | 230 +++++++++++------- setup.py | 3 - .../test_agent_engine_templates_adk.py | 131 +++++++++- vertexai/agent_engines/templates/adk.py | 230 +++++++++++------- .../reasoning_engines/templates/adk.py | 230 +++++++++++------- 5 files changed, 574 insertions(+), 250 deletions(-) diff --git a/agentplatform/agent_engines/templates/adk.py b/agentplatform/agent_engines/templates/adk.py index 9eaa49ce46..849b6eb587 100644 --- a/agentplatform/agent_engines/templates/adk.py +++ b/agentplatform/agent_engines/templates/adk.py @@ -18,7 +18,6 @@ import enum import os import queue -import sys import threading from typing import ( Any, @@ -133,6 +132,14 @@ # rather than inherit AuthorizedSession's 120s default. _TELEMETRY_API_CHECK_TIMEOUT_SECONDS = 5.0 +_DEFAULT_TELEMETRY_LOGS_ENDPOINT = "https://telemetry.googleapis.com/v1/logs" + +_GCP_LOG_NAME = "gcp.log_name" +_EVENT_NAME = "event.name" +_GCP_RESOURCE_TYPE = "gcp.resource_type" +_LOCATION = "location" +_REASONING_ENGINE_ID = "reasoning_engine_id" + class _MtlsEndpoint(enum.Enum): """Enum for the mTLS endpoint setting.""" @@ -357,8 +364,8 @@ def _warn_missing_dependency( ) MISSING_LOGGING_IMPORT_ERROR_MESSAGE = ( "proceeding with logging disabled because not all packages (i.e." - " `google-cloud-logging`, `opentelemetry-sdk`," - " `opentelemetry-exporter-gcp-logging`) for tracing have been installed" + " `opentelemetry-sdk`, `opentelemetry-exporter-otlp-proto-http`)" + " for logging have been installed" ) if needed_for_tracing and enable_tracing: @@ -367,15 +374,6 @@ def _warn_missing_dependency( _warn(MISSING_LOGGING_IMPORT_ERROR_MESSAGE) return None - def _detect_cloud_resource_id(project_id: str) -> Optional[str]: - location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( - "GOOGLE_CLOUD_LOCATION", "" - ) - agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID") - if all(v is not None for v in (location, agent_engine_id)): - return f"//aiplatform.googleapis.com/projects/{project_id}/locations/{location}/reasoningEngines/{agent_engine_id}" - return None - try: import opentelemetry import opentelemetry.trace @@ -396,30 +394,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: "opentelemetry-sdk", needed_for_tracing=True, needed_for_logging=True ) - import uuid - - # Provide a set of resource attributes but allow to override them with env - # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. - cloud_resource_id = _detect_cloud_resource_id(project_id) - resource = opentelemetry.sdk.resources.Resource.create( - attributes={ - "gcp.project_id": project_id, - "cloud.account.id": project_id, - "cloud.provider": "gcp", - "cloud.platform": "gcp.agent_engine", - "service.name": os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", ""), - "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", - "cloud.region": ( - os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") - or os.getenv("GOOGLE_CLOUD_LOCATION", "") - ), - } - | ( - {"cloud.resource_id": cloud_resource_id} - if cloud_resource_id is not None - else {} - ) - ).merge(opentelemetry.sdk.resources.OTELResourceDetector().detect()) + resource = _create_otel_resource(project_id) if enable_tracing: try: @@ -435,12 +410,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: import google.auth credentials, _ = google.auth.default() - vertex_sdk_version = aip_version.__version__ - otlp_http_version = opentelemetry.exporter.otlp.proto.http.version.__version__ - user_agent = ( - f"Vertex-Agent-Engine/{vertex_sdk_version}" - f" OTel-OTLP-Exporter-Python/{otlp_http_version}" - ) + user_agent = _get_user_agent() session = requests_auth.AuthorizedSession(credentials=credentials) @@ -500,54 +470,47 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: if enable_logging: try: - import opentelemetry.exporter.cloud_logging + import opentelemetry.exporter.otlp.proto.http._log_exporter + import google.auth.transport.requests except (ImportError, AttributeError): return _warn_missing_dependency( - "opentelemetry-exporter-gcp-logging", needed_for_logging=True + "opentelemetry-exporter-otlp-proto-http", needed_for_logging=True ) - class _SimpleLogRecordProcessor( - opentelemetry.sdk._logs.export.SimpleLogRecordProcessor - ): + import google.auth - def force_flush( - self, timeout_millis: int = 30000 - ) -> bool: # pylint: disable=no-self-use - sys.stdout.flush() - sys.stderr.flush() - return True - - logger_provider = opentelemetry.sdk._logs.LoggerProvider(resource=resource) - # Use the legacy log processor when experimental semconv is enabled. - # Exporting JSON logs to stdout is bugged; Agent Engine fails to - # correctly parse the `gen_ai.client.inference.operation.details` - # messages. - # TODO: b/480102541 - Unify both branches once the regression is fixed. - if "gen_ai_latest_experimental" in os.getenv( - "OTEL_SEMCONV_STABILITY_OPT_IN", "" - ).split(","): - logger_provider.add_log_record_processor( - opentelemetry.sdk._logs.export.BatchLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - ), - ) + credentials, _ = google.auth.default() + session = requests_auth.AuthorizedSession(credentials=credentials) + + if _use_client_cert_effective(): + client_cert_source = ( + mtls.default_client_cert_source() + if mtls.has_default_client_cert_source() + else None ) + session.configure_mtls_channel() + endpoint = _get_logs_api_endpoint(client_cert_source) else: - logger_provider.add_log_record_processor( - _SimpleLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - structured_json_file=sys.stdout, - ), - ) + endpoint = _DEFAULT_TELEMETRY_LOGS_ENDPOINT + + # One processor serves stable and experimental semconv. The stdout + # branch experimental records used to need is gone along with the + # Cloud Logging exporter (b/480102541). + logger_provider = opentelemetry.sdk._logs.LoggerProvider( + resource=_create_otel_resource(project_id, "logs") + ) + logger_provider.add_log_record_processor( + _named_batch_log_record_processor( + opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter( + session=session, + endpoint=endpoint, + headers={"User-Agent": _get_user_agent()}, + ), + default_log_name=os.getenv( + "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" + ), ) + ) opentelemetry._logs.set_logger_provider(logger_provider=logger_provider) @@ -637,6 +600,109 @@ def _warn_if_telemetry_api_disabled(): _warn(_TELEMETRY_API_DISABLED_WARNING % (project, project)) +def _get_user_agent() -> str: + """Returns the User-Agent to send on OTLP exports.""" + from google.cloud.aiplatform import version as aip_version + + user_agent = f"Vertex-Agent-Engine/{aip_version.__version__}" + try: + import opentelemetry.exporter.otlp.proto.http.version + + user_agent += ( + " OTel-OTLP-Exporter-Python/" + f"{opentelemetry.exporter.otlp.proto.http.version.__version__}" + ) + except (ImportError, AttributeError): + pass + return user_agent + + +def _get_logs_api_endpoint(client_cert_source: bytes | None = None) -> str: + """Returns the logs endpoint matching _get_api_endpoint's mTLS decision. + + Args: + client_cert_source (bytes | None): The client certificate source. + + Returns: + str: The logs API endpoint to be used. + """ + return _get_api_endpoint(client_cert_source).replace("/v1/traces", "/v1/logs") + + +def _create_otel_resource(project_id: str, for_signal: str = "unspecified"): + """Returns the OTel resource describing the Agent Engine deployment. + + Args: + project_id: Project to which to send telemetry. + for_signal: The signal the resource is for. `logs` adds the + MonitoredResource hints Cloud Logging needs, which must not be set + on the resource traces and metrics share. + + Returns: + The resource to set on the provider for `for_signal`. + """ + import os + import uuid + + import opentelemetry.sdk.resources + + location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( + "GOOGLE_CLOUD_LOCATION", "" + ) + agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", "") + attributes = { + "gcp.project_id": project_id, + "cloud.account.id": project_id, + "cloud.provider": "gcp", + "cloud.platform": "gcp.agent_engine", + "service.name": agent_engine_id, + "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", + "cloud.region": location, + } + if location and agent_engine_id: + attributes["cloud.resource_id"] = ( + f"//aiplatform.googleapis.com/projects/{project_id}" + f"/locations/{location}/reasoningEngines/{agent_engine_id}" + ) + if for_signal == "logs": + # Cloud Logging otherwise detects resource as `generic_task` + attributes[_GCP_RESOURCE_TYPE] = "aiplatform.googleapis.com/ReasoningEngine" + attributes[_LOCATION] = location + attributes[_REASONING_ENGINE_ID] = agent_engine_id + + # Provide a set of resource attributes but allow to override them with env + # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. + return opentelemetry.sdk.resources.Resource.create(attributes=attributes).merge( + opentelemetry.sdk.resources.OTELResourceDetector().detect() + ) + + +def _named_batch_log_record_processor(exporter, *, default_log_name: str): + """Returns a batch processor that keeps log names and labels stable. + + Args: + exporter: The OTLP log exporter to wrap. + default_log_name (str): Log name for records that carry none. + + Returns: + The configured log record processor. + """ + import opentelemetry.sdk._logs.export + + class _Processor(opentelemetry.sdk._logs.export.BatchLogRecordProcessor): + def on_emit(self, log_record) -> None: + record = log_record.log_record + attributes = dict(record.attributes or {}) + if record.event_name: + attributes.setdefault(_EVENT_NAME, record.event_name) + elif _GCP_LOG_NAME not in attributes: + attributes[_GCP_LOG_NAME] = default_log_name + record.attributes = attributes + super().on_emit(log_record) + + return _Processor(exporter) + + def _get_api_endpoint(client_cert_source: bytes | None = None) -> str: """Returns API endpoint based on mTLS configuration and cert availability. diff --git a/setup.py b/setup.py index 6c65920a57..9f36a4333b 100644 --- a/setup.py +++ b/setup.py @@ -151,7 +151,6 @@ reasoning_engine_extra_require = [ "cloudpickle >= 3.0, < 4.0", "opentelemetry-sdk < 2", - "opentelemetry-exporter-gcp-logging >= 1.11.0a0, < 2.0.0", "opentelemetry-exporter-otlp-proto-http < 2", "opentelemetry-instrumentation-google-genai>=0.3b0, <1.0.0", # TODO(b/538550724): update to stable version of @@ -165,9 +164,7 @@ agent_engines_extra_require = [ "packaging >= 24.0", "cloudpickle >= 3.0, < 4.0", - "google-cloud-logging < 4", "opentelemetry-sdk < 2", - "opentelemetry-exporter-gcp-logging >= 1.11.0a0, < 2.0.0", "opentelemetry-exporter-otlp-proto-http < 2", "pydantic >= 2.11.1, < 3", "typing_extensions", diff --git a/tests/unit/vertex_adk/test_agent_engine_templates_adk.py b/tests/unit/vertex_adk/test_agent_engine_templates_adk.py index 98078145dc..2c10a83022 100644 --- a/tests/unit/vertex_adk/test_agent_engine_templates_adk.py +++ b/tests/unit/vertex_adk/test_agent_engine_templates_adk.py @@ -591,6 +591,7 @@ async def test_streaming_agent_run_with_events_existing_session( # Define an async generator for run_async mock return value async def mock_run_async(*args, **kwargs): from google.adk.events import event + yield event.Event( **{ "author": "currency_exchange_agent", @@ -790,7 +791,6 @@ async def test_async_create_session_with_expire_time_kwargs( expire_time="2026-03-01T00:00:00Z", ) - @pytest.mark.asyncio async def test_async_get_session(self, get_project_id_mock: mock.Mock): app = agent_engines.AdkApp(agent=_TEST_AGENT) @@ -1761,3 +1761,132 @@ def test_default_instrumentor_builder_mtls_no_cert_source( mock_exporter.call_args.kwargs["endpoint"] == adk_template._DEFAULT_TELEMETRY_ENDPOINT ) + + +@pytest.mark.usefixtures("google_auth_mock") +class TestAdkTemplateLoggingExport: + """Covers _default_instrumentor_builder's logging branch. + + Agent Engine logs now go straight to telemetry.googleapis.com rather than + through CloudLoggingExporter, so the MonitoredResource and log naming that + exporter used to produce have to be reproduced by the resource and record + attributes we send. + """ + + def _build(self, monkeypatch, **env): + """Runs the instrumentor builder and returns the installed provider.""" + import opentelemetry._logs + + for key in ( + "OTEL_SEMCONV_STABILITY_OPT_IN", + "GOOGLE_CLOUD_AGENT_ENGINE_ID", + "GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", + "GOOGLE_CLOUD_LOCATION", + "GCP_DEFAULT_LOG_NAME", + ): + monkeypatch.delenv(key, raising=False) + for key, value in env.items(): + monkeypatch.setenv(key, value) + + installed = {} + monkeypatch.setattr( + opentelemetry._logs, + "set_logger_provider", + lambda logger_provider: installed.update(provider=logger_provider), + ) + with mock.patch( + "opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter" + ) as exporter, mock.patch.object(adk_template, "requests_auth"): + adk_template._default_instrumentor_builder( + _TEST_PROJECT, enable_logging=True + ) + return installed["provider"], exporter + + @pytest.mark.parametrize( + "semconv", + ["", "gen_ai_latest_experimental"], + ids=["stable_semconv", "experimental_semconv"], + ) + def test_logs_go_to_telemetry_api(self, monkeypatch, semconv): + """One OTLP path serves both semconv modes. + + The stdout branch experimental semconv used to need is gone with + CloudLoggingExporter (b/480102541). + """ + _, exporter = self._build(monkeypatch, OTEL_SEMCONV_STABILITY_OPT_IN=semconv) + + _, kwargs = exporter.call_args + assert kwargs["endpoint"] == "https://telemetry.googleapis.com/v1/logs" + assert kwargs["headers"]["User-Agent"].startswith("Vertex-Agent-Engine/") + + def test_monitored_resource_matches_cloud_logging_exporter(self, monkeypatch): + """Entries must stay on aiplatform.googleapis.com/ReasoningEngine.""" + provider, _ = self._build( + monkeypatch, + GOOGLE_CLOUD_AGENT_ENGINE_ID="1234567890", + GOOGLE_CLOUD_AGENT_ENGINE_LOCATION=_TEST_LOCATION, + ) + + attributes = provider.resource.attributes + assert ( + attributes["gcp.resource_type"] + == "aiplatform.googleapis.com/ReasoningEngine" + ) + assert attributes["location"] == _TEST_LOCATION + assert attributes["reasoning_engine_id"] == "1234567890" + # Cloud Logging fills resource_container from these. + assert attributes["gcp.project_id"] == _TEST_PROJECT + + def test_monitored_resource_pins_without_a_location(self, monkeypatch): + """A missing location empties the label rather than dropping the pin. + + Every log line from one deployment must land on the same + MonitoredResource, so the set of labels can't depend on which env vars + happen to be set. + + Args: + monkeypatch: The pytest monkeypatch fixture. + """ + provider, _ = self._build( + monkeypatch, GOOGLE_CLOUD_AGENT_ENGINE_ID="1234567890" + ) + + attributes = provider.resource.attributes + assert ( + attributes["gcp.resource_type"] + == "aiplatform.googleapis.com/ReasoningEngine" + ) + assert not attributes["location"] + + @pytest.mark.parametrize( + "log_record_kwargs, expected_attributes", + [ + ({}, {"gcp.log_name": "adk-on-agent-engine"}), + ( + {"event_name": "gen_ai.client.inference.operation.details"}, + {"event.name": "gen_ai.client.inference.operation.details"}, + ), + ( + {"attributes": {"gcp.log_name": "my-log"}}, + {"gcp.log_name": "my-log"}, + ), + ], + ids=["default_log_name", "event_name_as_label", "explicit_log_name"], + ) + def test_log_naming_matches_cloud_logging_exporter( + self, monkeypatch, log_record_kwargs, expected_attributes + ): + """Log name and labels must not shift under customers' log filters.""" + from opentelemetry.sdk._logs import ReadWriteLogRecord + from opentelemetry.sdk._logs._internal import LogRecord + + provider, _ = self._build(monkeypatch) + processor = provider._multi_log_record_processor._log_record_processors[0] + record = ReadWriteLogRecord(log_record=LogRecord(**log_record_kwargs)) + with mock.patch.object( + type(processor).__mro__[1], "on_emit", lambda self, _: None + ): + processor.on_emit(record) + + assert dict(record.log_record.attributes or {}) == expected_attributes + provider.shutdown() diff --git a/vertexai/agent_engines/templates/adk.py b/vertexai/agent_engines/templates/adk.py index 5f814b5b12..9001a319ad 100644 --- a/vertexai/agent_engines/templates/adk.py +++ b/vertexai/agent_engines/templates/adk.py @@ -18,7 +18,6 @@ import enum import os import queue -import sys import threading from typing import ( Any, @@ -117,6 +116,14 @@ # rather than inherit AuthorizedSession's 120s default. _TELEMETRY_API_CHECK_TIMEOUT_SECONDS = 5.0 +_DEFAULT_TELEMETRY_LOGS_ENDPOINT = "https://telemetry.googleapis.com/v1/logs" + +_GCP_LOG_NAME = "gcp.log_name" +_EVENT_NAME = "event.name" +_GCP_RESOURCE_TYPE = "gcp.resource_type" +_LOCATION = "location" +_REASONING_ENGINE_ID = "reasoning_engine_id" + class _MtlsEndpoint(enum.Enum): """Enum for the mTLS endpoint setting.""" @@ -341,8 +348,8 @@ def _warn_missing_dependency( ) MISSING_LOGGING_IMPORT_ERROR_MESSAGE = ( "proceeding with logging disabled because not all packages (i.e." - " `google-cloud-logging`, `opentelemetry-sdk`," - " `opentelemetry-exporter-gcp-logging`) for tracing have been installed" + " `opentelemetry-sdk`, `opentelemetry-exporter-otlp-proto-http`)" + " for logging have been installed" ) if needed_for_tracing and enable_tracing: @@ -351,15 +358,6 @@ def _warn_missing_dependency( _warn(MISSING_LOGGING_IMPORT_ERROR_MESSAGE) return None - def _detect_cloud_resource_id(project_id: str) -> Optional[str]: - location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( - "GOOGLE_CLOUD_LOCATION", "" - ) - agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID") - if all(v is not None for v in (location, agent_engine_id)): - return f"//aiplatform.googleapis.com/projects/{project_id}/locations/{location}/reasoningEngines/{agent_engine_id}" - return None - try: import opentelemetry import opentelemetry.trace @@ -380,30 +378,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: "opentelemetry-sdk", needed_for_tracing=True, needed_for_logging=True ) - import uuid - - # Provide a set of resource attributes but allow to override them with env - # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. - cloud_resource_id = _detect_cloud_resource_id(project_id) - resource = opentelemetry.sdk.resources.Resource.create( - attributes={ - "gcp.project_id": project_id, - "cloud.account.id": project_id, - "cloud.provider": "gcp", - "cloud.platform": "gcp.agent_engine", - "service.name": os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", ""), - "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", - "cloud.region": ( - os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") - or os.getenv("GOOGLE_CLOUD_LOCATION", "") - ), - } - | ( - {"cloud.resource_id": cloud_resource_id} - if cloud_resource_id is not None - else {} - ) - ).merge(opentelemetry.sdk.resources.OTELResourceDetector().detect()) + resource = _create_otel_resource(project_id) if enable_tracing: try: @@ -419,12 +394,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: import google.auth credentials, _ = google.auth.default() - vertex_sdk_version = aip_version.__version__ - otlp_http_version = opentelemetry.exporter.otlp.proto.http.version.__version__ - user_agent = ( - f"Vertex-Agent-Engine/{vertex_sdk_version}" - f" OTel-OTLP-Exporter-Python/{otlp_http_version}" - ) + user_agent = _get_user_agent() session = requests_auth.AuthorizedSession(credentials=credentials) @@ -484,54 +454,47 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: if enable_logging: try: - import opentelemetry.exporter.cloud_logging + import opentelemetry.exporter.otlp.proto.http._log_exporter + import google.auth.transport.requests except (ImportError, AttributeError): return _warn_missing_dependency( - "opentelemetry-exporter-gcp-logging", needed_for_logging=True + "opentelemetry-exporter-otlp-proto-http", needed_for_logging=True ) - class _SimpleLogRecordProcessor( - opentelemetry.sdk._logs.export.SimpleLogRecordProcessor - ): + import google.auth - def force_flush( - self, timeout_millis: int = 30000 - ) -> bool: # pylint: disable=no-self-use - sys.stdout.flush() - sys.stderr.flush() - return True - - logger_provider = opentelemetry.sdk._logs.LoggerProvider(resource=resource) - # Use the legacy log processor when experimental semconv is enabled. - # Exporting JSON logs to stdout is bugged; Agent Engine fails to - # correctly parse the `gen_ai.client.inference.operation.details` - # messages. - # TODO: b/480102541 - Unify both branches once the regression is fixed. - if "gen_ai_latest_experimental" in os.getenv( - "OTEL_SEMCONV_STABILITY_OPT_IN", "" - ).split(","): - logger_provider.add_log_record_processor( - opentelemetry.sdk._logs.export.BatchLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - ), - ) + credentials, _ = google.auth.default() + session = requests_auth.AuthorizedSession(credentials=credentials) + + if _use_client_cert_effective(): + client_cert_source = ( + mtls.default_client_cert_source() + if mtls.has_default_client_cert_source() + else None ) + session.configure_mtls_channel() + endpoint = _get_logs_api_endpoint(client_cert_source) else: - logger_provider.add_log_record_processor( - _SimpleLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - structured_json_file=sys.stdout, - ), - ) + endpoint = _DEFAULT_TELEMETRY_LOGS_ENDPOINT + + # One processor serves stable and experimental semconv. The stdout + # branch experimental records used to need is gone along with the + # Cloud Logging exporter (b/480102541). + logger_provider = opentelemetry.sdk._logs.LoggerProvider( + resource=_create_otel_resource(project_id, "logs") + ) + logger_provider.add_log_record_processor( + _named_batch_log_record_processor( + opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter( + session=session, + endpoint=endpoint, + headers={"User-Agent": _get_user_agent()}, + ), + default_log_name=os.getenv( + "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" + ), ) + ) opentelemetry._logs.set_logger_provider(logger_provider=logger_provider) @@ -621,6 +584,109 @@ def _warn_if_telemetry_api_disabled(): _warn(_TELEMETRY_API_DISABLED_WARNING % (project, project)) +def _get_user_agent() -> str: + """Returns the User-Agent to send on OTLP exports.""" + from google.cloud.aiplatform import version as aip_version + + user_agent = f"Vertex-Agent-Engine/{aip_version.__version__}" + try: + import opentelemetry.exporter.otlp.proto.http.version + + user_agent += ( + " OTel-OTLP-Exporter-Python/" + f"{opentelemetry.exporter.otlp.proto.http.version.__version__}" + ) + except (ImportError, AttributeError): + pass + return user_agent + + +def _get_logs_api_endpoint(client_cert_source: bytes | None = None) -> str: + """Returns the logs endpoint matching _get_api_endpoint's mTLS decision. + + Args: + client_cert_source (bytes | None): The client certificate source. + + Returns: + str: The logs API endpoint to be used. + """ + return _get_api_endpoint(client_cert_source).replace("/v1/traces", "/v1/logs") + + +def _create_otel_resource(project_id: str, for_signal: str = "unspecified"): + """Returns the OTel resource describing the Agent Engine deployment. + + Args: + project_id: Project to which to send telemetry. + for_signal: The signal the resource is for. `logs` adds the + MonitoredResource hints Cloud Logging needs, which must not be set + on the resource traces and metrics share. + + Returns: + The resource to set on the provider for `for_signal`. + """ + import os + import uuid + + import opentelemetry.sdk.resources + + location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( + "GOOGLE_CLOUD_LOCATION", "" + ) + agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", "") + attributes = { + "gcp.project_id": project_id, + "cloud.account.id": project_id, + "cloud.provider": "gcp", + "cloud.platform": "gcp.agent_engine", + "service.name": agent_engine_id, + "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", + "cloud.region": location, + } + if location and agent_engine_id: + attributes["cloud.resource_id"] = ( + f"//aiplatform.googleapis.com/projects/{project_id}" + f"/locations/{location}/reasoningEngines/{agent_engine_id}" + ) + if for_signal == "logs": + # Cloud Logging otherwise detects resource as `generic_task` + attributes[_GCP_RESOURCE_TYPE] = "aiplatform.googleapis.com/ReasoningEngine" + attributes[_LOCATION] = location + attributes[_REASONING_ENGINE_ID] = agent_engine_id + + # Provide a set of resource attributes but allow to override them with env + # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. + return opentelemetry.sdk.resources.Resource.create(attributes=attributes).merge( + opentelemetry.sdk.resources.OTELResourceDetector().detect() + ) + + +def _named_batch_log_record_processor(exporter, *, default_log_name: str): + """Returns a batch processor that keeps log names and labels stable. + + Args: + exporter: The OTLP log exporter to wrap. + default_log_name (str): Log name for records that carry none. + + Returns: + The configured log record processor. + """ + import opentelemetry.sdk._logs.export + + class _Processor(opentelemetry.sdk._logs.export.BatchLogRecordProcessor): + def on_emit(self, log_record) -> None: + record = log_record.log_record + attributes = dict(record.attributes or {}) + if record.event_name: + attributes.setdefault(_EVENT_NAME, record.event_name) + elif _GCP_LOG_NAME not in attributes: + attributes[_GCP_LOG_NAME] = default_log_name + record.attributes = attributes + super().on_emit(log_record) + + return _Processor(exporter) + + def _get_api_endpoint(client_cert_source: bytes | None = None) -> str: """Returns API endpoint based on mTLS configuration and cert availability. diff --git a/vertexai/preview/reasoning_engines/templates/adk.py b/vertexai/preview/reasoning_engines/templates/adk.py index 003812ab3a..9aafd4bbbd 100644 --- a/vertexai/preview/reasoning_engines/templates/adk.py +++ b/vertexai/preview/reasoning_engines/templates/adk.py @@ -28,7 +28,6 @@ from collections.abc import Awaitable import queue import os -import sys import threading import enum from google.auth.transport import mtls @@ -108,6 +107,14 @@ # rather than inherit AuthorizedSession's 120s default. _TELEMETRY_API_CHECK_TIMEOUT_SECONDS = 5.0 +_DEFAULT_TELEMETRY_LOGS_ENDPOINT = "https://telemetry.googleapis.com/v1/logs" + +_GCP_LOG_NAME = "gcp.log_name" +_EVENT_NAME = "event.name" +_GCP_RESOURCE_TYPE = "gcp.resource_type" +_LOCATION = "location" +_REASONING_ENGINE_ID = "reasoning_engine_id" + class _MtlsEndpoint(enum.Enum): """The mTLS endpoint setting.""" @@ -275,6 +282,109 @@ def _warn(msg: str): _warn._LOGGER.warning(msg) # pyright: ignore[reportFunctionMemberAccess] +def _get_user_agent() -> str: + """Returns the User-Agent to send on OTLP exports.""" + from google.cloud.aiplatform import version as aip_version + + user_agent = f"Vertex-Agent-Engine/{aip_version.__version__}" + try: + import opentelemetry.exporter.otlp.proto.http.version + + user_agent += ( + " OTel-OTLP-Exporter-Python/" + f"{opentelemetry.exporter.otlp.proto.http.version.__version__}" + ) + except (ImportError, AttributeError): + pass + return user_agent + + +def _get_logs_api_endpoint(client_cert_source: bytes | None = None) -> str: + """Returns the logs endpoint matching _get_api_endpoint's mTLS decision. + + Args: + client_cert_source (bytes | None): The client certificate source. + + Returns: + str: The logs API endpoint to be used. + """ + return _get_api_endpoint(client_cert_source).replace("/v1/traces", "/v1/logs") + + +def _create_otel_resource(project_id: str, for_signal: str = "unspecified"): + """Returns the OTel resource describing the Agent Engine deployment. + + Args: + project_id: Project to which to send telemetry. + for_signal: The signal the resource is for. `logs` adds the + MonitoredResource hints Cloud Logging needs, which must not be set + on the resource traces and metrics share. + + Returns: + The resource to set on the provider for `for_signal`. + """ + import os + import uuid + + import opentelemetry.sdk.resources + + location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( + "GOOGLE_CLOUD_LOCATION", "" + ) + agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", "") + attributes = { + "gcp.project_id": project_id, + "cloud.account.id": project_id, + "cloud.provider": "gcp", + "cloud.platform": "gcp.agent_engine", + "service.name": agent_engine_id, + "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", + "cloud.region": location, + } + if location and agent_engine_id: + attributes["cloud.resource_id"] = ( + f"//aiplatform.googleapis.com/projects/{project_id}" + f"/locations/{location}/reasoningEngines/{agent_engine_id}" + ) + if for_signal == "logs": + # Cloud Logging otherwise detects resource as `generic_task` + attributes[_GCP_RESOURCE_TYPE] = "aiplatform.googleapis.com/ReasoningEngine" + attributes[_LOCATION] = location + attributes[_REASONING_ENGINE_ID] = agent_engine_id + + # Provide a set of resource attributes but allow to override them with env + # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. + return opentelemetry.sdk.resources.Resource.create(attributes=attributes).merge( + opentelemetry.sdk.resources.OTELResourceDetector().detect() + ) + + +def _named_batch_log_record_processor(exporter, *, default_log_name: str): + """Returns a batch processor that keeps log names and labels stable. + + Args: + exporter: The OTLP log exporter to wrap. + default_log_name (str): Log name for records that carry none. + + Returns: + The configured log record processor. + """ + import opentelemetry.sdk._logs.export + + class _Processor(opentelemetry.sdk._logs.export.BatchLogRecordProcessor): + def on_emit(self, log_record) -> None: + record = log_record.log_record + attributes = dict(record.attributes or {}) + if record.event_name: + attributes.setdefault(_EVENT_NAME, record.event_name) + elif _GCP_LOG_NAME not in attributes: + attributes[_GCP_LOG_NAME] = default_log_name + record.attributes = attributes + super().on_emit(log_record) + + return _Processor(exporter) + + def _get_api_endpoint(client_cert_source: bytes | None = None) -> str: """Returns API endpoint based on mTLS configuration and cert availability. @@ -402,8 +512,8 @@ def _warn_missing_dependency( ) MISSING_LOGGING_IMPORT_ERROR_MESSAGE = ( "proceeding with logging disabled because not all packages (i.e." - " `google-cloud-logging`, `opentelemetry-sdk`," - " `opentelemetry-exporter-gcp-logging`) for tracing have been installed" + " `opentelemetry-sdk`, `opentelemetry-exporter-otlp-proto-http`)" + " for logging have been installed" ) if needed_for_tracing and enable_tracing: @@ -412,15 +522,6 @@ def _warn_missing_dependency( _warn(MISSING_LOGGING_IMPORT_ERROR_MESSAGE) return None - def _detect_cloud_resource_id(project_id: str) -> Optional[str]: - location = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") or os.getenv( - "GOOGLE_CLOUD_LOCATION", "" - ) - agent_engine_id = os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", None) - if all(v is not None for v in (location, agent_engine_id)): - return f"//aiplatform.googleapis.com/projects/{project_id}/locations/{location}/reasoningEngines/{agent_engine_id}" - return None - try: import opentelemetry import opentelemetry.trace @@ -441,30 +542,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: "opentelemetry-sdk", needed_for_tracing=True, needed_for_logging=True ) - import uuid - - # Provide a set of resource attributes but allow to override them with env - # variables like OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME. - cloud_resource_id = _detect_cloud_resource_id(project_id) - resource = opentelemetry.sdk.resources.Resource.create( - attributes={ - "gcp.project_id": project_id, - "cloud.account.id": project_id, - "cloud.provider": "gcp", - "cloud.platform": "gcp.agent_engine", - "service.name": os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_ID", ""), - "service.instance.id": f"{uuid.uuid4().hex}-{os.getpid()}", - "cloud.region": ( - os.getenv("GOOGLE_CLOUD_AGENT_ENGINE_LOCATION", "") - or os.getenv("GOOGLE_CLOUD_LOCATION", "") - ), - } - | ( - {"cloud.resource_id": cloud_resource_id} - if cloud_resource_id is not None - else {} - ) - ).merge(opentelemetry.sdk.resources.OTELResourceDetector().detect()) + resource = _create_otel_resource(project_id) if enable_tracing: try: @@ -480,12 +558,7 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: import google.auth credentials, _ = google.auth.default() - vertex_sdk_version = aip_version.__version__ - otlp_http_version = opentelemetry.exporter.otlp.proto.http.version.__version__ - user_agent = ( - f"Vertex-Agent-Engine/{vertex_sdk_version}" - f" OTel-OTLP-Exporter-Python/{otlp_http_version}" - ) + user_agent = _get_user_agent() session = requests_auth.AuthorizedSession(credentials=credentials) @@ -545,54 +618,47 @@ def _detect_cloud_resource_id(project_id: str) -> Optional[str]: if enable_logging: try: - import opentelemetry.exporter.cloud_logging + import opentelemetry.exporter.otlp.proto.http._log_exporter + import google.auth.transport.requests except (ImportError, AttributeError): return _warn_missing_dependency( - "opentelemetry-exporter-gcp-logging", needed_for_logging=True + "opentelemetry-exporter-otlp-proto-http", needed_for_logging=True ) - class _SimpleLogRecordProcessor( - opentelemetry.sdk._logs.export.SimpleLogRecordProcessor - ): + import google.auth - def force_flush( - self, timeout_millis: int = 30000 - ) -> bool: # pylint: disable=no-self-use - sys.stdout.flush() - sys.stderr.flush() - return True - - logger_provider = opentelemetry.sdk._logs.LoggerProvider(resource=resource) - # Use the legacy log processor when experimental semconv is enabled. - # Exporting JSON logs to stdout is bugged; Agent Engine fails to - # correctly parse the `gen_ai.client.inference.operation.details` - # messages. - # TODO: b/480102541 - Unify both branches once the regression is fixed. - if "gen_ai_latest_experimental" in os.getenv( - "OTEL_SEMCONV_STABILITY_OPT_IN", "" - ).split(","): - logger_provider.add_log_record_processor( - opentelemetry.sdk._logs.export.BatchLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - ), - ) + credentials, _ = google.auth.default() + session = requests_auth.AuthorizedSession(credentials=credentials) + + if _use_client_cert_effective(): + client_cert_source = ( + mtls.default_client_cert_source() + if mtls.has_default_client_cert_source() + else None ) + session.configure_mtls_channel() + endpoint = _get_logs_api_endpoint(client_cert_source) else: - logger_provider.add_log_record_processor( - _SimpleLogRecordProcessor( - opentelemetry.exporter.cloud_logging.CloudLoggingExporter( - project_id=project_id, - default_log_name=os.getenv( - "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" - ), - structured_json_file=sys.stdout, - ), - ) + endpoint = _DEFAULT_TELEMETRY_LOGS_ENDPOINT + + # One processor serves stable and experimental semconv. The stdout + # branch experimental records used to need is gone along with the + # Cloud Logging exporter (b/480102541). + logger_provider = opentelemetry.sdk._logs.LoggerProvider( + resource=_create_otel_resource(project_id, "logs") + ) + logger_provider.add_log_record_processor( + _named_batch_log_record_processor( + opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter( + session=session, + endpoint=endpoint, + headers={"User-Agent": _get_user_agent()}, + ), + default_log_name=os.getenv( + "GCP_DEFAULT_LOG_NAME", "adk-on-agent-engine" + ), ) + ) opentelemetry._logs.set_logger_provider(logger_provider=logger_provider)