From 78e49d6a134b0e42070c924be814fb73a31ee14b Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Fri, 28 Aug 2026 23:17:59 +0800 Subject: [PATCH] fix(studio): fall back to TOS intranet endpoint --- frontend/README.md | 13 +- .../server/evaluation_automation/__init__.py | 16 +- frontend/server/knowledge/uploads.py | 26 ++- frontend/server/skills/storage.py | 20 +- frontend/server/storage/tos.py | 124 +++++++++- frontend/server/video/storage.py | 11 +- frontend/service/studio_scheduler/app.py | 24 +- tests/frontend/server/storage/test_tos.py | 211 ++++++++++++++++++ tests/multimodal/test_storage.py | 58 ++++- veadk/multimodal/storage.py | 15 +- 10 files changed, 448 insertions(+), 70 deletions(-) create mode 100644 tests/frontend/server/storage/test_tos.py diff --git a/frontend/README.md b/frontend/README.md index 4d2180a25..b7d8236c4 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -636,10 +636,15 @@ export VEADK_STUDIO_TOS_BUCKET=teststudio ``` The server derives the provider-specific endpoint, such as -`tos-cn-beijing.volces.com`, and never sends TOS credentials to the -browser. Local Studio uses the configured Volcengine or BytePlus AK/SK; VeFaaS -uses its IAM role credentials. Studio objects use the versioned, user-first -layout +`tos-cn-beijing.volces.com`, and never sends TOS credentials to the browser. +For Volcengine, Studio probes the public endpoint once and automatically uses +the matching `tos-.ivolces.com` intranet endpoint when the public +endpoint has a transport-level connection failure. Authentication, permission, +and other TOS service errors do not trigger fallback. Browser-facing signed URLs +continue to use the public endpoint. BytePlus and custom endpoints are left +unchanged. Local Studio uses the configured Volcengine or BytePlus AK/SK; +VeFaaS uses its IAM role credentials. Studio objects use the versioned, +user-first layout `veadk-studio/v1/users/////`. Video reference assets currently use the `video///` namespace and store `content` plus `metadata.json` below it. diff --git a/frontend/server/evaluation_automation/__init__.py b/frontend/server/evaluation_automation/__init__.py index 0610bdd37..dbff9d93c 100644 --- a/frontend/server/evaluation_automation/__init__.py +++ b/frontend/server/evaluation_automation/__init__.py @@ -22,6 +22,7 @@ import httpx from frontend.server.storage import StudioProvider, StudioStorageConfig +from frontend.server.storage.tos import create_tos_client_factory from veadk.utils.logger import get_logger from .datasets import ensure_feedback_sets @@ -51,22 +52,9 @@ def create_service( models = StructuredEvaluationModels() storage = StudioStorageConfig.from_env(provider) if storage.configured and resolve_credentials is not None: - - def tos_client() -> Any: - import tos - - access_key, secret_key, session_token = resolve_credentials() - return tos.TosClientV2( - ak=access_key, - sk=secret_key, - security_token=session_token, - endpoint=storage.endpoint, - region=storage.region, - ) - optimizations = TosOptimizationRepository( bucket=storage.bucket, - client_factory=tos_client, + client_factory=create_tos_client_factory(storage, resolve_credentials), ) else: logger.warning( diff --git a/frontend/server/knowledge/uploads.py b/frontend/server/knowledge/uploads.py index 6955a126e..baddb36cf 100644 --- a/frontend/server/knowledge/uploads.py +++ b/frontend/server/knowledge/uploads.py @@ -92,7 +92,7 @@ def _delete_objects(client: Any, *, bucket: str, keys: tuple[str, ...]) -> None: for key in keys: try: client.delete_object(bucket=bucket, key=key) - except Exception as error: + except Exception as error: # noqa: BLE001 - collect every failed delete errors.append(error) if not errors: return @@ -297,6 +297,7 @@ def __init__( if self.max_file_bytes <= 0: raise ValueError("VEADK_KNOWLEDGE_MAX_FILE_BYTES must be positive") self._prepared_targets: set[tuple[str, str]] = set() + self._client_factories: dict[tuple[str, str, str], Callable[[], Any]] = {} self._lock = RLock() configured_account_id = str( environment.get("VEADK_STUDIO_ACCOUNT_ID") or "" @@ -680,13 +681,22 @@ def _resolve_account_id(self, normalized_region: str) -> str: return account_id def _client(self, target: _UploadTarget) -> Any: - config = StudioStorageConfig( - provider=self._provider, - bucket=target.bucket, - region=target.region, - endpoint=target.endpoint, - ) - return create_tos_client_factory(config, self._resolve_credentials)() + cache_key = (target.bucket, target.region, target.endpoint) + with self._lock: + factory = self._client_factories.get(cache_key) + if factory is None: + config = StudioStorageConfig( + provider=self._provider, + bucket=target.bucket, + region=target.region, + endpoint=target.endpoint, + ) + factory = create_tos_client_factory( + config, + self._resolve_credentials, + ) + self._client_factories[cache_key] = factory + return factory() def _prepare_target(self, client: Any, target: _UploadTarget) -> None: cache_key = (target.bucket, target.region) diff --git a/frontend/server/skills/storage.py b/frontend/server/skills/storage.py index 187619dbf..9af2737fa 100644 --- a/frontend/server/skills/storage.py +++ b/frontend/server/skills/storage.py @@ -25,6 +25,7 @@ from typing import Any, Literal from frontend.server.storage import StudioProvider, StudioStorageConfig +from frontend.server.storage.tos import create_tos_client_factory from veadk.utils.cloud_provider import cloud_provider_from_env _IAM_CREDENTIAL_PATH = Path("/var/run/secrets/iam/credential") @@ -206,15 +207,20 @@ def _create_tos_client( storage: SkillPublishStorage, credentials: SkillPublishCredentials, ) -> Any: - import tos - - return tos.TosClientV2( - ak=credentials.access_key, - sk=credentials.secret_key, - security_token=credentials.session_token, - endpoint=storage.endpoint, + config = StudioStorageConfig( + provider=storage.provider, + bucket=storage.bucket, region=storage.region, + endpoint=storage.endpoint, ) + return create_tos_client_factory( + config, + lambda: ( + credentials.access_key, + credentials.secret_key, + credentials.session_token, + ), + )() def _listed_buckets(client: Any) -> dict[str, str]: diff --git a/frontend/server/storage/tos.py b/frontend/server/storage/tos.py index e2ca17dfe..f63b9758b 100644 --- a/frontend/server/storage/tos.py +++ b/frontend/server/storage/tos.py @@ -17,32 +17,144 @@ from __future__ import annotations from collections.abc import Callable +from threading import Lock from typing import Any +from veadk.utils.logger import get_logger + from . import StudioStorageConfig CredentialResolver = Callable[[], tuple[str, str, str | None]] TosClientFactory = Callable[[], Any] +_PUBLIC_ENDPOINT_PROBE_TIMEOUT_SECONDS = 3 +logger = get_logger(__name__) + + +def _endpoint_candidates(config: StudioStorageConfig) -> tuple[str, ...]: + """Return safe endpoint candidates without overriding custom endpoints.""" + public_endpoint = f"tos-{config.region}.volces.com" + if config.provider == "volcengine" and config.endpoint == public_endpoint: + return public_endpoint, f"tos-{config.region}.ivolces.com" + return (config.endpoint,) + + +def _new_client( + tos_module: Any, + *, + endpoint: str, + region: str, + access_key: str, + secret_key: str, + session_token: str | None, + probe: bool = False, +) -> Any: + options: dict[str, Any] = { + "ak": access_key, + "sk": secret_key, + "security_token": session_token, + "endpoint": endpoint, + "region": region, + } + client = tos_module.TosClientV2(**options) + if probe: + # Endpoint selection should not inherit the SDK's three retries and turn + # a predictable private-network fallback into a long cold-start delay. + client.max_retry_count = 0 + client.connection_time = _PUBLIC_ENDPOINT_PROBE_TIMEOUT_SECONDS + return client + + +def _is_network_error(error: Exception, tos_module: Any) -> bool: + """Return whether the SDK error represents a transport failure.""" + client_error = getattr( + getattr(tos_module, "exceptions", None), + "TosClientError", + None, + ) + if client_error is None or not isinstance(error, client_error): + return False + try: + import requests + except ImportError: + return False + return isinstance( + getattr(error, "cause", None), + requests.exceptions.RequestException, + ) + def create_tos_client_factory( config: StudioStorageConfig, resolve_credentials: CredentialResolver, ) -> TosClientFactory: - """Create clients lazily so refreshed temporary credentials are respected.""" + """Create clients lazily and select a reachable Volcengine endpoint once.""" if not config.configured: raise ValueError(config.unavailable_reason) + candidates = _endpoint_candidates(config) + selected_endpoint = candidates[0] if len(candidates) == 1 else "" + selection_lock = Lock() + def factory() -> Any: + nonlocal selected_endpoint import tos access_key, secret_key, session_token = resolve_credentials() - return tos.TosClientV2( - ak=access_key, - sk=secret_key, - security_token=session_token, - endpoint=config.endpoint, + if not selected_endpoint: + with selection_lock: + if not selected_endpoint: + public_endpoint = candidates[0] + intranet_endpoint = f"tos-{config.region}.ivolces.com" + probe = _new_client( + tos, + endpoint=public_endpoint, + region=config.region, + access_key=access_key, + secret_key=secret_key, + session_token=session_token, + probe=True, + ) + head_bucket = getattr(probe, "head_bucket", None) + if not callable(head_bucket): + selected_endpoint = public_endpoint + else: + try: + head_bucket(bucket=config.bucket) + except Exception as error: + tos_exceptions = getattr(tos, "exceptions", None) + expected_errors = tuple( + error_type + for error_type in ( + getattr(tos_exceptions, "TosClientError", None), + getattr(tos_exceptions, "TosServerError", None), + ) + if isinstance(error_type, type) + ) + if not expected_errors or not isinstance( + error, expected_errors + ): + raise + if not _is_network_error(error, tos): + selected_endpoint = public_endpoint + else: + selected_endpoint = intranet_endpoint + logger.warning( + "Studio TOS public endpoint %s is unreachable; " + "using intranet endpoint %s.", + public_endpoint, + intranet_endpoint, + ) + else: + selected_endpoint = public_endpoint + + return _new_client( + tos, + endpoint=selected_endpoint, region=config.region, + access_key=access_key, + secret_key=secret_key, + session_token=session_token, ) return factory diff --git a/frontend/server/video/storage.py b/frontend/server/video/storage.py index 3a3dc89e8..9b95dc234 100644 --- a/frontend/server/video/storage.py +++ b/frontend/server/video/storage.py @@ -31,6 +31,7 @@ StudioStorageConfig, StudioTosMediaStorage, ) +from frontend.server.storage.tos import create_tos_client_factory from veadk.multimodal.models import MediaRecord, MediaRef from veadk.multimodal.service import MediaService @@ -246,16 +247,18 @@ def video_asset_repository_factory( if not config.configured: return None, max_bytes + client_factory = create_tos_client_factory(config, resolve_credentials) + def factory() -> VideoAssetRepository: - access_key, secret_key, session_token = resolve_credentials() storage = StudioTosMediaStorage( bucket=config.bucket, region=config.region, endpoint=config.endpoint, - access_key=access_key, - secret_key=secret_key, - session_token=session_token or "", + access_key="", + secret_key="", key_prefix=STUDIO_STORAGE_ROOT_PREFIX, + client=client_factory(), + signed_url_endpoint=config.endpoint, ) return VideoAssetRepository(MediaService(storage, max_file_bytes=max_bytes)) diff --git a/frontend/service/studio_scheduler/app.py b/frontend/service/studio_scheduler/app.py index c6daa7f3d..81b9b8d57 100644 --- a/frontend/service/studio_scheduler/app.py +++ b/frontend/service/studio_scheduler/app.py @@ -21,6 +21,9 @@ from dataclasses import dataclass from typing import Any, cast +from frontend.server.storage import StudioStorageConfig +from frontend.server.storage.tos import create_tos_client_factory + from .dispatcher import Dispatcher from .entrypoint import make_handler from .executor import ProviderRuntimeExecutor @@ -130,19 +133,22 @@ def create_dispatcher( def _tos_client_factory(settings: SchedulerSettings) -> Callable[[], Any]: - def create() -> Any: - import tos + config = StudioStorageConfig( + provider=settings.provider, + bucket=settings.bucket, + region=settings.storage_region, + endpoint=settings.storage_endpoint, + ) + def credentials() -> tuple[str, str, str | None]: credentials = resolve_service_credentials(settings.provider) - return tos.TosClientV2( - ak=credentials.access_key, - sk=credentials.secret_key, - security_token=credentials.session_token or None, - endpoint=settings.storage_endpoint, - region=settings.storage_region, + return ( + credentials.access_key, + credentials.secret_key, + credentials.session_token or None, ) - return create + return create_tos_client_factory(config, credentials) def handler(event: Any, context: Any) -> dict[str, int]: diff --git a/tests/frontend/server/storage/test_tos.py b/tests/frontend/server/storage/test_tos.py new file mode 100644 index 000000000..020e8be39 --- /dev/null +++ b/tests/frontend/server/storage/test_tos.py @@ -0,0 +1,211 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# 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 sys +from types import SimpleNamespace + +import requests + +from frontend.server.storage import StudioStorageConfig +from frontend.server.storage.tos import create_tos_client_factory + + +class _TosClientError(Exception): + def __init__(self, cause: Exception | None = None) -> None: + self.cause = cause + + +class _TosServerError(Exception): + pass + + +def _config(provider: str = "volcengine") -> StudioStorageConfig: + region = "cn-beijing" if provider == "volcengine" else "ap-southeast-1" + domain = "volces.com" if provider == "volcengine" else "bytepluses.com" + return StudioStorageConfig( + provider=provider, # type: ignore[arg-type] + bucket="studio", + region=region, + endpoint=f"tos-{region}.{domain}", + ) + + +def test_factory_falls_back_to_volcengine_intranet_on_public_network_error( + monkeypatch, +) -> None: + created: list[dict[str, object]] = [] + probes: list[tuple[str, int, int]] = [] + + class _Client: + def __init__(self, **kwargs: object) -> None: + self.endpoint = str(kwargs["endpoint"]) + self.max_retry_count = 3 + self.connection_time = 10 + created.append(kwargs) + + def head_bucket(self, *, bucket: str) -> None: + assert bucket == "studio" + probes.append((self.endpoint, self.max_retry_count, self.connection_time)) + if self.endpoint.endswith(".volces.com"): + raise _TosClientError(requests.ConnectionError("no public route")) + + monkeypatch.setitem( + sys.modules, + "tos", + SimpleNamespace( + TosClientV2=_Client, + exceptions=SimpleNamespace( + TosClientError=_TosClientError, + TosServerError=_TosServerError, + ), + ), + ) + factory = create_tos_client_factory(_config(), lambda: ("ak", "sk", "token")) + + first = factory() + second = factory() + + assert first.endpoint == "tos-cn-beijing.ivolces.com" + assert second.endpoint == "tos-cn-beijing.ivolces.com" + assert probes == [("tos-cn-beijing.volces.com", 0, 3)] + + +def test_factory_keeps_public_endpoint_when_tos_returns_a_server_error( + monkeypatch, +) -> None: + probes: list[str] = [] + + class _Client: + def __init__(self, **kwargs: object) -> None: + self.endpoint = str(kwargs["endpoint"]) + + def head_bucket(self, *, bucket: str) -> None: + assert bucket == "studio" + probes.append(self.endpoint) + raise _TosServerError("AccessDenied") + + monkeypatch.setitem( + sys.modules, + "tos", + SimpleNamespace( + TosClientV2=_Client, + exceptions=SimpleNamespace( + TosClientError=_TosClientError, + TosServerError=_TosServerError, + ), + ), + ) + + client = create_tos_client_factory(_config(), lambda: ("ak", "sk", None))() + + assert client.endpoint == "tos-cn-beijing.volces.com" + assert probes == ["tos-cn-beijing.volces.com"] + + +def test_factory_keeps_public_endpoint_for_non_network_client_errors( + monkeypatch, +) -> None: + class _Client: + def __init__(self, **kwargs: object) -> None: + self.endpoint = str(kwargs["endpoint"]) + + def head_bucket(self, *, bucket: str) -> None: + assert bucket == "studio" + raise _TosClientError(ValueError("invalid request")) + + monkeypatch.setitem( + sys.modules, + "tos", + SimpleNamespace( + TosClientV2=_Client, + exceptions=SimpleNamespace( + TosClientError=_TosClientError, + TosServerError=_TosServerError, + ), + ), + ) + + client = create_tos_client_factory(_config(), lambda: ("ak", "sk", None))() + + assert client.endpoint == "tos-cn-beijing.volces.com" + + +def test_factory_does_not_probe_or_infer_an_intranet_endpoint_for_byteplus( + monkeypatch, +) -> None: + created: list[dict[str, object]] = [] + + class _Client: + def __init__(self, **kwargs: object) -> None: + self.endpoint = str(kwargs["endpoint"]) + created.append(kwargs) + + def head_bucket(self, *, bucket: str) -> None: + raise AssertionError(f"unexpected probe for {bucket}") + + monkeypatch.setitem( + sys.modules, + "tos", + SimpleNamespace( + TosClientV2=_Client, + exceptions=SimpleNamespace( + TosClientError=_TosClientError, + TosServerError=_TosServerError, + ), + ), + ) + + client = create_tos_client_factory( + _config("byteplus"), lambda: ("ak", "sk", None) + )() + + assert client.endpoint == "tos-ap-southeast-1.bytepluses.com" + assert len(created) == 1 + + +def test_factory_preserves_a_custom_volcengine_endpoint(monkeypatch) -> None: + created: list[dict[str, object]] = [] + + class _Client: + def __init__(self, **kwargs: object) -> None: + self.endpoint = str(kwargs["endpoint"]) + created.append(kwargs) + + def head_bucket(self, *, bucket: str) -> None: + raise AssertionError(f"unexpected probe for {bucket}") + + monkeypatch.setitem( + sys.modules, + "tos", + SimpleNamespace( + TosClientV2=_Client, + exceptions=SimpleNamespace( + TosClientError=_TosClientError, + TosServerError=_TosServerError, + ), + ), + ) + config = StudioStorageConfig( + provider="volcengine", + bucket="studio", + region="cn-beijing", + endpoint="tos.example.com", + ) + + client = create_tos_client_factory(config, lambda: ("ak", "sk", None))() + + assert client.endpoint == "tos.example.com" + assert len(created) == 1 diff --git a/tests/multimodal/test_storage.py b/tests/multimodal/test_storage.py index a878872bc..2c6d90755 100644 --- a/tests/multimodal/test_storage.py +++ b/tests/multimodal/test_storage.py @@ -15,20 +15,21 @@ from __future__ import annotations +import sys from pathlib import Path from types import SimpleNamespace -import sys -from typing import Any +from typing import Any, ClassVar import pytest -from veadk.multimodal.models import MediaRecord -from veadk.multimodal.models import MediaRef -from veadk.multimodal.service import MediaService from veadk.multimodal import storage as storage_module -from veadk.multimodal.storage import LocalMediaStorage -from veadk.multimodal.storage import TosMediaStorage -from veadk.multimodal.storage import create_media_storage +from veadk.multimodal.models import MediaRecord, MediaRef +from veadk.multimodal.service import MediaService +from veadk.multimodal.storage import ( + LocalMediaStorage, + TosMediaStorage, + create_media_storage, +) def _record(ref: MediaRef, *, size_bytes: int = 5) -> MediaRecord: @@ -166,10 +167,10 @@ def __init__(self, status_code: int) -> None: class _FakeTosClient: - objects: dict[str, bytes] = {} + objects: ClassVar[dict[str, bytes]] = {} def __init__(self, **_: object) -> None: - self.objects = self.__class__.objects + pass def put_object(self, *, key: str, content: Any, **_: object) -> None: data = content.read() if hasattr(content, "read") else content @@ -180,8 +181,11 @@ def get_object(self, *, key: str, **_: object) -> SimpleNamespace: raise _FakeTosServerError(404) return SimpleNamespace(read=lambda: self.objects[key]) - def pre_signed_url(self, _: object, *, key: str, **__: object) -> SimpleNamespace: - return SimpleNamespace(signed_url=f"https://tos.example/{key}?signed=1") + def pre_signed_url( + self, _: object, *, key: str, **kwargs: object + ) -> SimpleNamespace: + endpoint = str(kwargs.get("alternative_endpoint") or "tos.example") + return SimpleNamespace(signed_url=f"https://{endpoint}/{key}?signed=1") def delete_object(self, *, key: str, **_: object) -> None: self.objects.pop(key, None) @@ -237,3 +241,33 @@ async def test_tos_storage_persists_signs_and_deletes( await storage.save_bytes(record, b"hello") await storage.delete_session("demo", "user", "session") assert await storage.get_record(ref) is None + + +@pytest.mark.asyncio +async def test_tos_storage_keeps_signed_urls_on_the_public_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_tos = SimpleNamespace( + TosClientV2=_FakeTosClient, + HttpMethodType=SimpleNamespace(Http_Method_Get="GET"), + exceptions=SimpleNamespace(TosServerError=_FakeTosServerError), + ) + monkeypatch.setitem(sys.modules, "tos", fake_tos) + storage = TosMediaStorage( + bucket="bucket", + region="cn-beijing", + endpoint="tos-cn-beijing.ivolces.com", + access_key="", + secret_key="", + client=_FakeTosClient(), + signed_url_endpoint="tos-cn-beijing.volces.com", + ) + ref = MediaRef("demo", "user", "session", "media-id") + + signed_url = await storage.signed_url(ref) + + assert signed_url == ( + "https://tos-cn-beijing.volces.com/" + "veadk-media/users/user/apps/demo/sessions/session/media/media-id/content" + "?signed=1" + ) diff --git a/veadk/multimodal/storage.py b/veadk/multimodal/storage.py index 2e5dd5cbf..84161fcce 100644 --- a/veadk/multimodal/storage.py +++ b/veadk/multimodal/storage.py @@ -19,13 +19,12 @@ import hashlib import json import os -from pathlib import Path import shutil -from typing import Protocol +from pathlib import Path +from typing import Any, Protocol from urllib.parse import quote -from .models import MediaRecord -from .models import MediaRef +from .models import MediaRecord, MediaRef _DEFAULT_LOCAL_MEDIA_DIR = Path("/tmp/veadk-media") @@ -217,8 +216,10 @@ def __init__( secret_key: str, session_token: str = "", key_prefix: str = "veadk-media", + client: Any | None = None, + signed_url_endpoint: str = "", ) -> None: - if not bucket or not access_key or not secret_key: + if not bucket or (client is None and (not access_key or not secret_key)): raise ValueError( "TOS media storage requires bucket, access key, and secret key." ) @@ -227,7 +228,8 @@ def __init__( self._tos = tos self._bucket = bucket self._key_prefix = key_prefix.strip("/") - self._client = tos.TosClientV2( + self._signed_url_endpoint = signed_url_endpoint.strip() or endpoint + self._client = client or tos.TosClientV2( ak=access_key, sk=secret_key, security_token=session_token, @@ -314,6 +316,7 @@ def _signed_url(self, ref: MediaRef) -> str: bucket=self._bucket, key=self._key(ref, "content"), expires=900, + alternative_endpoint=self._signed_url_endpoint, ) return output.signed_url