move decoder to utils

Author: DylanRussell

instrumentation-genai/opentelemetry-instrumentation-vertexai/src/opentelemetry/instrumentation/vertexai/patch.py

@@ -53,6 +53,7 @@
     ContentCapturingMode,
     Text,
 )
+from opentelemetry.util.genai.utils import Base64JsonEncoder
 
 if TYPE_CHECKING:
     from google.cloud.aiplatform_v1.services.prediction_service import client
@@ -213,7 +214,7 @@ def handle_response(
                     ):
                         span.set_attributes(
                             {
-                                k: json.dumps(v)
+                                k: json.dumps(v, cls=Base64JsonEncoder)
                                 for k, v in content_attributes.items()
                             }
                         )

instrumentation-genai/opentelemetry-instrumentation-vertexai/tests/conftest.py

@@ -192,7 +192,7 @@ def instrument_with_experimental_semconvs(
 
 @pytest.fixture(scope="function")
 def instrument_with_upload_hook(
-    tracer_provider, event_logger_provider, meter_provider, request
+    tracer_provider, event_logger_provider, meter_provider
 ):
     # Reset global state..
     _OpenTelemetrySemanticConventionStability._initialized = False

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -19,7 +19,6 @@
 import logging
 import posixpath
 import threading
-from base64 import b64encode
 from concurrent.futures import Future, ThreadPoolExecutor
 from contextlib import ExitStack
 from dataclasses import asdict, dataclass
@@ -34,7 +33,7 @@
 from opentelemetry._logs import LogRecord
 from opentelemetry.semconv._incubating.attributes import gen_ai_attributes
 from opentelemetry.trace import Span
-from opentelemetry.util.genai import types
+from opentelemetry.util.genai import Base64JsonEncoder, types
 from opentelemetry.util.genai.completion_hook import CompletionHook
 from opentelemetry.util.genai.environment_variables import (
     OTEL_INSTRUMENTATION_GENAI_UPLOAD_FORMAT,
@@ -281,10 +280,3 @@ def shutdown(self, *, timeout_sec: float = 10.0) -> None:
 
             # Queue is flushed and blocked, start shutdown
             self._executor.shutdown(wait=False)
-
-
-class Base64JsonEncoder(json.JSONEncoder):
-    def default(self, o: Any) -> Any:
-        if isinstance(o, bytes):
-            return b64encode(o).decode()
-        return super().default(o)

util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py

@@ -12,8 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import logging
 import os
+from base64 import b64encode
+from typing import Any
 
 from opentelemetry.instrumentation._semconv import (
     _OpenTelemetrySemanticConventionStability,
@@ -54,3 +57,10 @@ def get_content_capturing_mode() -> ContentCapturingMode:
             ", ".join(e.name for e in ContentCapturingMode),
         )
         return ContentCapturingMode.NO_CONTENT
+
+
+class Base64JsonEncoder(json.JSONEncoder):
+    def default(self, o: Any) -> Any:
+        if isinstance(o, bytes):
+            return b64encode(o).decode()
+        return super().default(o)