From 824b54790d21789708e48bf9fae4fd51658600e7 Mon Sep 17 00:00:00 2001 From: xiongyun Date: Wed, 22 Jul 2026 14:22:30 +0800 Subject: [PATCH 1/4] add multimodal runtime config hot-reload --- .../util/genai/_multimodal_processing.py | 100 +++-- .../util/genai/_multimodal_upload/__init__.py | 8 + .../util/genai/_multimodal_upload/_base.py | 3 + .../util/genai/_multimodal_upload/config.py | 356 ++++++++++++++++++ .../genai/_multimodal_upload/fs_uploader.py | 17 +- .../multimodal_upload_hook.py | 218 ++++++++--- .../genai/_multimodal_upload/pre_uploader.py | 74 +++- .../src/opentelemetry/util/genai/utils.py | 51 +-- .../tests/_multimodal_upload/conftest.py | 38 ++ .../multimodal_test_helpers.py | 68 ++++ .../_multimodal_upload/test_default_hooks.py | 5 + .../test_multimodal_runtime_config.py | 290 ++++++++++++++ .../test_multimodal_upload_hook.py | 252 +++++++++++-- .../_multimodal_upload/test_pre_uploader.py | 22 +- .../test_pre_uploader_audio.py | 26 +- 15 files changed, 1355 insertions(+), 173 deletions(-) create mode 100644 util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py create mode 100644 util/opentelemetry-util-genai/tests/_multimodal_upload/conftest.py create mode 100644 util/opentelemetry-util-genai/tests/_multimodal_upload/multimodal_test_helpers.py create mode 100644 util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py index 5f21f00f1..455489576 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py @@ -80,7 +80,6 @@ def __init__(self, ...): from opentelemetry.util.genai.utils import ( # pylint: disable=no-name-in-module gen_ai_json_dumps, get_content_capturing_mode, - get_multimodal_upload_mode, is_experimental_mode, ) @@ -140,30 +139,7 @@ class MultimodalProcessingMixin: def _init_multimodal(self) -> None: """Initialize multimodal-related instance attributes, called in subclass __init__""" - self._multimodal_enabled = False - - if get_multimodal_upload_mode() == "none": - return - - try: - capture_enabled = ( - is_experimental_mode() - and get_content_capturing_mode() - in ( - ContentCapturingMode.SPAN_ONLY, - ContentCapturingMode.SPAN_AND_EVENT, - ) - ) - except ValueError: - # get_content_capturing_mode raises ValueError when GEN_AI stability mode is DEFAULT - capture_enabled = False - - if not capture_enabled: - return - - uploader, pre_uploader = self._get_uploader_and_pre_uploader() - if uploader is not None and pre_uploader is not None: - self._multimodal_enabled = True + self._multimodal_enabled = True # ==================== Public Methods ==================== @@ -370,19 +346,42 @@ def _at_fork_reinit(cls) -> None: # ==================== Internal Methods ==================== - def _should_async_process(self, invocation: _MultimodalInvocation) -> bool: - """Determine whether async processing is needed + @staticmethod + def _content_capture_supports_multimodal() -> bool: + try: + return is_experimental_mode() and get_content_capturing_mode() in ( + ContentCapturingMode.SPAN_ONLY, + ContentCapturingMode.SPAN_AND_EVENT, + ) + except Exception: # pylint: disable=broad-except + return False - Condition: Has multimodal data and multimodal upload switch is not 'none' - """ + def _should_async_process(self, invocation: _MultimodalInvocation) -> bool: + """Determine whether async processing is needed.""" if not self._multimodal_enabled: return False - return MultimodalProcessingMixin._quick_has_multimodal(invocation) + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + snapshot = get_multimodal_config_snapshot() + if snapshot.upload_mode == "none": + return False + if not self._content_capture_supports_multimodal(): + return False + if not self._quick_has_multimodal_for_mode(invocation, snapshot): + return False + return True @staticmethod - def _quick_has_multimodal(invocation: _MultimodalInvocation) -> bool: - """Quick detection of multimodal data (O(n), no network)""" + def _quick_has_multimodal_for_mode( + invocation: _MultimodalInvocation, + snapshot: Any, + ) -> bool: + """Quick detection of multimodal data respecting upload_mode.""" + check_input = snapshot.process_input + check_output = snapshot.process_output def _check_messages( messages: Optional[List[InputMessage] | List[OutputMessage]], @@ -397,8 +396,22 @@ def _check_messages( return True return False - return _check_messages(invocation.input_messages) or _check_messages( - invocation.output_messages + if check_input and _check_messages(invocation.input_messages): + return True + if check_output and _check_messages(invocation.output_messages): + return True + return False + + @staticmethod + def _quick_has_multimodal(invocation: _MultimodalInvocation) -> bool: + """Quick detection of multimodal data (O(n), no network)""" + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + return MultimodalProcessingMixin._quick_has_multimodal_for_mode( + invocation, + get_multimodal_config_snapshot(), ) @classmethod @@ -751,10 +764,14 @@ def _get_uploader_and_pre_uploader( # pylint: disable=no-self-use """ try: from opentelemetry.util.genai._multimodal_upload import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 - get_or_load_uploader_pair, + get_or_rebuild_uploader_pair, + ) + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, ) - return get_or_load_uploader_pair() + snapshot = get_multimodal_config_snapshot() + return get_or_rebuild_uploader_pair(snapshot) except ImportError: return None, None @@ -766,7 +783,14 @@ def _upload_and_set_metadata( pre_uploader: "PreUploader", ) -> None: """Upload multimodal data and set metadata attributes on span""" - self._separate_and_upload(span, invocation, uploader, pre_uploader) + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + snapshot = get_multimodal_config_snapshot() + self._separate_and_upload( + span, invocation, uploader, pre_uploader, snapshot + ) input_metadata, output_metadata = ( MultimodalProcessingMixin._extract_multimodal_metadata( @@ -790,6 +814,7 @@ def _separate_and_upload( # pylint: disable=no-self-use invocation: _MultimodalInvocation, uploader: "Uploader", pre_uploader: "PreUploader", + config_snapshot: Any, ) -> None: """Separate multimodal data and submit for upload""" try: @@ -803,6 +828,7 @@ def _separate_and_upload( # pylint: disable=no-self-use start_time_ns, invocation.input_messages, invocation.output_messages, + config_snapshot=config_snapshot, ) for item in upload_items: diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py index 4ec95d3b5..38a4c6ad6 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py @@ -26,12 +26,17 @@ get_or_load_pre_uploader, get_or_load_uploader, get_or_load_uploader_pair, + get_or_rebuild_uploader_pair, get_pre_uploader, get_uploader, get_uploader_pair, load_pre_uploader_hook, load_uploader_hook, ) +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + get_multimodal_config_snapshot, + update_multimodal_runtime_config, +) try: from opentelemetry.util.genai._multimodal_upload.fs_uploader import ( @@ -58,6 +63,9 @@ "get_or_load_uploader", "get_or_load_pre_uploader", "get_or_load_uploader_pair", + "get_or_rebuild_uploader_pair", + "get_multimodal_config_snapshot", + "update_multimodal_runtime_config", "get_uploader", "get_pre_uploader", ] diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/_base.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/_base.py index a8714e799..c7f9b88da 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/_base.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/_base.py @@ -104,6 +104,7 @@ def pre_upload( start_time_utc_nano: int, input_messages: Optional[List[Any]], output_messages: Optional[List[Any]], + config_snapshot: Optional[Any] = None, ) -> List[PreUploadItem]: """ Preprocess multimodal data in messages: @@ -117,6 +118,8 @@ def pre_upload( start_time_utc_nano: Start time in nanoseconds input_messages: List of input messages (with .parts attribute), will be modified in-place output_messages: List of output messages (with .parts attribute), will be modified in-place + config_snapshot: Optional runtime config snapshot; when omitted, implementation + may read process-wide multimodal config itself Returns: List of PreUploadItem to be uploaded, returns empty list on failure diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py new file mode 100644 index 000000000..169bc99b8 --- /dev/null +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py @@ -0,0 +1,356 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Runtime configuration snapshot for multimodal upload.""" + +from __future__ import annotations + +import logging +import os +import re +import threading +from dataclasses import dataclass, replace +from typing import Any, Optional, Sequence, Tuple, Union + +from opentelemetry.util.genai.extended_environment_variables import ( # pylint: disable=no-name-in-module + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_ALLOWED_ROOT_PATHS, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, +) + +_logger = logging.getLogger(__name__) + +_VALID_UPLOAD_MODES = frozenset({"none", "input", "output", "both"}) +_VALID_HOOKS = frozenset({"arms", "fs"}) +DEFAULT_SLS_LOGSTORE = "logstore-multimodal" +DEFAULT_MULTIMODAL_UPLOADER_HOOK = "fs" +DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK = "fs" + +# Optional env for bootstrap / tests (also used by commercial SLS hooks). +_APSARA_SLS_PROJECT_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_PROJECT" +_APSARA_SLS_LOGSTORE_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_LOGSTORE" +_APSARA_SLS_ENDPOINT_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ENDPOINT" +_APSARA_SLS_AUTH_TYPE_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_AUTH_TYPE" +_APSARA_SLS_ACCESS_KEY_ID_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ACCESS_KEY_ID" +_APSARA_SLS_ACCESS_KEY_SECRET_ENV = ( + "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ACCESS_KEY_SECRET" +) +_APSARA_SLS_STS_TOKEN_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_STS_TOKEN" + +STRATEGY_FIELDS = frozenset( + { + "upload_mode", + "download_enabled", + "local_file_enabled", + "allowed_root_paths", + } +) + +UPLOADER_GENERATION_FIELDS = frozenset( + { + "storage_base_path", + "uploader_hook_name", + "pre_uploader_hook_name", + "sls_project", + "sls_logstore", + } +) + + +@dataclass(frozen=True) +class MultimodalConfigSnapshot: + upload_mode: str + download_enabled: bool + local_file_enabled: bool + allowed_root_paths: Tuple[str, ...] + storage_base_path: Optional[str] + uploader_hook_name: str + pre_uploader_hook_name: str + sls_project: Optional[str] + sls_logstore: Optional[str] + sls_endpoint: Optional[str] + sls_auth_type: Optional[str] + sls_access_key_id: Optional[str] + sls_access_key_secret: Optional[str] + sls_sts_token: Optional[str] + version: int = 0 + strategy_version: int = 0 + uploader_generation: int = 0 + + @property + def process_input(self) -> bool: + return self.upload_mode in ("input", "both") + + @property + def process_output(self) -> bool: + return self.upload_mode in ("output", "both") + + @property + def effective_storage_base_path(self) -> Optional[str]: + if self.uploader_hook_name == "arms": + project = self.sls_project + if not project: + return None + logstore = self.sls_logstore or DEFAULT_SLS_LOGSTORE + return f"sls://{project}/{logstore}" + return self.storage_base_path + + +def _parse_env_bool(value: Optional[str], default: bool = False) -> bool: + if not value: + return default + return value.lower() in ("true", "1", "yes") + + +def _coerce_bool(value: Any, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + text = str(value).strip() + if not text: + return default + return text.lower() in ("true", "1", "yes") + + +def _normalize_upload_mode(value: Optional[str]) -> str: + if not value: + return "none" + mode = str(value).strip().lower() + if mode not in _VALID_UPLOAD_MODES: + _logger.warning("Invalid multimodal upload_mode %r, fallback to none", value) + return "none" + return mode + + +def normalize_multimodal_hook_name(value: Any) -> Optional[str]: + """Return normalized hook name, or None if unsupported.""" + if value is None: + return None + hook_name = str(value).strip().lower() + if hook_name not in _VALID_HOOKS: + _logger.warning("Unsupported multimodal hook name: %r", value) + return None + return hook_name + + +def _normalize_allowed_root_paths( + value: Union[str, Sequence[str], None], +) -> Tuple[str, ...]: + if value is None: + return () + if isinstance(value, str): + if not value.strip(): + return () + parts = [p.strip() for p in re.split(r"[,]", value) if p.strip()] + else: + parts = [str(p).strip() for p in value if str(p).strip()] + normalized: list[str] = [] + seen: set[str] = set() + for part in parts: + abs_path = os.path.abspath(part) + if abs_path in seen: + continue + seen.add(abs_path) + if not os.path.exists(abs_path): + _logger.debug( + "Multimodal allowed_root_paths entry does not exist: %s", + abs_path, + ) + normalized.append(abs_path) + return tuple(normalized) + + +def _snapshot_from_env(*, version: int = 0) -> MultimodalConfigSnapshot: + upload_mode = _normalize_upload_mode( + os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, "none") + ) + download_enabled = _parse_env_bool( + os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED) + ) + local_file_enabled = _parse_env_bool( + os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED) + ) + allowed_root_paths = _normalize_allowed_root_paths( + os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_ALLOWED_ROOT_PATHS, "") + ) + storage_base_path = os.getenv( + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH + ) + uploader_hook_name = normalize_multimodal_hook_name( + os.getenv( + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, + DEFAULT_MULTIMODAL_UPLOADER_HOOK, + ) + ) or DEFAULT_MULTIMODAL_UPLOADER_HOOK + pre_uploader_hook_name = normalize_multimodal_hook_name( + os.getenv( + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, + DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK, + ) + ) or DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK + + return MultimodalConfigSnapshot( + upload_mode=upload_mode, + download_enabled=download_enabled, + local_file_enabled=local_file_enabled, + allowed_root_paths=allowed_root_paths, + storage_base_path=storage_base_path, + uploader_hook_name=uploader_hook_name, + pre_uploader_hook_name=pre_uploader_hook_name, + sls_project=os.getenv(_APSARA_SLS_PROJECT_ENV) or None, + sls_logstore=os.getenv(_APSARA_SLS_LOGSTORE_ENV) or None, + sls_endpoint=os.getenv(_APSARA_SLS_ENDPOINT_ENV) or None, + sls_auth_type=os.getenv(_APSARA_SLS_AUTH_TYPE_ENV) or None, + sls_access_key_id=os.getenv(_APSARA_SLS_ACCESS_KEY_ID_ENV) or None, + sls_access_key_secret=os.getenv(_APSARA_SLS_ACCESS_KEY_SECRET_ENV) + or None, + sls_sts_token=os.getenv(_APSARA_SLS_STS_TOKEN_ENV) or None, + version=version, + strategy_version=0, + uploader_generation=0, + ) + + +def _coalesce_optional_str(value: Any) -> Optional[str]: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _normalize_and_validate( + old: MultimodalConfigSnapshot, + fields: dict[str, Any], +) -> MultimodalConfigSnapshot: + merged: dict[str, Any] = { + "upload_mode": old.upload_mode, + "download_enabled": old.download_enabled, + "local_file_enabled": old.local_file_enabled, + "allowed_root_paths": old.allowed_root_paths, + "storage_base_path": old.storage_base_path, + "uploader_hook_name": old.uploader_hook_name, + "pre_uploader_hook_name": old.pre_uploader_hook_name, + "sls_project": old.sls_project, + "sls_logstore": old.sls_logstore, + "sls_endpoint": old.sls_endpoint, + "sls_auth_type": old.sls_auth_type, + "sls_access_key_id": old.sls_access_key_id, + "sls_access_key_secret": old.sls_access_key_secret, + "sls_sts_token": old.sls_sts_token, + } + + for key, value in fields.items(): + if value is None or key not in merged: + continue + if key == "upload_mode": + merged[key] = _normalize_upload_mode(value) + elif key in ("download_enabled", "local_file_enabled"): + merged[key] = _coerce_bool(value) + elif key == "allowed_root_paths": + merged[key] = _normalize_allowed_root_paths(value) + elif key in ("uploader_hook_name", "pre_uploader_hook_name"): + normalized_hook = normalize_multimodal_hook_name(value) + if normalized_hook is not None: + merged[key] = normalized_hook + elif key in ( + "storage_base_path", + "sls_project", + "sls_logstore", + ): + merged[key] = _coalesce_optional_str(value) + + return MultimodalConfigSnapshot( + upload_mode=merged["upload_mode"], + download_enabled=merged["download_enabled"], + local_file_enabled=merged["local_file_enabled"], + allowed_root_paths=merged["allowed_root_paths"], + storage_base_path=merged["storage_base_path"], + uploader_hook_name=merged["uploader_hook_name"], + pre_uploader_hook_name=merged["pre_uploader_hook_name"], + sls_project=merged["sls_project"], + sls_logstore=merged["sls_logstore"], + sls_endpoint=merged["sls_endpoint"], + sls_auth_type=merged["sls_auth_type"], + sls_access_key_id=merged["sls_access_key_id"], + sls_access_key_secret=merged["sls_access_key_secret"], + sls_sts_token=merged["sls_sts_token"], + version=old.version, + strategy_version=old.strategy_version, + uploader_generation=old.uploader_generation, + ) + + +def _next_strategy_version( + old: MultimodalConfigSnapshot, + new: MultimodalConfigSnapshot, +) -> int: + if any( + getattr(old, field) != getattr(new, field) for field in STRATEGY_FIELDS + ): + return old.strategy_version + 1 + return old.strategy_version + + +def _next_uploader_generation( + old: MultimodalConfigSnapshot, + new: MultimodalConfigSnapshot, +) -> int: + if any( + getattr(old, field) != getattr(new, field) + for field in UPLOADER_GENERATION_FIELDS + ): + return old.uploader_generation + 1 + return old.uploader_generation + + +class MultimodalRuntimeConfig: + """Process-wide mutable multimodal runtime configuration.""" + + def __init__(self) -> None: + self._lock = threading.RLock() + self._snapshot = _snapshot_from_env(version=0) + + def get_snapshot(self) -> MultimodalConfigSnapshot: + return self._snapshot + + def update(self, **fields: Any) -> MultimodalConfigSnapshot: + with self._lock: + old = self._snapshot + new = _normalize_and_validate(old, fields) + if new == old: + return self._snapshot + new = replace( + new, + version=old.version + 1, + strategy_version=_next_strategy_version(old, new), + uploader_generation=_next_uploader_generation(old, new), + ) + self._snapshot = new + return self._snapshot + + +_runtime_config = MultimodalRuntimeConfig() + + +def get_multimodal_config_snapshot() -> MultimodalConfigSnapshot: + return _runtime_config.get_snapshot() + + +def update_multimodal_runtime_config(**fields: Any) -> MultimodalConfigSnapshot: + return _runtime_config.update(**fields) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py index 884ee6528..4e4379740 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py @@ -60,11 +60,18 @@ def hash_content(content: bytes | str) -> str: return hashlib.sha256(content, usedforsecurity=False).hexdigest() -def fs_uploader_hook() -> Optional[Uploader]: - """Create default FsUploader from environment variables.""" - base_path = os.environ.get( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH - ) +def fs_uploader_hook( + snapshot: Optional[Any] = None, +) -> Optional[Uploader]: + """Create default FsUploader from runtime snapshot.""" + if snapshot is None: + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + snapshot = get_multimodal_config_snapshot() + + base_path = snapshot.effective_storage_base_path or snapshot.storage_base_path if not base_path: _logger.warning( "%s is required but not set, multimodal uploader disabled", diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py index d0c8111ae..88212e0b4 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py @@ -15,14 +15,13 @@ from __future__ import annotations import logging +import threading from importlib import metadata -from typing import Any, Optional, Protocol, runtime_checkable +from typing import Any, Callable, Optional, Protocol, runtime_checkable -from opentelemetry.util._once import Once -from opentelemetry.util.genai.utils import ( # pylint: disable=no-name-in-module - get_multimodal_pre_uploader_hook_name, - get_multimodal_upload_mode, - get_multimodal_uploader_hook_name, +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + MultimodalConfigSnapshot, + get_multimodal_config_snapshot, ) from ._base import PreUploader, Uploader @@ -40,7 +39,10 @@ _uploader: Optional[Uploader] = None _pre_uploader: Optional[PreUploader] = None -_load_once = Once() +_uploader_generation = -1 +_failed_generation = -1 +_building_generation = -1 +_uploader_pair_lock = threading.RLock() def _iter_entry_points(group: str) -> list[Any]: @@ -53,52 +55,108 @@ def _iter_entry_points(group: str) -> list[Any]: @runtime_checkable class UploaderHook(Protocol): - def __call__(self) -> Optional[Uploader]: ... + def __call__( + self, + snapshot: Optional[MultimodalConfigSnapshot] = None, + ) -> Optional[Uploader]: ... @runtime_checkable class PreUploaderHook(Protocol): - def __call__(self) -> Optional[PreUploader]: ... + def __call__( + self, + snapshot: Optional[MultimodalConfigSnapshot] = None, + ) -> Optional[PreUploader]: ... + + +def _call_hook( + hook: Callable[..., Any], + snapshot: MultimodalConfigSnapshot, +) -> Optional[object]: + try: + return hook(snapshot) + except TypeError: + return hook() def _load_by_name( *, hook_name: str, group: str, + snapshot: MultimodalConfigSnapshot, ) -> Optional[object]: for entry_point in _iter_entry_points(group): name = str(entry_point.name) if name != hook_name: continue try: - return entry_point.load()() + hook = entry_point.load() + return _call_hook(hook, snapshot) except Exception: # pylint: disable=broad-except _logger.exception("%s hook %s configuration failed", group, name) return None return None -def load_uploader_hook() -> Optional[Uploader]: +def _is_complete_pair( + pair: tuple[Optional[Uploader], Optional[PreUploader]], +) -> bool: + uploader, pre_uploader = pair + return uploader is not None and pre_uploader is not None + + +def _schedule_retired_pair_shutdown( + pair: tuple[Optional[Uploader], Optional[PreUploader]], +) -> None: + uploader, pre_uploader = pair + if uploader is None and pre_uploader is None: + return + + def _shutdown() -> None: + try: + if pre_uploader is not None: + pre_uploader.shutdown() + except Exception: # pylint: disable=broad-except + _logger.debug("Failed to shutdown retired pre-uploader", exc_info=True) + try: + if uploader is not None: + uploader.shutdown() + except Exception: # pylint: disable=broad-except + _logger.debug("Failed to shutdown retired uploader", exc_info=True) + + thread = threading.Thread( + target=_shutdown, + name="multimodal-uploader-retire", + daemon=True, + ) + thread.start() + + +def load_uploader_hook( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> Optional[Uploader]: """Load multimodal uploader hook from entry points. Mechanism: - - read hook name from env var - `OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER` + - read hook name from runtime snapshot + (`otel.instrumentation.genai.multimodal.uploader`, default: ``fs``) - resolve hook factory from entry-point group - `opentelemetry_genai_multimodal_uploader` - - call zero-arg hook factory to build uploader instance - - validate returned object type (`Uploader`) + ``opentelemetry_genai_multimodal_uploader`` + - call hook factory with snapshot (legacy zero-arg hooks still supported) + - validate returned object type (``Uploader``) """ - upload_mode = get_multimodal_upload_mode() - if upload_mode == _UPLOAD_MODE_NONE: + cfg = snapshot or get_multimodal_config_snapshot() + if cfg.upload_mode == _UPLOAD_MODE_NONE: return None - hook_name = get_multimodal_uploader_hook_name() + hook_name = cfg.uploader_hook_name or None if not hook_name: return None uploader = _load_by_name( - hook_name=hook_name, group=_MULTIMODAL_UPLOADER_ENTRY_POINT_GROUP + hook_name=hook_name, + group=_MULTIMODAL_UPLOADER_ENTRY_POINT_GROUP, + snapshot=cfg, ) if uploader is None: return None @@ -109,28 +167,30 @@ def load_uploader_hook() -> Optional[Uploader]: return uploader -def load_pre_uploader_hook() -> Optional[PreUploader]: +def load_pre_uploader_hook( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> Optional[PreUploader]: """Load multimodal pre-uploader hook from entry points. Mechanism: - - read hook name from env var - `OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER` - (default: `fs`) + - read hook name from runtime snapshot + (`otel.instrumentation.genai.multimodal.pre-uploader`, default: ``fs``) - resolve hook factory from entry-point group - `opentelemetry_genai_multimodal_pre_uploader` - - call zero-arg hook factory to build pre-uploader instance - - validate returned object type (`PreUploader`) + ``opentelemetry_genai_multimodal_pre_uploader`` + - call hook factory with snapshot (legacy zero-arg hooks still supported) + - validate returned object type (``PreUploader``) """ - upload_mode = get_multimodal_upload_mode() - if upload_mode == _UPLOAD_MODE_NONE: + cfg = snapshot or get_multimodal_config_snapshot() + if cfg.upload_mode == _UPLOAD_MODE_NONE: return None - hook_name = get_multimodal_pre_uploader_hook_name() + hook_name = cfg.pre_uploader_hook_name or None if not hook_name: return None pre_uploader = _load_by_name( hook_name=hook_name, group=_MULTIMODAL_PRE_UPLOADER_ENTRY_POINT_GROUP, + snapshot=cfg, ) if pre_uploader is None: return None @@ -141,26 +201,78 @@ def load_pre_uploader_hook() -> Optional[PreUploader]: return pre_uploader -def get_or_load_uploader_pair() -> tuple[ - Optional[Uploader], Optional[PreUploader] -]: - """Get lazily loaded singleton uploader/pre-uploader pair. +def _load_pair_from_snapshot( + snapshot: MultimodalConfigSnapshot, +) -> tuple[Optional[Uploader], Optional[PreUploader]]: + uploader = load_uploader_hook(snapshot) + pre_uploader = load_pre_uploader_hook(snapshot) + if uploader is None or pre_uploader is None: + return None, None + return uploader, pre_uploader - First call performs one-time loading; subsequent calls return cache. - If either side fails to load, both are downgraded to `(None, None)`. - """ - def _load() -> None: - global _uploader # pylint: disable=global-statement - global _pre_uploader # pylint: disable=global-statement - _uploader = load_uploader_hook() - _pre_uploader = load_pre_uploader_hook() - if _uploader is None or _pre_uploader is None: - _uploader = None - _pre_uploader = None +def get_or_rebuild_uploader_pair( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> tuple[Optional[Uploader], Optional[PreUploader]]: + """Get or rebuild uploader/pre-uploader pair for the current generation. - _load_once.do_once(_load) - return _uploader, _pre_uploader + Generation-aware hot reload: + - cache hit: return cached pair when ``uploader_generation`` unchanged + - cache miss: load a new pair outside the lock, then install atomically + - in-flight build: concurrent callers for the same generation get ``(None, None)`` + - stale build: if generation advanced during load, discard the new pair + - failed generation: remember load failure and skip retry for that generation + """ + global _uploader # pylint: disable=global-statement + global _pre_uploader # pylint: disable=global-statement + global _uploader_generation # pylint: disable=global-statement + global _failed_generation # pylint: disable=global-statement + global _building_generation # pylint: disable=global-statement + + cfg = snapshot or get_multimodal_config_snapshot() + if cfg.upload_mode == _UPLOAD_MODE_NONE: + return None, None + + generation = cfg.uploader_generation + + with _uploader_pair_lock: + if _failed_generation == generation: + return None, None + if ( + _uploader_generation == generation + and _uploader is not None + and _pre_uploader is not None + ): + return _uploader, _pre_uploader + if _building_generation == generation: + return None, None + _building_generation = generation + + new_pair = _load_pair_from_snapshot(cfg) + + with _uploader_pair_lock: + _building_generation = -1 + + if get_multimodal_config_snapshot().uploader_generation != generation: + _schedule_retired_pair_shutdown(new_pair) + return None, None + + if not _is_complete_pair(new_pair): + _failed_generation = generation + return None, None + + old_pair = (_uploader, _pre_uploader) + _uploader, _pre_uploader = new_pair + _uploader_generation = generation + _schedule_retired_pair_shutdown(old_pair) + return _uploader, _pre_uploader + + +def get_or_load_uploader_pair( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> tuple[Optional[Uploader], Optional[PreUploader]]: + """Backward-compatible alias for generation-aware pair loading.""" + return get_or_rebuild_uploader_pair(snapshot) def get_uploader_pair() -> tuple[Optional[Uploader], Optional[PreUploader]]: @@ -168,14 +280,18 @@ def get_uploader_pair() -> tuple[Optional[Uploader], Optional[PreUploader]]: return _uploader, _pre_uploader -def get_or_load_uploader() -> Optional[Uploader]: +def get_or_load_uploader( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> Optional[Uploader]: """Get uploader and trigger lazy loading when needed.""" - return get_or_load_uploader_pair()[0] + return get_or_rebuild_uploader_pair(snapshot)[0] -def get_or_load_pre_uploader() -> Optional[PreUploader]: +def get_or_load_pre_uploader( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> Optional[PreUploader]: """Get pre-uploader and trigger lazy loading when needed.""" - return get_or_load_uploader_pair()[1] + return get_or_rebuild_uploader_pair(snapshot)[1] def get_uploader() -> Optional[Uploader]: diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py index 133e9beb9..80da50012 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py @@ -144,7 +144,7 @@ def __init__( self._active_tasks = 0 self._active_cond = threading.Condition() - # Read multimodal upload configuration (static config, read once only) + # Bootstrap strategy fields from runtime snapshot at construction time. self._process_input = should_process_multimodal_input() self._process_output = should_process_multimodal_output() self._download_enabled = is_multimodal_download_enabled() @@ -156,6 +156,9 @@ def __init__( # Local file configuration self._local_file_enabled = is_multimodal_local_file_enabled() self._allowed_root_paths = get_multimodal_allowed_root_paths() + self._applied_strategy_version: int = ( + self._read_applied_strategy_version() + ) if self._local_file_enabled and not self._allowed_root_paths: _logger.warning( @@ -1186,12 +1189,39 @@ def _collect_http_uris( return uris + @staticmethod + def _read_applied_strategy_version() -> int: + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + return get_multimodal_config_snapshot().strategy_version + + def _sync_strategy_from_snapshot(self, config_snapshot: Any) -> bool: + """Apply strategy fields when snapshot strategy_version changes.""" + if ( + config_snapshot.strategy_version + != self._applied_strategy_version + ): + self._process_input = config_snapshot.process_input + self._process_output = config_snapshot.process_output + self._download_enabled = config_snapshot.download_enabled + self._local_file_enabled = config_snapshot.local_file_enabled + self._allowed_root_paths = list( + config_snapshot.allowed_root_paths + ) + self._applied_strategy_version = ( + config_snapshot.strategy_version + ) + return self._process_input or self._process_output + def pre_upload( # pylint: disable=too-many-branches self, span_context: Optional[SpanContext], start_time_utc_nano: int, input_messages: Optional[List[Any]], output_messages: Optional[List[Any]], + config_snapshot: Optional[Any] = None, ) -> List[PreUploadItem]: """ Preprocess multimodal data in messages: @@ -1211,10 +1241,33 @@ def pre_upload( # pylint: disable=too-many-branches """ uploads: List[PreUploadItem] = [] - # If not processing either, return directly (use config read in __init__) - if not self._process_input and not self._process_output: + if config_snapshot is None: + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + config_snapshot = get_multimodal_config_snapshot() + + if not self._sync_strategy_from_snapshot(config_snapshot): return uploads + return self._pre_upload_with_current_config( + span_context, + start_time_utc_nano, + input_messages, + output_messages, + uploads, + ) + + def _pre_upload_with_current_config( + self, + span_context: Optional[SpanContext], + start_time_utc_nano: int, + input_messages: Optional[List[Any]], + output_messages: Optional[List[Any]], + uploads: List[PreUploadItem], + ) -> List[PreUploadItem]: + trace_id: Optional[str] = None span_id: Optional[str] = None try: @@ -1273,9 +1326,18 @@ def pre_upload( # pylint: disable=too-many-branches return uploads -def fs_pre_uploader_hook() -> Optional[PreUploader]: - """Create file-system pre-uploader from environment variables.""" - base_path = get_multimodal_storage_base_path() +def fs_pre_uploader_hook( + snapshot: Optional[Any] = None, +) -> Optional[PreUploader]: + """Create file-system pre-uploader from runtime snapshot.""" + if snapshot is None: + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + snapshot = get_multimodal_config_snapshot() + + base_path = snapshot.effective_storage_base_path or snapshot.storage_base_path if not base_path: _logger.warning( "%s is required but not set, multimodal pre-uploader disabled", diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py index 6cd446d56..eb0363624 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py @@ -15,7 +15,6 @@ import json import logging import os -import re # LoongSuite Extension from base64 import b64encode from functools import partial from typing import Any, List, Optional @@ -30,15 +29,8 @@ OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, ) from opentelemetry.util.genai.extended_environment_variables import ( # pylint: disable=no-name-in-module - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_ALLOWED_ROOT_PATHS, # LoongSuite Extension OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_AUDIO_CONVERSION_ENABLED, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED, # LoongSuite Extension OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_SSL_VERIFY, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, # LoongSuite Extension - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, # LoongSuite Extension ) from opentelemetry.util.genai.types import ContentCapturingMode @@ -143,10 +135,16 @@ def _parse_env_bool(value: Optional[str], default: bool) -> bool: return value.lower() in ("true", "1", "yes") +def _get_multimodal_snapshot(): + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + get_multimodal_config_snapshot, + ) + + return get_multimodal_config_snapshot() + + def get_multimodal_upload_mode() -> str: - return os.getenv( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, "none" - ).lower() + return _get_multimodal_snapshot().upload_mode def should_process_multimodal_input() -> bool: @@ -158,10 +156,7 @@ def should_process_multimodal_output() -> bool: def is_multimodal_download_enabled() -> bool: - return _parse_env_bool( - os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED), - default=False, - ) + return _get_multimodal_snapshot().download_enabled def should_verify_multimodal_download_ssl() -> bool: @@ -174,10 +169,7 @@ def should_verify_multimodal_download_ssl() -> bool: def is_multimodal_local_file_enabled() -> bool: - return _parse_env_bool( - os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED), - default=False, - ) + return _get_multimodal_snapshot().local_file_enabled def is_multimodal_audio_conversion_enabled() -> bool: @@ -190,33 +182,22 @@ def is_multimodal_audio_conversion_enabled() -> bool: def get_multimodal_allowed_root_paths() -> List[str]: - allowed_roots_str = os.getenv( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_ALLOWED_ROOT_PATHS, - "", - ) - if not allowed_roots_str: - return [] - - paths = [ - p.strip() for p in re.split(r"[,]", allowed_roots_str) if p.strip() - ] - return [os.path.abspath(p) for p in paths] + return list(_get_multimodal_snapshot().allowed_root_paths) def get_multimodal_uploader_hook_name() -> Optional[str]: - hook_name = os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, "fs") + hook_name = _get_multimodal_snapshot().uploader_hook_name return hook_name or None def get_multimodal_pre_uploader_hook_name() -> Optional[str]: - hook_name = os.getenv( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, "fs" - ) + hook_name = _get_multimodal_snapshot().pre_uploader_hook_name return hook_name or None def get_multimodal_storage_base_path() -> Optional[str]: - return os.getenv(OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH) + snapshot = _get_multimodal_snapshot() + return snapshot.effective_storage_base_path or snapshot.storage_base_path class _GenAiJsonEncoder(json.JSONEncoder): diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/conftest.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/conftest.py new file mode 100644 index 000000000..7f056fbf9 --- /dev/null +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/conftest.py @@ -0,0 +1,38 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Shared pytest hooks for multimodal upload tests.""" + +# pyright: reportUnusedFunction=false + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest + +from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test + + +@pytest.fixture(autouse=True) +def _isolate_multimodal_runtime_state() -> Iterator[None]: + """Reset singleton state before/after each test. + + Tests that patch ``os.environ`` must call + ``reset_multimodal_runtime_state_for_test()`` again inside the patch + so the runtime snapshot is rebuilt from the patched env. + """ + reset_multimodal_runtime_state_for_test() + yield + reset_multimodal_runtime_state_for_test() diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/multimodal_test_helpers.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/multimodal_test_helpers.py new file mode 100644 index 000000000..e66015581 --- /dev/null +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/multimodal_test_helpers.py @@ -0,0 +1,68 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test helpers for resetting multimodal runtime singleton state.""" + +from __future__ import annotations + +import importlib +from types import ModuleType + +from opentelemetry.util.genai._multimodal_upload import ( # pylint: disable=no-name-in-module + config as multimodal_config, +) +from opentelemetry.util.genai._multimodal_upload import ( # pylint: disable=no-name-in-module + multimodal_upload_hook, +) + +MULTIMODAL_UPLOAD_HOOK_MODULE = ( + "opentelemetry.util.genai._multimodal_upload.multimodal_upload_hook" +) + + +def reset_multimodal_runtime_config_for_test(): + """Reset process-wide multimodal config snapshot from current env.""" + multimodal_config._runtime_config = ( # pylint: disable=protected-access + multimodal_config.MultimodalRuntimeConfig() + ) + return multimodal_config._runtime_config.get_snapshot() # pylint: disable=protected-access + + +def reset_multimodal_runtime_state_for_test() -> None: + """Reset multimodal config snapshot and cached uploader pair state.""" + reset_multimodal_runtime_config_for_test() + hook = multimodal_upload_hook + with hook._uploader_pair_lock: # pylint: disable=protected-access + hook._uploader = None # pylint: disable=protected-access + hook._pre_uploader = None # pylint: disable=protected-access + hook._uploader_generation = -1 # pylint: disable=protected-access + hook._failed_generation = -1 # pylint: disable=protected-access + hook._building_generation = -1 # pylint: disable=protected-access + + +def get_default_uploader_hook_name() -> str: + """Return bootstrap default uploader hook (community ``fs``, enterprise ``arms``).""" + return multimodal_config.DEFAULT_MULTIMODAL_UPLOADER_HOOK + + +def get_default_pre_uploader_hook_name() -> str: + """Return bootstrap default pre-uploader hook (community ``fs``, enterprise ``arms``).""" + return multimodal_config.DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK + + +def reload_multimodal_upload_hook_module() -> ModuleType: + """Reset runtime state and reload the hook module under current env.""" + reset_multimodal_runtime_state_for_test() + module = importlib.import_module(MULTIMODAL_UPLOAD_HOOK_MODULE) + return importlib.reload(module) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py index 8e447bf3f..332e501df 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py @@ -24,6 +24,7 @@ MultimodalPreUploader, fs_pre_uploader_hook, ) +from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test from opentelemetry.util.genai.extended_environment_variables import ( OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, ) @@ -50,8 +51,10 @@ def test_fs_uploader_hook_returns_none_without_base_path(self): clear=True, ) def test_fs_uploader_hook_returns_uploader(self): + reset_multimodal_runtime_state_for_test() uploader = fs_uploader_hook() self.assertIsInstance(uploader, FsUploader) + assert uploader is not None uploader.shutdown(timeout=0.1) @patch.dict("os.environ", {}, clear=True) @@ -74,6 +77,8 @@ def test_fs_pre_uploader_hook_returns_none_without_base_path(self): clear=True, ) def test_fs_pre_uploader_hook_returns_pre_uploader(self): + reset_multimodal_runtime_state_for_test() pre_uploader = fs_pre_uploader_hook() self.assertIsInstance(pre_uploader, MultimodalPreUploader) + assert pre_uploader is not None pre_uploader.shutdown(timeout=0.1) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py new file mode 100644 index 000000000..9f150b45c --- /dev/null +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py @@ -0,0 +1,290 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import os +import tempfile +import unittest +from dataclasses import dataclass +from typing import Any, Callable, List, Optional +from unittest.mock import patch + +from opentelemetry.trace import SpanContext +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + DEFAULT_SLS_LOGSTORE, + UPLOADER_GENERATION_FIELDS, + get_multimodal_config_snapshot, + normalize_multimodal_hook_name, + update_multimodal_runtime_config, +) +from opentelemetry.util.genai.extended_environment_variables import ( # pylint: disable=no-name-in-module + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, +) + +from .multimodal_test_helpers import ( + get_default_pre_uploader_hook_name, + get_default_uploader_hook_name, + reload_multimodal_upload_hook_module, + reset_multimodal_runtime_state_for_test, +) + + +@dataclass +class FakeEntryPoint: + name: str + load: Callable[[], Callable[..., Any]] + + +class TestMultimodalRuntimeConfig(unittest.TestCase): + @patch.dict("os.environ", {}, clear=True) + def test_env_fallback_defaults(self) -> None: + reset_multimodal_runtime_state_for_test() + snapshot = get_multimodal_config_snapshot() + self.assertEqual(snapshot.upload_mode, "none") + self.assertFalse(snapshot.download_enabled) + self.assertEqual( + snapshot.uploader_hook_name, get_default_uploader_hook_name() + ) + self.assertEqual( + snapshot.pre_uploader_hook_name, get_default_pre_uploader_hook_name() + ) + self.assertIsNone(snapshot.effective_storage_base_path) + self.assertFalse(snapshot.process_input) + self.assertFalse(snapshot.process_output) + + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED: "true", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "arms", + }, + clear=True, + ) + def test_env_arms_hook_builds_sls_effective_path(self) -> None: + reset_multimodal_runtime_state_for_test() + snapshot = get_multimodal_config_snapshot() + self.assertEqual(snapshot.uploader_hook_name, "arms") + self.assertTrue(snapshot.process_input) + self.assertTrue(snapshot.process_output) + self.assertTrue(snapshot.download_enabled) + self.assertIsNone(snapshot.effective_storage_base_path) + + def test_upload_mode_input_and_output_flags(self) -> None: + input_only = update_multimodal_runtime_config(upload_mode="input") + self.assertTrue(input_only.process_input) + self.assertFalse(input_only.process_output) + + output_only = update_multimodal_runtime_config(upload_mode="output") + self.assertFalse(output_only.process_input) + self.assertTrue(output_only.process_output) + + def test_arms_default_logstore_when_project_only(self) -> None: + after = update_multimodal_runtime_config( + uploader_hook_name="arms", + sls_project="project-a", + ) + self.assertIsNone(after.sls_logstore) + self.assertEqual( + after.effective_storage_base_path, + f"sls://project-a/{DEFAULT_SLS_LOGSTORE}", + ) + + def test_strategy_update_does_not_bump_uploader_generation(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config( + upload_mode="input", + download_enabled=True, + local_file_enabled="", + ) + self.assertGreater(after.version, before.version) + self.assertGreater(after.strategy_version, before.strategy_version) + self.assertEqual(after.uploader_generation, before.uploader_generation) + self.assertFalse(after.local_file_enabled) + + def test_uploader_generation_fields_bump_generation(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config( + storage_base_path="oss://bucket/prefix", + uploader_hook_name="fs", + pre_uploader_hook_name="fs", + ) + self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertEqual(after.storage_base_path, "oss://bucket/prefix") + self.assertEqual(after.effective_storage_base_path, "oss://bucket/prefix") + + def test_sls_project_change_bumps_uploader_generation(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config( + uploader_hook_name="arms", + pre_uploader_hook_name="arms", + sls_project="project-a", + sls_logstore="logstore-a", + ) + self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertEqual(after.uploader_hook_name, "arms") + self.assertEqual( + after.effective_storage_base_path, + "sls://project-a/logstore-a", + ) + + def test_logstore_change_bumps_uploader_generation(self) -> None: + update_multimodal_runtime_config( + uploader_hook_name="arms", + pre_uploader_hook_name="arms", + sls_project="p", + sls_logstore="l1", + ) + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config(sls_logstore="l2") + self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertIn("sls_logstore", UPLOADER_GENERATION_FIELDS) + self.assertEqual(after.effective_storage_base_path, "sls://p/l2") + + def test_invalid_hook_name_is_rejected(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config(uploader_hook_name="invalid") + self.assertEqual(after, before) + + def test_invalid_pre_uploader_hook_is_rejected(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config( + pre_uploader_hook_name="invalid" + ) + self.assertEqual(after, before) + + def test_invalid_upload_mode_fallback_to_none(self) -> None: + after = update_multimodal_runtime_config(upload_mode="invalid-mode") + self.assertEqual(after.upload_mode, "none") + + def test_empty_upload_mode_fallback_to_none(self) -> None: + after = update_multimodal_runtime_config(upload_mode="") + self.assertEqual(after.upload_mode, "none") + + def test_no_op_update_keeps_snapshot(self) -> None: + before = get_multimodal_config_snapshot() + after = update_multimodal_runtime_config(upload_mode=before.upload_mode) + self.assertEqual(after, before) + + def test_blank_sls_project_is_coalesced_to_none(self) -> None: + update_multimodal_runtime_config( + uploader_hook_name="arms", + sls_project="project-a", + ) + after = update_multimodal_runtime_config(sls_project=" ") + self.assertIsNone(after.sls_project) + + def test_allowed_root_paths_normalized_to_absolute_tuple(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + after = update_multimodal_runtime_config( + allowed_root_paths=f"{tmpdir}, {tmpdir}" + ) + self.assertEqual(len(after.allowed_root_paths), 1) + self.assertTrue(os.path.isabs(after.allowed_root_paths[0])) + + def test_allowed_root_paths_accepts_sequence(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + after = update_multimodal_runtime_config( + allowed_root_paths=[tmpdir, tmpdir] + ) + self.assertEqual(len(after.allowed_root_paths), 1) + + def test_normalize_multimodal_hook_name_none(self) -> None: + self.assertIsNone(normalize_multimodal_hook_name(None)) + + @patch.dict("os.environ", {}, clear=True) + def test_none_to_both_can_reload_uploader_pair(self) -> None: + module = reload_multimodal_upload_hook_module() + + uploader, pre_uploader = module.get_or_rebuild_uploader_pair() + self.assertIsNone(uploader) + self.assertIsNone(pre_uploader) + + with patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH: "file:///tmp/mm", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + }, + clear=False, + ): + module = reload_multimodal_upload_hook_module() + + from opentelemetry.util.genai._multimodal_upload._base import ( # pylint: disable=no-name-in-module + PreUploader, + PreUploadItem, + Uploader, + UploadItem, + ) + + class FakeUploader(Uploader): + def upload( + self, + item: UploadItem, + *, + skip_if_exists: bool = True, + ) -> bool: + return True + + def shutdown(self, timeout: float = 10.0) -> None: + return None + + class FakePreUploader(PreUploader): + def pre_upload( + self, + span_context: Optional[SpanContext], + start_time_utc_nano: int, + input_messages: Optional[List[Any]], + output_messages: Optional[List[Any]], + config_snapshot: Optional[Any] = None, + ) -> List[PreUploadItem]: + return [] + + def fake_entry_points(group: str) -> List[FakeEntryPoint]: + def uploader_loader() -> Callable[..., FakeUploader]: + def build(_snapshot: Any = None) -> FakeUploader: + return FakeUploader() + + return build + + def pre_uploader_loader() -> Callable[..., FakePreUploader]: + def build(_snapshot: Any = None) -> FakePreUploader: + return FakePreUploader() + + return build + + if group == "opentelemetry_genai_multimodal_uploader": + return [FakeEntryPoint("fs", uploader_loader)] + if group == "opentelemetry_genai_multimodal_pre_uploader": + return [FakeEntryPoint("fs", pre_uploader_loader)] + return [] + + with patch.object( + module, "_iter_entry_points", side_effect=fake_entry_points + ): + uploader, pre_uploader = module.get_or_rebuild_uploader_pair() + + self.assertIsNotNone(uploader) + self.assertIsNotNone(pre_uploader) + + +if __name__ == "__main__": + unittest.main() diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py index 4ad596eeb..c91431175 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py @@ -12,7 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -import importlib +# Aliyun Python Agent Extension +from __future__ import annotations + +import threading from dataclasses import dataclass from typing import Any, Callable, Optional from unittest import TestCase @@ -24,15 +27,18 @@ Uploader, UploadItem, ) +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + MultimodalConfigSnapshot, + update_multimodal_runtime_config, +) from opentelemetry.util.genai.extended_environment_variables import ( # pylint: disable=no-name-in-module OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE, OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, ) -HOOK_MODULE = ( - "opentelemetry.util.genai._multimodal_upload.multimodal_upload_hook" -) +from .multimodal_test_helpers import reload_multimodal_upload_hook_module class FakeUploader(Uploader): @@ -50,6 +56,7 @@ def pre_upload( start_time_utc_nano: int, input_messages: Optional[list[Any]], output_messages: Optional[list[Any]], + config_snapshot: Optional[Any] = None, ) -> list[PreUploadItem]: return [] @@ -65,14 +72,9 @@ class FakeEntryPoint: class TestMultimodalUploadHook(TestCase): - @staticmethod - def _reload_module(): - module = importlib.import_module(HOOK_MODULE) - return importlib.reload(module) - @patch.dict("os.environ", {}, clear=True) def test_get_or_load_without_uploader_env(self): - module = self._reload_module() + module = reload_multimodal_upload_hook_module() uploader, pre_uploader = module.get_or_load_uploader_pair() self.assertIsNone(uploader) self.assertIsNone(pre_uploader) @@ -87,7 +89,7 @@ def test_get_or_load_without_uploader_env(self): clear=True, ) def test_getters_do_not_trigger_loading(self): - module = self._reload_module() + module = reload_multimodal_upload_hook_module() with patch.object(module, "_iter_entry_points") as mock_iter: self.assertIsNone(module.get_uploader()) self.assertIsNone(module.get_pre_uploader()) @@ -104,7 +106,7 @@ def test_getters_do_not_trigger_loading(self): clear=True, ) def test_load_hooks_success(self): - module = self._reload_module() + module = reload_multimodal_upload_hook_module() calls = {"uploader": 0, "pre": 0} def uploader_hook(): @@ -115,7 +117,7 @@ def pre_hook(): calls["pre"] += 1 return FakePreUploader() - def fake_entry_points(group: str): + def fake_entry_points(group: str) -> list[FakeEntryPoint]: if group == "opentelemetry_genai_multimodal_uploader": return [FakeEntryPoint("fs", lambda: uploader_hook)] if group == "opentelemetry_genai_multimodal_pre_uploader": @@ -145,8 +147,15 @@ def fake_entry_points(group: str): }, clear=True, ) - def test_load_uploader_and_pre_uploader_default_to_fs(self): - module = self._reload_module() + def test_load_uploader_and_pre_uploader_use_configured_defaults(self): + from .multimodal_test_helpers import ( + get_default_pre_uploader_hook_name, + get_default_uploader_hook_name, + ) + + module = reload_multimodal_upload_hook_module() + default_uploader = get_default_uploader_hook_name() + default_pre_uploader = get_default_pre_uploader_hook_name() def uploader_factory(): return FakeUploader() @@ -160,11 +169,15 @@ def load_uploader_factory(): def load_pre_uploader_factory(): return pre_uploader_factory - def fake_entry_points(group: str): + def fake_entry_points(group: str) -> list[FakeEntryPoint]: if group == "opentelemetry_genai_multimodal_uploader": - return [FakeEntryPoint("fs", load_uploader_factory)] + return [FakeEntryPoint(default_uploader, load_uploader_factory)] if group == "opentelemetry_genai_multimodal_pre_uploader": - return [FakeEntryPoint("fs", load_pre_uploader_factory)] + return [ + FakeEntryPoint( + default_pre_uploader, load_pre_uploader_factory + ) + ] return [] with patch.object( @@ -184,7 +197,7 @@ def fake_entry_points(group: str): clear=True, ) def test_invalid_hook_result_fallback(self): - module = self._reload_module() + module = reload_multimodal_upload_hook_module() def invalid_factory(): return InvalidHookResult() @@ -198,7 +211,7 @@ def load_invalid_factory(): def load_pre_uploader_factory(): return pre_uploader_factory - def fake_entry_points(group: str): + def fake_entry_points(group: str) -> list[FakeEntryPoint]: if group == "opentelemetry_genai_multimodal_uploader": return [FakeEntryPoint("fs", load_invalid_factory)] if group == "opentelemetry_genai_multimodal_pre_uploader": @@ -222,10 +235,207 @@ def fake_entry_points(group: str): clear=True, ) def test_upload_mode_none_disables_hooks(self): - module = self._reload_module() + module = reload_multimodal_upload_hook_module() with patch.object(module, "_iter_entry_points") as mock_iter: uploader, pre_uploader = module.get_or_load_uploader_pair() self.assertIsNone(uploader) self.assertIsNone(pre_uploader) mock_iter.assert_not_called() + + +def _fake_fs_entry_points( + uploader_factory: Callable[[], Uploader], + pre_uploader_factory: Callable[[], PreUploader], +) -> Callable[[str], list[FakeEntryPoint]]: + def fake_entry_points(group: str) -> list[FakeEntryPoint]: + if group == "opentelemetry_genai_multimodal_uploader": + return [FakeEntryPoint("fs", lambda: uploader_factory)] + if group == "opentelemetry_genai_multimodal_pre_uploader": + return [FakeEntryPoint("fs", lambda: pre_uploader_factory)] + return [] + + return fake_entry_points + + +class TestUploaderPairHotReload(TestCase): + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH: "file:///tmp/mm", + }, + clear=True, + ) + def test_generation_bump_rebuilds_uploader_pair(self): + module = reload_multimodal_upload_hook_module() + created_uploaders: list[FakeUploader] = [] + + def uploader_factory(): + uploader = FakeUploader() + created_uploaders.append(uploader) + return uploader + + with patch.object( + module, + "_iter_entry_points", + side_effect=_fake_fs_entry_points( + uploader_factory, lambda: FakePreUploader() + ), + ): + first_uploader, _ = module.get_or_rebuild_uploader_pair() + update_multimodal_runtime_config( + storage_base_path="file:///tmp/mm-v2" + ) + second_uploader, _ = module.get_or_rebuild_uploader_pair() + + self.assertIsNotNone(first_uploader) + self.assertIsNotNone(second_uploader) + self.assertIsNot(first_uploader, second_uploader) + self.assertEqual(len(created_uploaders), 2) + + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH: "file:///tmp/mm", + }, + clear=True, + ) + def test_building_same_generation_returns_none_until_complete(self): + module = reload_multimodal_upload_hook_module() + load_started = threading.Event() + release_load = threading.Event() + + def slow_load_pair(_snapshot: MultimodalConfigSnapshot): + load_started.set() + release_load.wait(timeout=5) + return FakeUploader(), FakePreUploader() + + with patch.object( + module, "_load_pair_from_snapshot", side_effect=slow_load_pair + ): + builder = threading.Thread( + target=module.get_or_rebuild_uploader_pair, + daemon=True, + ) + builder.start() + self.assertTrue(load_started.wait(timeout=5)) + + concurrent_uploader, concurrent_pre = ( + module.get_or_rebuild_uploader_pair() + ) + self.assertIsNone(concurrent_uploader) + self.assertIsNone(concurrent_pre) + + release_load.set() + builder.join(timeout=5) + + cached_uploader, cached_pre = module.get_or_rebuild_uploader_pair() + self.assertIsInstance(cached_uploader, FakeUploader) + self.assertIsInstance(cached_pre, FakePreUploader) + + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH: "file:///tmp/mm", + }, + clear=True, + ) + def test_stale_build_discarded_when_generation_advances_during_load(self): + module = reload_multimodal_upload_hook_module() + shutdown_calls = {"count": 0} + original_schedule = module._schedule_retired_pair_shutdown + + def track_shutdown( + pair: tuple[Optional[Uploader], Optional[PreUploader]], + ) -> None: + shutdown_calls["count"] += 1 + original_schedule(pair) + + def load_pair_and_bump_generation( + _snapshot: MultimodalConfigSnapshot, + ): + update_multimodal_runtime_config( + storage_base_path="file:///tmp/mm-new" + ) + return FakeUploader(), FakePreUploader() + + with patch.object( + module, + "_load_pair_from_snapshot", + side_effect=load_pair_and_bump_generation, + ), patch.object( + module, + "_schedule_retired_pair_shutdown", + side_effect=track_shutdown, + ): + uploader, pre_uploader = module.get_or_rebuild_uploader_pair() + + self.assertIsNone(uploader) + self.assertIsNone(pre_uploader) + self.assertEqual(shutdown_calls["count"], 1) + self.assertIsNone(module.get_uploader()) + self.assertIsNone(module.get_pre_uploader()) + + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + }, + clear=True, + ) + def test_failed_generation_is_not_retried(self): + module = reload_multimodal_upload_hook_module() + load_calls = {"count": 0} + + def counting_load_pair(_snapshot: MultimodalConfigSnapshot): + load_calls["count"] += 1 + return None, None + + with patch.object( + module, "_load_pair_from_snapshot", side_effect=counting_load_pair + ): + first = module.get_or_rebuild_uploader_pair() + second = module.get_or_rebuild_uploader_pair() + + self.assertEqual(first, (None, None)) + self.assertEqual(second, (None, None)) + self.assertEqual(load_calls["count"], 1) + + @patch.dict( + "os.environ", + { + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE: "both", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER: "fs", + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH: "file:///tmp/mm", + }, + clear=True, + ) + def test_same_generation_keeps_cached_pair(self): + module = reload_multimodal_upload_hook_module() + load_calls = {"count": 0} + + def counting_load_pair(_snapshot: MultimodalConfigSnapshot): + load_calls["count"] += 1 + return FakeUploader(), FakePreUploader() + + with patch.object( + module, "_load_pair_from_snapshot", side_effect=counting_load_pair + ): + first = module.get_or_rebuild_uploader_pair() + second = module.get_or_rebuild_uploader_pair() + + self.assertIs(first[0], second[0]) + self.assertIs(first[1], second[1]) + self.assertEqual(load_calls["count"], 1) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py index d22896fdd..afac42910 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py @@ -39,6 +39,8 @@ ) from opentelemetry.util.genai.types import Base64Blob, Blob, InputMessage, Uri +from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test + # Test audio file directory for integration tests TEST_AUDIO_DIR = Path(__file__).parent / "test_audio_samples" @@ -52,6 +54,7 @@ def _default_upload_mode_enabled_for_tests(): "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED": "true", }, ): + reset_multimodal_runtime_state_for_test() yield @@ -854,6 +857,7 @@ def test_upload_mode_none_skips_all(): "os.environ", {"OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE": "none"}, ): + reset_multimodal_runtime_state_for_test() pre_uploader = MultimodalPreUploader("/tmp/test") assert pre_uploader._process_input is False assert pre_uploader._process_output is False @@ -881,6 +885,7 @@ def test_upload_mode_input_only(): "os.environ", {"OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE": "input"}, ): + reset_multimodal_runtime_state_for_test() pre_uploader = MultimodalPreUploader("/tmp/test") assert pre_uploader._process_input is True assert pre_uploader._process_output is False @@ -892,6 +897,7 @@ def test_upload_mode_output_only(): "os.environ", {"OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE": "output"}, ): + reset_multimodal_runtime_state_for_test() pre_uploader = MultimodalPreUploader("/tmp/test") assert pre_uploader._process_input is False assert pre_uploader._process_output is True @@ -905,6 +911,7 @@ def test_download_disabled_skips_uri(): "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED": "false" }, ): + reset_multimodal_runtime_state_for_test() pre_uploader = MultimodalPreUploader("/tmp/test") assert pre_uploader._download_enabled is False @@ -960,6 +967,7 @@ def test_download_enabled_fetch_failed_uri_kept_in_metadata(mock_fetch): "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_DOWNLOAD_ENABLED": "true", }, ): + reset_multimodal_runtime_state_for_test() mock_fetch.return_value = {} pre_uploader = MultimodalPreUploader("/tmp/test") @@ -1110,6 +1118,7 @@ def test_local_file_processing_allowed(pre_uploader_factory): ), }, ): + reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() part = Uri(modality="image", mime_type="image/png", uri=file_uri) message = InputMessage(role="user", parts=[part]) @@ -1143,8 +1152,8 @@ def test_local_file_processing_relative_path(pre_uploader_factory): # OR we try to pass a relative path if we can force os.getcwd() to match. relative_path = test_file.name - with ( - patch.dict( + # Aliyun Python Agent Extension: for compatibility with Python 3.8 + with patch.dict( os.environ, { "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED": "true", @@ -1152,9 +1161,9 @@ def test_local_file_processing_relative_path(pre_uploader_factory): test_dir ), }, - ), - patch("os.getcwd", return_value=str(test_dir)), - ): + ), \ + patch("os.getcwd", return_value=str(test_dir)): + reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() # Test with simple filename (relative path) part = Uri( @@ -1207,6 +1216,7 @@ def test_local_file_processing_forbidden_path(pre_uploader_factory): "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_ALLOWED_ROOT_PATHS": allowed_root, }, ): + reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() file_uri = f"file://{test_file}" part = Uri(modality="image", mime_type="image/png", uri=file_uri) @@ -1254,6 +1264,7 @@ def test_local_file_processing_symlink_traversal_blocked( ), }, ): + reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() part = Uri(modality="image", mime_type="image/png", uri=file_uri) message = InputMessage(role="user", parts=[part]) @@ -1280,6 +1291,7 @@ def test_local_file_processing_no_allowed_roots(pre_uploader_factory): # No ALLOWED_ROOT_PATHS }, ): + reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() file_uri = f"file://{test_file}" part = Uri(modality="image", mime_type="image/png", uri=file_uri) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py index 385ba6836..df9983f04 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py @@ -33,6 +33,8 @@ ) from opentelemetry.util.genai.types import Blob, InputMessage +from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test + # Test audio file directory TEST_AUDIO_DIR = Path(__file__).parent / "test_audio_samples" @@ -47,6 +49,7 @@ def _default_upload_mode_enabled_for_tests(): "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_AUDIO_CONVERSION_ENABLED": "true", }, ): + reset_multimodal_runtime_state_for_test() yield @@ -232,19 +235,15 @@ def test_pcm16_conversion_missing_audio_libs_logs_single_warning( caplog, ): """Missing optional audio libs should only log the actual conversion skip.""" - with ( - patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader._audio_libs_available", - False, - ), - patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader.np", - None, - ), - patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader.sf", - None, - ), + with patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader._audio_libs_available", + False, + ), patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader.np", + None, + ), patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader.sf", + None, ): pre_uploader = MultimodalPreUploader(base_path="/tmp/test_upload") part = Blob( @@ -285,6 +284,7 @@ def test_pcm16_conversion_disabled_by_default(): }, clear=True, ): + reset_multimodal_runtime_state_for_test() pre_uploader = MultimodalPreUploader(base_path="/tmp/test_upload") pcm_data = b"\x00\x01" * 1000 part = Blob( From 6c850e2d91ad92bfcfd7ba8e08583dc1454223d0 Mon Sep 17 00:00:00 2001 From: xiongyun Date: Wed, 22 Jul 2026 14:56:08 +0800 Subject: [PATCH 2/4] add changelog and doc --- .../CHANGELOG-loongsuite.md | 6 +++++ .../README-loongsuite.rst | 25 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/util/opentelemetry-util-genai/CHANGELOG-loongsuite.md b/util/opentelemetry-util-genai/CHANGELOG-loongsuite.md index 3812b57b4..ba3bc64cf 100644 --- a/util/opentelemetry-util-genai/CHANGELOG-loongsuite.md +++ b/util/opentelemetry-util-genai/CHANGELOG-loongsuite.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Added + +- Add multimodal runtime config snapshot and generation-aware uploader hot-reload + (`MultimodalRuntimeConfig`, `update_multimodal_runtime_config`, + `get_or_rebuild_uploader_pair`). + ## Version 0.7.0 (2026-07-03) ### Added diff --git a/util/opentelemetry-util-genai/README-loongsuite.rst b/util/opentelemetry-util-genai/README-loongsuite.rst index 3b2de9f80..57c7fdc1b 100644 --- a/util/opentelemetry-util-genai/README-loongsuite.rst +++ b/util/opentelemetry-util-genai/README-loongsuite.rst @@ -414,6 +414,31 @@ Framework 探针若已声明对本包的依赖，会随探针一并安装；单 **自定义上传实现**：通过 ``pyproject.toml`` 中的 entry point ``opentelemetry_genai_multimodal_uploader``、``opentelemetry_genai_multimodal_pre_uploader`` 注册实现；本仓库默认提供 ``fs`` hook（见包内 ``pyproject.toml``）。 +**运行时动态配置** + +进程内维护一份多模态配置 snapshot（``MultimodalRuntimeConfig`` / ``MultimodalConfigSnapshot``）。启动时仍由上述环境变量完成 bootstrap；之后可通过公开 API 热更新，无需重启进程： + +:: + + from opentelemetry.util.genai._multimodal_upload import ( + get_multimodal_config_snapshot, + update_multimodal_runtime_config, + ) + + update_multimodal_runtime_config( + upload_mode="both", + storage_base_path="file:///var/log/genai/multimodal", + uploader_hook_name="fs", + pre_uploader_hook_name="fs", + ) + snapshot = get_multimodal_config_snapshot() + +策略字段（如 ``upload_mode``、``download_enabled``、``local_file_enabled``、 +``allowed_root_paths``）变更会递增 ``strategy_version``，立即作用于后续请求； +构造字段（如 ``storage_base_path``、uploader/pre-uploader hook 名）变更会递增 +``uploader_generation``，并在下次上传前通过 ``get_or_rebuild_uploader_pair()`` +热重建 uploader/pre-uploader。 + ------------------------------------------------------------------------ 5. 补充说明 ------------------------------------------------------------------------ From 50c52a49d23d3c54dbcc76fea6bc5465da10d0c2 Mon Sep 17 00:00:00 2001 From: xiongyun Date: Thu, 23 Jul 2026 20:21:58 +0800 Subject: [PATCH 3/4] fix code and unit tests per review feedback --- .../util/genai/_multimodal_upload/config.py | 23 ++++- .../multimodal_upload_hook.py | 82 +++++++++++------ .../genai/_multimodal_upload/pre_uploader.py | 7 +- .../test_multimodal_runtime_config.py | 13 ++- .../test_multimodal_upload_hook.py | 16 ++-- .../tests/test_extended_handler.py | 92 +++++++++++++++++-- 6 files changed, 178 insertions(+), 55 deletions(-) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py index 169bc99b8..5eadae3db 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py @@ -71,6 +71,18 @@ } ) +# SLS connection/credential fields are env-only for security: runtime updates must +# not rotate credentials or endpoints through update_multimodal_runtime_config(). +_READ_ONLY_RUNTIME_FIELDS = frozenset( + { + "sls_endpoint", + "sls_auth_type", + "sls_access_key_id", + "sls_access_key_secret", + "sls_sts_token", + } +) + @dataclass(frozen=True) class MultimodalConfigSnapshot: @@ -256,7 +268,16 @@ def _normalize_and_validate( } for key, value in fields.items(): - if value is None or key not in merged: + if value is None: + continue + if key in _READ_ONLY_RUNTIME_FIELDS: + _logger.warning( + "Ignoring runtime update for read-only multimodal field %r " + "(SLS credentials/endpoint must come from environment)", + key, + ) + continue + if key not in merged: continue if key == "upload_mode": merged[key] = _normalize_upload_mode(value) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py index 88212e0b4..d36b49df2 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py @@ -14,6 +14,7 @@ from __future__ import annotations +import inspect import logging import threading from importlib import metadata @@ -74,9 +75,12 @@ def _call_hook( snapshot: MultimodalConfigSnapshot, ) -> Optional[object]: try: + params = inspect.signature(hook).parameters + except (TypeError, ValueError): + params = {} + if params: return hook(snapshot) - except TypeError: - return hook() + return hook() def _load_by_name( @@ -211,44 +215,36 @@ def _load_pair_from_snapshot( return uploader, pre_uploader -def get_or_rebuild_uploader_pair( - snapshot: Optional[MultimodalConfigSnapshot] = None, -) -> tuple[Optional[Uploader], Optional[PreUploader]]: - """Get or rebuild uploader/pre-uploader pair for the current generation. - - Generation-aware hot reload: - - cache hit: return cached pair when ``uploader_generation`` unchanged - - cache miss: load a new pair outside the lock, then install atomically - - in-flight build: concurrent callers for the same generation get ``(None, None)`` - - stale build: if generation advanced during load, discard the new pair - - failed generation: remember load failure and skip retry for that generation - """ - global _uploader # pylint: disable=global-statement - global _pre_uploader # pylint: disable=global-statement - global _uploader_generation # pylint: disable=global-statement - global _failed_generation # pylint: disable=global-statement +def _resolve_cached_uploader_pair( + generation: int, +) -> tuple[bool, tuple[Optional[Uploader], Optional[PreUploader]]]: + """Return (resolved, pair). When resolved is True, pair is the final result.""" global _building_generation # pylint: disable=global-statement - cfg = snapshot or get_multimodal_config_snapshot() - if cfg.upload_mode == _UPLOAD_MODE_NONE: - return None, None - - generation = cfg.uploader_generation - with _uploader_pair_lock: if _failed_generation == generation: - return None, None + return True, (None, None) if ( _uploader_generation == generation and _uploader is not None and _pre_uploader is not None ): - return _uploader, _pre_uploader + return True, (_uploader, _pre_uploader) if _building_generation == generation: - return None, None + return True, (None, None) _building_generation = generation + return False, (None, None) - new_pair = _load_pair_from_snapshot(cfg) + +def _commit_uploader_pair( + generation: int, + new_pair: tuple[Optional[Uploader], Optional[PreUploader]], +) -> tuple[Optional[Uploader], Optional[PreUploader]]: + global _uploader # pylint: disable=global-statement + global _pre_uploader # pylint: disable=global-statement + global _uploader_generation # pylint: disable=global-statement + global _failed_generation # pylint: disable=global-statement + global _building_generation # pylint: disable=global-statement with _uploader_pair_lock: _building_generation = -1 @@ -259,6 +255,11 @@ def get_or_rebuild_uploader_pair( if not _is_complete_pair(new_pair): _failed_generation = generation + _logger.warning( + "Multimodal uploader pair build failed for generation %s; " + "uploads disabled until generation changes", + generation, + ) return None, None old_pair = (_uploader, _pre_uploader) @@ -268,6 +269,31 @@ def get_or_rebuild_uploader_pair( return _uploader, _pre_uploader +def get_or_rebuild_uploader_pair( + snapshot: Optional[MultimodalConfigSnapshot] = None, +) -> tuple[Optional[Uploader], Optional[PreUploader]]: + """Get or rebuild uploader/pre-uploader pair for the current generation. + + Generation-aware hot reload: + - cache hit: return cached pair when ``uploader_generation`` unchanged + - cache miss: load a new pair outside the lock, then install atomically + - in-flight build: concurrent callers for the same generation get ``(None, None)`` + - stale build: if generation advanced during load, discard the new pair + - failed generation: remember load failure and skip retry for that generation + """ + cfg = snapshot or get_multimodal_config_snapshot() + if cfg.upload_mode == _UPLOAD_MODE_NONE: + return None, None + + generation = cfg.uploader_generation + resolved, pair = _resolve_cached_uploader_pair(generation) + if resolved: + return pair + + new_pair = _load_pair_from_snapshot(cfg) + return _commit_uploader_pair(generation, new_pair) + + def get_or_load_uploader_pair( snapshot: Optional[MultimodalConfigSnapshot] = None, ) -> tuple[Optional[Uploader], Optional[PreUploader]]: diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py index 80da50012..5f45007ed 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py @@ -56,7 +56,6 @@ from opentelemetry.util.genai.types import Base64Blob, Blob, Modality, Uri from opentelemetry.util.genai.utils import ( # pylint: disable=no-name-in-module get_multimodal_allowed_root_paths, - get_multimodal_storage_base_path, is_multimodal_audio_conversion_enabled, is_multimodal_download_enabled, is_multimodal_local_file_enabled, @@ -67,8 +66,8 @@ # Try importing audio processing libraries (optional dependencies) try: - import numpy as np # pyright: ignore[reportMissingImports] - import soundfile as sf # pyright: ignore[reportMissingImports] + import numpy as np + import soundfile as sf _audio_libs_available = True except ImportError: @@ -592,7 +591,7 @@ def _convert_pcm16_to_wav( try: # Convert PCM16 byte data to numpy int16 array - audio_np = np.frombuffer(pcm_data, dtype=np.int16) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] + audio_np = np.frombuffer(pcm_data, dtype=np.int16) # pyright: ignore[reportUnknownMemberType] # Write WAV data to memory buffer buffer = io.BytesIO() diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py index 9f150b45c..64705c9b2 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py @@ -22,6 +22,12 @@ from unittest.mock import patch from opentelemetry.trace import SpanContext +from opentelemetry.util.genai._multimodal_upload._base import ( # pylint: disable=no-name-in-module + PreUploader, + PreUploadItem, + Uploader, + UploadItem, +) from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module DEFAULT_SLS_LOGSTORE, UPLOADER_GENERATION_FIELDS, @@ -228,13 +234,6 @@ def test_none_to_both_can_reload_uploader_pair(self) -> None: ): module = reload_multimodal_upload_hook_module() - from opentelemetry.util.genai._multimodal_upload._base import ( # pylint: disable=no-name-in-module - PreUploader, - PreUploadItem, - Uploader, - UploadItem, - ) - class FakeUploader(Uploader): def upload( self, diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py index c91431175..f8f51d219 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py @@ -38,7 +38,11 @@ OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, ) -from .multimodal_test_helpers import reload_multimodal_upload_hook_module +from .multimodal_test_helpers import ( + get_default_pre_uploader_hook_name, + get_default_uploader_hook_name, + reload_multimodal_upload_hook_module, +) class FakeUploader(Uploader): @@ -148,11 +152,6 @@ def fake_entry_points(group: str) -> list[FakeEntryPoint]: clear=True, ) def test_load_uploader_and_pre_uploader_use_configured_defaults(self): - from .multimodal_test_helpers import ( - get_default_pre_uploader_hook_name, - get_default_uploader_hook_name, - ) - module = reload_multimodal_upload_hook_module() default_uploader = get_default_uploader_hook_name() default_pre_uploader = get_default_pre_uploader_hook_name() @@ -278,11 +277,14 @@ def uploader_factory(): created_uploaders.append(uploader) return uploader + def pre_uploader_factory() -> FakePreUploader: + return FakePreUploader() + with patch.object( module, "_iter_entry_points", side_effect=_fake_fs_entry_points( - uploader_factory, lambda: FakePreUploader() + uploader_factory, pre_uploader_factory ), ): first_uploader, _ = module.get_or_rebuild_uploader_pair() diff --git a/util/opentelemetry-util-genai/tests/test_extended_handler.py b/util/opentelemetry-util-genai/tests/test_extended_handler.py index e279d9f43..b0be4562b 100644 --- a/util/opentelemetry-util-genai/tests/test_extended_handler.py +++ b/util/opentelemetry-util-genai/tests/test_extended_handler.py @@ -54,6 +54,13 @@ MultimodalProcessingMixin, _MultimodalAsyncTask, ) +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + update_multimodal_runtime_config, +) + +from ._multimodal_upload.multimodal_test_helpers import ( + reset_multimodal_runtime_state_for_test, +) from opentelemetry.util.genai.environment_variables import ( OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, @@ -1585,10 +1592,12 @@ def setUp(self): # Reset class-level state before each test MultimodalProcessingMixin._async_queue = None MultimodalProcessingMixin._async_worker = None + reset_multimodal_runtime_state_for_test() def tearDown(self): MultimodalProcessingMixin._async_queue = None MultimodalProcessingMixin._async_worker = None + reset_multimodal_runtime_state_for_test() @staticmethod def _create_mock_handler(enabled=True): @@ -1633,6 +1642,8 @@ def _create_invocation_with_multimodal(with_context=False): def test_quick_has_multimodal_orthogonal_cases(self): """Test _quick_has_multimodal with all multimodal types and edge cases.""" + # Detection respects upload_mode; enable both sides for this case matrix. + update_multimodal_runtime_config(upload_mode="both") mixin = MultimodalProcessingMixin # No multimodal: Text only @@ -1785,8 +1796,9 @@ def test_extract_multimodal_metadata_orthogonal(self): os.environ, {"OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE": "none"}, ) - def test_init_multimodal_disabled_when_mode_none(self): - """Test _init_multimodal with mode=none.""" + def test_init_multimodal_mode_none_gated_at_process_time(self): + """_init_multimodal stays enabled; upload_mode=none gates at process time.""" + reset_multimodal_runtime_state_for_test() class Handler(MultimodalProcessingMixin): def _get_uploader_and_pre_uploader(self): @@ -1794,31 +1806,93 @@ def _get_uploader_and_pre_uploader(self): handler = Handler() handler._init_multimodal() - self.assertFalse(handler._multimodal_enabled) + self.assertTrue(handler._multimodal_enabled) + + inv = self._create_invocation_with_multimodal(with_context=True) + self.assertFalse( + handler.process_multimodal_stop(inv, method="stop_llm") # pylint: disable=unexpected-keyword-arg + ) @patch_env_vars( "gen_ai_latest_experimental", content_capturing="SPAN_ONLY", OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOAD_MODE="both", ) - def test_init_multimodal_enabled_or_disabled_by_uploader(self): - """Test _init_multimodal enabled when uploader available, disabled when None.""" + def test_init_multimodal_always_enabled(self): + """_init_multimodal always enables; uploader availability is resolved later.""" + reset_multimodal_runtime_state_for_test() class HandlerWithUploader(MultimodalProcessingMixin): + def __init__(self): + self._logger = MagicMock() + def _get_uploader_and_pre_uploader(self): return MagicMock(), MagicMock() + def _record_llm_metrics(self, *_args: Any, **_kwargs: Any) -> None: + pass + h1 = HandlerWithUploader() h1._init_multimodal() self.assertTrue(h1._multimodal_enabled) class HandlerWithoutUploader(MultimodalProcessingMixin): + def __init__(self): + self._logger = MagicMock() + def _get_uploader_and_pre_uploader(self): return None, None + def _record_llm_metrics(self, *_args: Any, **_kwargs: Any) -> None: + pass + h2 = HandlerWithoutUploader() h2._init_multimodal() - self.assertFalse(h2._multimodal_enabled) + self.assertTrue(h2._multimodal_enabled) + + # Runtime entry is not gated by uploader availability. + inv = self._create_invocation_with_multimodal(with_context=True) + self.assertTrue(h1._should_async_process(inv)) + self.assertTrue(h2._should_async_process(inv)) + + # Runtime async path: with uploader pair → upload; without → skip + # upload but still finish the span. + mock_span = MagicMock() + mock_span._start_time = 1000000000 + async_inv = LLMInvocation(request_model="gpt-4") + async_inv.span = mock_span + async_inv.input_messages = inv.input_messages + + with patch.object( + h1, "_upload_and_set_metadata" + ) as mock_upload, patch( + "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" + ), patch( + "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + ): + h1._async_stop_llm( + _MultimodalAsyncTask( + invocation=async_inv, method="stop_llm", handler=h1 + ) + ) + mock_upload.assert_called_once() + mock_span.end.assert_called_once() + + mock_span.reset_mock() + with patch.object( + h2, "_upload_and_set_metadata" + ) as mock_upload2, patch( + "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" + ), patch( + "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + ): + h2._async_stop_llm( + _MultimodalAsyncTask( + invocation=async_inv, method="stop_llm", handler=h2 + ) + ) + mock_upload2.assert_not_called() + mock_span.end.assert_called_once() # ==================== process_multimodal_stop/fail Tests ==================== @@ -1871,6 +1945,7 @@ def test_process_multimodal_returns_false_on_precondition_failure(self): ) def test_process_multimodal_fallback_on_queue_issues(self): """Test process_multimodal_stop/fail uses fallback when queue is None or full.""" + reset_multimodal_runtime_state_for_test() handler = self._create_mock_handler() inv = self._create_invocation_with_multimodal(with_context=True) error = Error(message="err", type=RuntimeError) @@ -1915,6 +1990,7 @@ def test_process_multimodal_fallback_on_queue_issues(self): ) def test_process_multimodal_enqueues_task(self): """Test process_multimodal_stop/fail enqueues tasks correctly.""" + reset_multimodal_runtime_state_for_test() handler = self._create_mock_handler() error = Error(message="err", type=RuntimeError) @@ -2351,7 +2427,7 @@ class Handler(MultimodalProcessingMixin): inv = LLMInvocation(request_model="gpt-4") handler._separate_and_upload( - mock_span, inv, mock_uploader, mock_pre_uploader + mock_span, inv, mock_uploader, mock_pre_uploader, None ) mock_pre_uploader.pre_upload.assert_called_once() self.assertEqual(mock_uploader.upload.call_count, 2) @@ -2360,7 +2436,7 @@ class Handler(MultimodalProcessingMixin): mock_span2 = MagicMock() mock_span2.get_span_context.side_effect = RuntimeError("err") handler._separate_and_upload( - mock_span2, inv, mock_uploader, mock_pre_uploader + mock_span2, inv, mock_uploader, mock_pre_uploader, None ) # Should not raise From b597868815513f18c992da13ee6714b4442c9edd Mon Sep 17 00:00:00 2001 From: xiongyun Date: Fri, 24 Jul 2026 10:02:26 +0800 Subject: [PATCH 4/4] fix: satisfy precommit and typecheck for multimodal hot-reload --- util/opentelemetry-util-genai/CHANGELOG.md | 1 + .../util/genai/_multimodal_processing.py | 8 ++-- .../util/genai/_multimodal_upload/__init__.py | 8 ++-- .../util/genai/_multimodal_upload/config.py | 38 ++++++++++++------- .../genai/_multimodal_upload/fs_uploader.py | 6 ++- .../multimodal_upload_hook.py | 4 +- .../genai/_multimodal_upload/pre_uploader.py | 30 ++++++--------- .../src/opentelemetry/util/genai/utils.py | 2 +- .../_multimodal_upload/test_default_hooks.py | 3 +- .../test_multimodal_runtime_config.py | 23 ++++++++--- .../test_multimodal_upload_hook.py | 23 ++++++----- .../_multimodal_upload/test_pre_uploader.py | 8 ++-- .../test_pre_uploader_audio.py | 22 ++++++----- .../tests/test_extended_handler.py | 36 ++++++++++-------- 14 files changed, 125 insertions(+), 87 deletions(-) diff --git a/util/opentelemetry-util-genai/CHANGELOG.md b/util/opentelemetry-util-genai/CHANGELOG.md index 44716b83c..b46b0b80a 100644 --- a/util/opentelemetry-util-genai/CHANGELOG.md +++ b/util/opentelemetry-util-genai/CHANGELOG.md @@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Add multimodal runtime config snapshot and generation-aware uploader hot-reload. - Avoid import-time warnings when optional audio dependencies for PCM16-to-WAV conversion are not installed. ## Version 0.3b0 (2026-02-20) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py index 455489576..ac3ee240b 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_processing.py @@ -361,7 +361,7 @@ def _should_async_process(self, invocation: _MultimodalInvocation) -> bool: if not self._multimodal_enabled: return False - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) @@ -405,7 +405,7 @@ def _check_messages( @staticmethod def _quick_has_multimodal(invocation: _MultimodalInvocation) -> bool: """Quick detection of multimodal data (O(n), no network)""" - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) @@ -766,7 +766,7 @@ def _get_uploader_and_pre_uploader( # pylint: disable=no-self-use from opentelemetry.util.genai._multimodal_upload import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_or_rebuild_uploader_pair, ) - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) @@ -783,7 +783,7 @@ def _upload_and_set_metadata( pre_uploader: "PreUploader", ) -> None: """Upload multimodal data and set metadata attributes on span""" - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py index 38a4c6ad6..cef3b0b4a 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/__init__.py @@ -22,6 +22,10 @@ Uploader, UploadItem, ) +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + get_multimodal_config_snapshot, + update_multimodal_runtime_config, +) from opentelemetry.util.genai._multimodal_upload.multimodal_upload_hook import ( # pylint: disable=no-name-in-module get_or_load_pre_uploader, get_or_load_uploader, @@ -33,10 +37,6 @@ load_pre_uploader_hook, load_uploader_hook, ) -from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module - get_multimodal_config_snapshot, - update_multimodal_runtime_config, -) try: from opentelemetry.util.genai._multimodal_upload.fs_uploader import ( diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py index 5eadae3db..3d6bb5a0a 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py @@ -46,7 +46,9 @@ _APSARA_SLS_LOGSTORE_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_LOGSTORE" _APSARA_SLS_ENDPOINT_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ENDPOINT" _APSARA_SLS_AUTH_TYPE_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_AUTH_TYPE" -_APSARA_SLS_ACCESS_KEY_ID_ENV = "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ACCESS_KEY_ID" +_APSARA_SLS_ACCESS_KEY_ID_ENV = ( + "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ACCESS_KEY_ID" +) _APSARA_SLS_ACCESS_KEY_SECRET_ENV = ( "APSARA_APM_COLLECTOR_MULTIMODAL_SLS_ACCESS_KEY_SECRET" ) @@ -145,7 +147,9 @@ def _normalize_upload_mode(value: Optional[str]) -> str: return "none" mode = str(value).strip().lower() if mode not in _VALID_UPLOAD_MODES: - _logger.warning("Invalid multimodal upload_mode %r, fallback to none", value) + _logger.warning( + "Invalid multimodal upload_mode %r, fallback to none", value + ) return "none" return mode @@ -204,18 +208,24 @@ def _snapshot_from_env(*, version: int = 0) -> MultimodalConfigSnapshot: storage_base_path = os.getenv( OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH ) - uploader_hook_name = normalize_multimodal_hook_name( - os.getenv( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, - DEFAULT_MULTIMODAL_UPLOADER_HOOK, + uploader_hook_name = ( + normalize_multimodal_hook_name( + os.getenv( + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_UPLOADER, + DEFAULT_MULTIMODAL_UPLOADER_HOOK, + ) ) - ) or DEFAULT_MULTIMODAL_UPLOADER_HOOK - pre_uploader_hook_name = normalize_multimodal_hook_name( - os.getenv( - OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, - DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK, + or DEFAULT_MULTIMODAL_UPLOADER_HOOK + ) + pre_uploader_hook_name = ( + normalize_multimodal_hook_name( + os.getenv( + OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_PRE_UPLOADER, + DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK, + ) ) - ) or DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK + or DEFAULT_MULTIMODAL_PRE_UPLOADER_HOOK + ) return MultimodalConfigSnapshot( upload_mode=upload_mode, @@ -373,5 +383,7 @@ def get_multimodal_config_snapshot() -> MultimodalConfigSnapshot: return _runtime_config.get_snapshot() -def update_multimodal_runtime_config(**fields: Any) -> MultimodalConfigSnapshot: +def update_multimodal_runtime_config( + **fields: Any, +) -> MultimodalConfigSnapshot: return _runtime_config.update(**fields) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py index 4e4379740..439596a44 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/fs_uploader.py @@ -65,13 +65,15 @@ def fs_uploader_hook( ) -> Optional[Uploader]: """Create default FsUploader from runtime snapshot.""" if snapshot is None: - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) snapshot = get_multimodal_config_snapshot() - base_path = snapshot.effective_storage_base_path or snapshot.storage_base_path + base_path = ( + snapshot.effective_storage_base_path or snapshot.storage_base_path + ) if not base_path: _logger.warning( "%s is required but not set, multimodal uploader disabled", diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py index d36b49df2..0f9e8fc24 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/multimodal_upload_hook.py @@ -121,7 +121,9 @@ def _shutdown() -> None: if pre_uploader is not None: pre_uploader.shutdown() except Exception: # pylint: disable=broad-except - _logger.debug("Failed to shutdown retired pre-uploader", exc_info=True) + _logger.debug( + "Failed to shutdown retired pre-uploader", exc_info=True + ) try: if uploader is not None: uploader.shutdown() diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py index 5f45007ed..910938344 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/pre_uploader.py @@ -66,8 +66,8 @@ # Try importing audio processing libraries (optional dependencies) try: - import numpy as np - import soundfile as sf + import numpy as np # pyright: ignore[reportMissingImports] + import soundfile as sf # pyright: ignore[reportMissingImports] _audio_libs_available = True except ImportError: @@ -591,7 +591,7 @@ def _convert_pcm16_to_wav( try: # Convert PCM16 byte data to numpy int16 array - audio_np = np.frombuffer(pcm_data, dtype=np.int16) # pyright: ignore[reportUnknownMemberType] + audio_np = np.frombuffer(pcm_data, dtype=np.int16) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] # Write WAV data to memory buffer buffer = io.BytesIO() @@ -1190,7 +1190,7 @@ def _collect_http_uris( @staticmethod def _read_applied_strategy_version() -> int: - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) @@ -1198,20 +1198,13 @@ def _read_applied_strategy_version() -> int: def _sync_strategy_from_snapshot(self, config_snapshot: Any) -> bool: """Apply strategy fields when snapshot strategy_version changes.""" - if ( - config_snapshot.strategy_version - != self._applied_strategy_version - ): + if config_snapshot.strategy_version != self._applied_strategy_version: self._process_input = config_snapshot.process_input self._process_output = config_snapshot.process_output self._download_enabled = config_snapshot.download_enabled self._local_file_enabled = config_snapshot.local_file_enabled - self._allowed_root_paths = list( - config_snapshot.allowed_root_paths - ) - self._applied_strategy_version = ( - config_snapshot.strategy_version - ) + self._allowed_root_paths = list(config_snapshot.allowed_root_paths) + self._applied_strategy_version = config_snapshot.strategy_version return self._process_input or self._process_output def pre_upload( # pylint: disable=too-many-branches @@ -1241,7 +1234,7 @@ def pre_upload( # pylint: disable=too-many-branches uploads: List[PreUploadItem] = [] if config_snapshot is None: - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) @@ -1266,7 +1259,6 @@ def _pre_upload_with_current_config( output_messages: Optional[List[Any]], uploads: List[PreUploadItem], ) -> List[PreUploadItem]: - trace_id: Optional[str] = None span_id: Optional[str] = None try: @@ -1330,13 +1322,15 @@ def fs_pre_uploader_hook( ) -> Optional[PreUploader]: """Create file-system pre-uploader from runtime snapshot.""" if snapshot is None: - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) snapshot = get_multimodal_config_snapshot() - base_path = snapshot.effective_storage_base_path or snapshot.storage_base_path + base_path = ( + snapshot.effective_storage_base_path or snapshot.storage_base_path + ) if not base_path: _logger.warning( "%s is required but not set, multimodal pre-uploader disabled", diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py index eb0363624..539bcde63 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/utils.py @@ -136,7 +136,7 @@ def _parse_env_bool(value: Optional[str], default: bool) -> bool: def _get_multimodal_snapshot(): - from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module + from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=import-outside-toplevel,no-name-in-module # noqa: PLC0415 get_multimodal_config_snapshot, ) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py index 332e501df..876d73b64 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_default_hooks.py @@ -24,11 +24,12 @@ MultimodalPreUploader, fs_pre_uploader_hook, ) -from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test from opentelemetry.util.genai.extended_environment_variables import ( OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_STORAGE_BASE_PATH, ) +from .multimodal_test_helpers import reset_multimodal_runtime_state_for_test + class TestDefaultHooks(TestCase): @patch.dict("os.environ", {}, clear=True) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py index 64705c9b2..1bb76b1da 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py @@ -68,7 +68,8 @@ def test_env_fallback_defaults(self) -> None: snapshot.uploader_hook_name, get_default_uploader_hook_name() ) self.assertEqual( - snapshot.pre_uploader_hook_name, get_default_pre_uploader_hook_name() + snapshot.pre_uploader_hook_name, + get_default_pre_uploader_hook_name(), ) self.assertIsNone(snapshot.effective_storage_base_path) self.assertFalse(snapshot.process_input) @@ -131,9 +132,13 @@ def test_uploader_generation_fields_bump_generation(self) -> None: uploader_hook_name="fs", pre_uploader_hook_name="fs", ) - self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertGreater( + after.uploader_generation, before.uploader_generation + ) self.assertEqual(after.storage_base_path, "oss://bucket/prefix") - self.assertEqual(after.effective_storage_base_path, "oss://bucket/prefix") + self.assertEqual( + after.effective_storage_base_path, "oss://bucket/prefix" + ) def test_sls_project_change_bumps_uploader_generation(self) -> None: before = get_multimodal_config_snapshot() @@ -143,7 +148,9 @@ def test_sls_project_change_bumps_uploader_generation(self) -> None: sls_project="project-a", sls_logstore="logstore-a", ) - self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertGreater( + after.uploader_generation, before.uploader_generation + ) self.assertEqual(after.uploader_hook_name, "arms") self.assertEqual( after.effective_storage_base_path, @@ -159,7 +166,9 @@ def test_logstore_change_bumps_uploader_generation(self) -> None: ) before = get_multimodal_config_snapshot() after = update_multimodal_runtime_config(sls_logstore="l2") - self.assertGreater(after.uploader_generation, before.uploader_generation) + self.assertGreater( + after.uploader_generation, before.uploader_generation + ) self.assertIn("sls_logstore", UPLOADER_GENERATION_FIELDS) self.assertEqual(after.effective_storage_base_path, "sls://p/l2") @@ -185,7 +194,9 @@ def test_empty_upload_mode_fallback_to_none(self) -> None: def test_no_op_update_keeps_snapshot(self) -> None: before = get_multimodal_config_snapshot() - after = update_multimodal_runtime_config(upload_mode=before.upload_mode) + after = update_multimodal_runtime_config( + upload_mode=before.upload_mode + ) self.assertEqual(after, before) def test_blank_sls_project_is_coalesced_to_none(self) -> None: diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py index f8f51d219..3c9e6b438 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_upload_hook.py @@ -170,7 +170,9 @@ def load_pre_uploader_factory(): def fake_entry_points(group: str) -> list[FakeEntryPoint]: if group == "opentelemetry_genai_multimodal_uploader": - return [FakeEntryPoint(default_uploader, load_uploader_factory)] + return [ + FakeEntryPoint(default_uploader, load_uploader_factory) + ] if group == "opentelemetry_genai_multimodal_pre_uploader": return [ FakeEntryPoint( @@ -370,14 +372,17 @@ def load_pair_and_bump_generation( ) return FakeUploader(), FakePreUploader() - with patch.object( - module, - "_load_pair_from_snapshot", - side_effect=load_pair_and_bump_generation, - ), patch.object( - module, - "_schedule_retired_pair_shutdown", - side_effect=track_shutdown, + with ( + patch.object( + module, + "_load_pair_from_snapshot", + side_effect=load_pair_and_bump_generation, + ), + patch.object( + module, + "_schedule_retired_pair_shutdown", + side_effect=track_shutdown, + ), ): uploader, pre_uploader = module.get_or_rebuild_uploader_pair() diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py index afac42910..2e65e4a19 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader.py @@ -1153,7 +1153,8 @@ def test_local_file_processing_relative_path(pre_uploader_factory): relative_path = test_file.name # Aliyun Python Agent Extension: for compatibility with Python 3.8 - with patch.dict( + with ( + patch.dict( os.environ, { "OTEL_INSTRUMENTATION_GENAI_MULTIMODAL_LOCAL_FILE_ENABLED": "true", @@ -1161,8 +1162,9 @@ def test_local_file_processing_relative_path(pre_uploader_factory): test_dir ), }, - ), \ - patch("os.getcwd", return_value=str(test_dir)): + ), + patch("os.getcwd", return_value=str(test_dir)), + ): reset_multimodal_runtime_state_for_test() pre_uploader = pre_uploader_factory() # Test with simple filename (relative path) diff --git a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py index df9983f04..39bb6dcc5 100644 --- a/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_pre_uploader_audio.py @@ -235,15 +235,19 @@ def test_pcm16_conversion_missing_audio_libs_logs_single_warning( caplog, ): """Missing optional audio libs should only log the actual conversion skip.""" - with patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader._audio_libs_available", - False, - ), patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader.np", - None, - ), patch( - "opentelemetry.util.genai._multimodal_upload.pre_uploader.sf", - None, + with ( + patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader._audio_libs_available", + False, + ), + patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader.np", + None, + ), + patch( + "opentelemetry.util.genai._multimodal_upload.pre_uploader.sf", + None, + ), ): pre_uploader = MultimodalPreUploader(base_path="/tmp/test_upload") part = Blob( diff --git a/util/opentelemetry-util-genai/tests/test_extended_handler.py b/util/opentelemetry-util-genai/tests/test_extended_handler.py index b0be4562b..64ccf0976 100644 --- a/util/opentelemetry-util-genai/tests/test_extended_handler.py +++ b/util/opentelemetry-util-genai/tests/test_extended_handler.py @@ -57,10 +57,6 @@ from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module update_multimodal_runtime_config, ) - -from ._multimodal_upload.multimodal_test_helpers import ( - reset_multimodal_runtime_state_for_test, -) from opentelemetry.util.genai.environment_variables import ( OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, @@ -111,6 +107,10 @@ Uri, ) +from ._multimodal_upload.multimodal_test_helpers import ( + reset_multimodal_runtime_state_for_test, +) + _AGENT_NAME_BAGGAGE_KEY = "gen_ai.agent.name" @@ -1863,12 +1863,14 @@ def _record_llm_metrics(self, *_args: Any, **_kwargs: Any) -> None: async_inv.span = mock_span async_inv.input_messages = inv.input_messages - with patch.object( - h1, "_upload_and_set_metadata" - ) as mock_upload, patch( - "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" - ), patch( - "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + with ( + patch.object(h1, "_upload_and_set_metadata") as mock_upload, + patch( + "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" + ), + patch( + "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + ), ): h1._async_stop_llm( _MultimodalAsyncTask( @@ -1879,12 +1881,14 @@ def _record_llm_metrics(self, *_args: Any, **_kwargs: Any) -> None: mock_span.end.assert_called_once() mock_span.reset_mock() - with patch.object( - h2, "_upload_and_set_metadata" - ) as mock_upload2, patch( - "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" - ), patch( - "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + with ( + patch.object(h2, "_upload_and_set_metadata") as mock_upload2, + patch( + "opentelemetry.util.genai._multimodal_processing._apply_llm_finish_attributes" + ), + patch( + "opentelemetry.util.genai._multimodal_processing._maybe_emit_llm_event" + ), ): h2._async_stop_llm( _MultimodalAsyncTask(