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/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/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. 补充说明 ------------------------------------------------------------------------ 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..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 @@ -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 # noqa: PLC0415 + 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 # noqa: PLC0415 + 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 # noqa: PLC0415 + 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 # noqa: PLC0415 + 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..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,10 +22,15 @@ 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, get_or_load_uploader_pair, + get_or_rebuild_uploader_pair, get_pre_uploader, get_uploader, get_uploader_pair, @@ -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..3d6bb5a0a --- /dev/null +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_multimodal_upload/config.py @@ -0,0 +1,389 @@ +# 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", + } +) + +# 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: + 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: + 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) + 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..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 @@ -60,10 +60,19 @@ 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 # noqa: PLC0415 + 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( 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..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 @@ -14,15 +14,15 @@ from __future__ import annotations +import inspect 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 +40,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 +56,113 @@ 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: + params = inspect.signature(hook).parameters + except (TypeError, ValueError): + params = {} + if params: + return hook(snapshot) + 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 +173,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 +207,100 @@ 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. - - First call performs one-time loading; subsequent calls return cache. - If either side fails to load, both are downgraded to `(None, None)`. +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 + + +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 + + with _uploader_pair_lock: + if _failed_generation == generation: + return True, (None, None) + if ( + _uploader_generation == generation + and _uploader is not None + and _pre_uploader is not None + ): + return True, (_uploader, _pre_uploader) + if _building_generation == generation: + return True, (None, None) + _building_generation = generation + return False, (None, None) + + +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 + + 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 + _logger.warning( + "Multimodal uploader pair build failed for generation %s; " + "uploads disabled until generation changes", + 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_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 - 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 + generation = cfg.uploader_generation + resolved, pair = _resolve_cached_uploader_pair(generation) + if resolved: + return pair - _load_once.do_once(_load) - return _uploader, _pre_uploader + 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]]: + """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 +308,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..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 @@ -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, @@ -144,7 +143,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 +155,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 +1188,32 @@ 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 # noqa: PLC0415 + 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 +1233,32 @@ 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 # noqa: PLC0415 + 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 +1317,20 @@ 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 # noqa: PLC0415 + 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..539bcde63 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 # noqa: PLC0415 + 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..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 @@ -28,6 +28,8 @@ 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) @@ -50,8 +52,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 +78,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..1bb76b1da --- /dev/null +++ b/util/opentelemetry-util-genai/tests/_multimodal_upload/test_multimodal_runtime_config.py @@ -0,0 +1,300 @@ +# 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._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, + 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() + + 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..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 @@ -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,14 +27,21 @@ 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 ( + get_default_pre_uploader_hook_name, + get_default_uploader_hook_name, + reload_multimodal_upload_hook_module, ) @@ -50,6 +60,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 +76,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 +93,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 +110,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 +121,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 +151,10 @@ 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): + 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 +168,17 @@ 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 +198,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 +212,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 +236,213 @@ 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 + + def pre_uploader_factory() -> FakePreUploader: + return FakePreUploader() + + with patch.object( + module, + "_iter_entry_points", + side_effect=_fake_fs_entry_points( + uploader_factory, pre_uploader_factory + ), + ): + 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..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 @@ -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,6 +1152,7 @@ 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 + # Aliyun Python Agent Extension: for compatibility with Python 3.8 with ( patch.dict( os.environ, @@ -1155,6 +1165,7 @@ def test_local_file_processing_relative_path(pre_uploader_factory): ), 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 +1218,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 +1266,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 +1293,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..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 @@ -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 @@ -285,6 +288,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( diff --git a/util/opentelemetry-util-genai/tests/test_extended_handler.py b/util/opentelemetry-util-genai/tests/test_extended_handler.py index e279d9f43..64ccf0976 100644 --- a/util/opentelemetry-util-genai/tests/test_extended_handler.py +++ b/util/opentelemetry-util-genai/tests/test_extended_handler.py @@ -54,6 +54,9 @@ MultimodalProcessingMixin, _MultimodalAsyncTask, ) +from opentelemetry.util.genai._multimodal_upload.config import ( # pylint: disable=no-name-in-module + update_multimodal_runtime_config, +) from opentelemetry.util.genai.environment_variables import ( OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, @@ -104,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" @@ -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,97 @@ 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 +1949,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 +1994,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 +2431,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 +2440,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