Add temporary support for aiplatform.googleapis.com/ReasoningEngine monitored resource in logging exporter

Author: aabmass

opentelemetry-exporter-gcp-logging/src/opentelemetry/exporter/cloud_logging/__init__.py

@@ -147,7 +147,7 @@ def _convert_any_value_to_string(value: Any) -> str:
 
 # Be careful not to mutate original body. Make copies of anything that needs to change.
 def _sanitized_body(
-    body: Mapping[str, AnyValue]
+    body: Mapping[str, AnyValue],
 ) -> MutableMapping[str, AnyValue]:
     new_body: MutableMapping[str, AnyValue] = {}
     for key, value in body.items():
@@ -212,6 +212,45 @@ def is_log_id_valid(log_id: str) -> bool:
     )
 
 
+def _get_monitored_resource(
+    resource: Optional[Resource],
+) -> MonitoredResource | None:
+    if not resource:
+        return None
+
+    # TODO: Remove temporary special case for Vertex Agent Engine
+    # https://github.com/GoogleCloudPlatform/opentelemetry-operations-python/issues/444
+    cloud_resource_id = resource.attributes.get("cloud.resource_id")
+    if isinstance(cloud_resource_id, str) and (
+        match := re.match(
+            r"//aiplatform\.googleapis\.com/projects/(?P<project_id>[^/]+)"
+            r"/locations/(?P<location>[^/]+)"
+            r"/reasoningEngines/(?P<agent_engine_id>[^/]+)",
+            cloud_resource_id,
+        )
+    ):
+        project_id = match.group("project_id")
+        location = match.group("location")
+        agent_engine_id = match.group("agent_engine_id")
+        return MonitoredResource(
+            type="aiplatform.googleapis.com/ReasoningEngine",
+            labels={
+                "resource_container": project_id,
+                "location": location,
+                "reasoning_engine_id": agent_engine_id,
+            },
+        )
+
+    monitored_resource_data = get_monitored_resource(resource)
+    if not monitored_resource_data:
+        return None
+
+    return MonitoredResource(
+        type=monitored_resource_data.type,
+        labels=monitored_resource_data.labels,
+    )
+
+
 class CloudLoggingExporter(LogExporter):
     def __init__(
         self,
@@ -302,14 +341,10 @@ def export(self, batch: Sequence[LogData]):
             else:
                 ts.FromDatetime(now)
             log_entry.timestamp = ts
-            monitored_resource_data = get_monitored_resource(
-                log_record.resource or Resource({})
-            )
-            if monitored_resource_data:
-                log_entry.resource = MonitoredResource(
-                    type=monitored_resource_data.type,
-                    labels=monitored_resource_data.labels,
-                )
+            if monitored_resource := _get_monitored_resource(
+                log_record.resource
+            ):
+                log_entry.resource = monitored_resource
             log_entry.trace_sampled = (
                 log_record.trace_flags is not None
                 and log_record.trace_flags.sampled
@@ -366,9 +401,9 @@ def _write_log_entries_to_file(file: TextIO, log_entries: list[LogEntry]):
             )
             json_dict["logging.googleapis.com/labels"] = dict(entry.labels)
             json_dict["logging.googleapis.com/spanId"] = entry.span_id
-            json_dict[
-                "logging.googleapis.com/trace_sampled"
-            ] = entry.trace_sampled
+            json_dict["logging.googleapis.com/trace_sampled"] = (
+                entry.trace_sampled
+            )
             json_dict["logging.googleapis.com/trace"] = entry.trace
 
             if entry.text_payload:

opentelemetry-exporter-gcp-logging/tests/__snapshots__/test_cloud_logging/test_agent_engine_monitored_resources[client].json

@@ -0,0 +1,19 @@
+[
+  {
+    "entries": [
+      {
+        "logName": "projects/fakeproject/logs/test",
+        "resource": {
+          "labels": {
+            "location": "europe-west3",
+            "reasoning_engine_id": "8477639270431981568",
+            "resource_container": "some-project123-321"
+          },
+          "type": "aiplatform.googleapis.com/ReasoningEngine"
+        },
+        "timestamp": "2025-01-15T21:25:10.997977393Z"
+      }
+    ],
+    "partialSuccess": true
+  }
+]

opentelemetry-exporter-gcp-logging/tests/__snapshots__/test_cloud_logging/test_agent_engine_monitored_resources[structured_json].json

@@ -0,0 +1,10 @@
+[
+  {
+    "logging.googleapis.com/labels": {},
+    "logging.googleapis.com/spanId": "",
+    "logging.googleapis.com/trace": "",
+    "logging.googleapis.com/trace_sampled": false,
+    "severity": "DEFAULT",
+    "time": "2025-01-15T21:25:10.997977393Z"
+  }
+]

opentelemetry-exporter-gcp-logging/tests/test_cloud_logging.py

@@ -165,6 +165,25 @@ def test_user_agent(cloudloggingfake: CloudLoggingFake) -> None:
         )
 
 
+def test_agent_engine_monitored_resources(
+    export_and_assert_snapshot: ExportAndAssertSnapshot,
+) -> None:
+    log_data = [
+        LogData(
+            log_record=LogRecord(
+                timestamp=1736976310997977393,
+                resource=Resource(
+                    {
+                        "cloud.resource_id": "//aiplatform.googleapis.com/projects/some-project123-321/locations/europe-west3/reasoningEngines/8477639270431981568"
+                    }
+                ),
+            ),
+            instrumentation_scope=InstrumentationScope("test"),
+        ),
+    ]
+    export_and_assert_snapshot(log_data)
+
+
 def test_convert_otlp_dict_body(
     export_and_assert_snapshot: ExportAndAssertSnapshot,
 ) -> None: