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)