feat(benchmark): 导出器接入 Cosmos NoSQL + Blob 存储账户(可组合多 target)
阶段3 数据落库：BENCHMARK_EXPORT_TARGET 支持逗号组合 cosmos,blob（默认 none）。 - CosmosExporter：Cosmos DB for NoSQL，一 run 一文档(id=swarm_id，分区键 /swarm_id)。 - BlobExporter：存储账户归档 <scenario>/<swarm_id>.json。 - get_exporters() 解析多 target；某 target 缺凭据/未知则跳过+告警，不影响其它与 run。 - capture 对每个 exporter 独立 to_thread 导出，互不影响。 - orchestrator/requirements：加 azure-cosmos/azure-storage-blob/azure-identity(懒导入)。 安全：所有连接串/AccountKey 仅从环境读(经 Secret/secret_ref 注入)， 绝不写进代码/日志/提交(组织安全规则)。 验证：新增 test-benchmark-export.py(target 解析/隔离/doc 成形) + capture/selfcert/ collector 回归全 PASS。 影响范围：agent_swarm benchmark 导出层 + 依赖；运行时只读钩子不变；导出默认关、无密钥落地。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Fasthei
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
94ef894ace
commit
27739af446
@@ -16,7 +16,7 @@ import logging
|
||||
import math
|
||||
|
||||
from .run_collector import SwarmRunMetricsCollector
|
||||
from ..export import get_exporter
|
||||
from ..export import get_exporters
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -44,10 +44,13 @@ async def capture_run_metrics(run) -> dict:
|
||||
}
|
||||
await swarm_runtime.record_benchmark_metrics(run, payload)
|
||||
|
||||
# Export is best-effort and OFF by default (NoopExporter). Run in a thread so a blocking
|
||||
# blob upload never stalls the event loop; never fails the run.
|
||||
try:
|
||||
await asyncio.to_thread(get_exporter().export, run.swarm_id, payload)
|
||||
except Exception as exc: # pragma: no cover - export is opt-in
|
||||
logger.warning("benchmark export failed for run %s: %s", run.swarm_id, exc)
|
||||
# Export to each configured target (cosmos / blob; none by default). Best-effort + isolated:
|
||||
# one target failing never breaks another or the run. Run in a thread so blocking network I/O
|
||||
# never stalls the event loop.
|
||||
for exporter in get_exporters():
|
||||
try:
|
||||
await asyncio.to_thread(exporter.export, run.swarm_id, payload)
|
||||
except Exception as exc: # pragma: no cover - export is opt-in / network-dependent
|
||||
logger.warning("benchmark export (%s) failed for run %s: %s",
|
||||
getattr(exporter, "name", "?"), run.swarm_id, exc)
|
||||
return payload
|
||||
|
||||
@@ -1,30 +1,34 @@
|
||||
"""Benchmark metrics 导出器（阶段2 数据采集）。
|
||||
"""Benchmark metrics 导出器（阶段2/3 数据采集落库）。
|
||||
|
||||
把每次真实 run 终态采集到的 `SwarmMetrics` 快照导出到外部留存，供后续【经验标定】
|
||||
（O / BASE_COEFFICIENTS / S_gain）累积数据。**默认 NoopExporter**：不导出、无外部依赖、
|
||||
不需任何凭据——指标仍持久化在 `run.metadata['benchmark']`（Redis）。
|
||||
（O / BASE_COEFFICIENTS / S_gain）累积真实用户数据。
|
||||
|
||||
启用 Azure Blob 导出（凭据经环境/连接串注入，**绝不写进代码或日志**）：
|
||||
BENCHMARK_EXPORT_TARGET=blob
|
||||
BENCHMARK_BLOB_CONTAINER=benchmark-selfcert # 默认 benchmark-selfcert
|
||||
AZURE_STORAGE_CONNECTION_STRING=... # 方式一：连接串
|
||||
# 或方式二（无密钥，推荐）：账户 URL + Workload Identity / Managed Identity
|
||||
BENCHMARK_BLOB_ACCOUNT_URL=https://heicode.blob.core.windows.net
|
||||
目标可组合，由 `BENCHMARK_EXPORT_TARGET`（逗号分隔）选择：
|
||||
- `none`（默认） → NoopExporter：不导出、无依赖、无凭据（指标仍落 run.metadata Redis）。
|
||||
- `cosmos` → CosmosExporter：写 Cosmos DB for NoSQL（结构化指标，便于查询/排行榜）。
|
||||
- `blob` → BlobExporter：写 Azure Blob 存储账户（全量快照归档）。
|
||||
例：`BENCHMARK_EXPORT_TARGET=cosmos,blob`。
|
||||
|
||||
归档路径：`<scenario>/<swarm_id>.json`。
|
||||
**凭据一律经环境/Secret 注入，绝不写进代码、日志或提交**（组织安全规则）：
|
||||
Cosmos: BENCHMARK_COSMOS_CONNECTION_STRING（必填）
|
||||
BENCHMARK_COSMOS_DATABASE（默认 benchmark）/ BENCHMARK_COSMOS_CONTAINER（默认 selfcert）
|
||||
Blob: AZURE_STORAGE_CONNECTION_STRING（或 BENCHMARK_BLOB_ACCOUNT_URL + Managed/Workload Identity）
|
||||
BENCHMARK_BLOB_CONTAINER（默认 benchmark-selfcert）
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
from typing import List, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MetricsExporter:
|
||||
"""导出一次 run 的 benchmark 指标快照。子类实现 export()。"""
|
||||
"""导出一次 run 的 benchmark 指标快照。子类实现 export()。构造不连网（懒连接）。"""
|
||||
|
||||
name = "abstract"
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
raise NotImplementedError
|
||||
@@ -33,13 +37,44 @@ class MetricsExporter:
|
||||
class NoopExporter(MetricsExporter):
|
||||
"""默认：不导出（指标已落 run.metadata）。无外部依赖、无凭据。"""
|
||||
|
||||
name = "noop"
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
logger.debug("benchmark export disabled (BENCHMARK_EXPORT_TARGET=none); "
|
||||
"run %s metrics persisted to run.metadata only", swarm_id)
|
||||
logger.debug("benchmark export disabled; run %s metrics persisted to run.metadata only", swarm_id)
|
||||
|
||||
|
||||
class CosmosExporter(MetricsExporter):
|
||||
"""Azure Cosmos DB for NoSQL（SQL API）。一 run 一文档，id=swarm_id，分区键 /swarm_id。"""
|
||||
|
||||
name = "cosmos"
|
||||
|
||||
def __init__(self, connection_string: str, database: str, container: str):
|
||||
if not connection_string:
|
||||
raise RuntimeError("CosmosExporter 需 BENCHMARK_COSMOS_CONNECTION_STRING")
|
||||
self.connection_string = connection_string
|
||||
self.database = database
|
||||
self.container = container
|
||||
|
||||
@staticmethod
|
||||
def build_doc(payload: dict) -> dict:
|
||||
"""Cosmos 文档：需要字符串 id；分区键字段 swarm_id 已在 payload 内。"""
|
||||
return {"id": str(payload.get("swarm_id")), **payload}
|
||||
|
||||
def _container_client(self):
|
||||
from azure.cosmos import CosmosClient # lazy: keep core import-light & key-free
|
||||
client = CosmosClient.from_connection_string(self.connection_string)
|
||||
db = client.get_database_client(self.database)
|
||||
return db.get_container_client(self.container)
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
self._container_client().upsert_item(self.build_doc(payload))
|
||||
logger.info("benchmark metrics → cosmos %s/%s (id=%s)", self.database, self.container, swarm_id)
|
||||
|
||||
|
||||
class BlobExporter(MetricsExporter):
|
||||
"""Azure Blob 导出（懒加载 SDK；凭据来自连接串或 Managed/Workload Identity）。"""
|
||||
"""Azure Blob 存储账户（懒加载 SDK）。归档路径 <scenario>/<swarm_id>.json。"""
|
||||
|
||||
name = "blob"
|
||||
|
||||
def __init__(self, container: str, *, connection_string: Optional[str] = None,
|
||||
account_url: Optional[str] = None):
|
||||
@@ -50,30 +85,48 @@ class BlobExporter(MetricsExporter):
|
||||
self.account_url = account_url
|
||||
|
||||
def _client(self):
|
||||
from azure.storage.blob import BlobServiceClient # lazy: keep core import-light & key-free
|
||||
from azure.storage.blob import BlobServiceClient # lazy
|
||||
if self.connection_string:
|
||||
return BlobServiceClient.from_connection_string(self.connection_string)
|
||||
from azure.identity import DefaultAzureCredential # Workload/Managed Identity, no secret
|
||||
return BlobServiceClient(account_url=self.account_url, credential=DefaultAzureCredential())
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
scenario = (payload.get("scenario") or "unknown")
|
||||
scenario = payload.get("scenario") or "unknown"
|
||||
blob_name = f"{scenario}/{swarm_id}.json"
|
||||
client = self._client().get_blob_client(container=self.container, blob=blob_name)
|
||||
client.upload_blob(json.dumps(payload, ensure_ascii=False), overwrite=True)
|
||||
logger.info("benchmark metrics exported → %s/%s", self.container, blob_name)
|
||||
logger.info("benchmark metrics → blob %s/%s", self.container, blob_name)
|
||||
|
||||
|
||||
def get_exporter() -> MetricsExporter:
|
||||
"""按 BENCHMARK_EXPORT_TARGET 选择导出器（默认 none → NoopExporter）。"""
|
||||
target = (os.getenv("BENCHMARK_EXPORT_TARGET") or "none").strip().lower()
|
||||
def _build_target(target: str) -> Optional[MetricsExporter]:
|
||||
"""Build one exporter from a target name; None (with a warning) if misconfigured/unknown."""
|
||||
if target in ("none", "off", ""):
|
||||
return NoopExporter()
|
||||
if target == "blob":
|
||||
return BlobExporter(
|
||||
container=os.getenv("BENCHMARK_BLOB_CONTAINER", "benchmark-selfcert"),
|
||||
connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING"),
|
||||
account_url=os.getenv("BENCHMARK_BLOB_ACCOUNT_URL"),
|
||||
)
|
||||
logger.warning("unknown BENCHMARK_EXPORT_TARGET=%s; falling back to noop", target)
|
||||
return NoopExporter()
|
||||
try:
|
||||
if target == "cosmos":
|
||||
return CosmosExporter(
|
||||
connection_string=os.getenv("BENCHMARK_COSMOS_CONNECTION_STRING", ""),
|
||||
database=os.getenv("BENCHMARK_COSMOS_DATABASE", "benchmark"),
|
||||
container=os.getenv("BENCHMARK_COSMOS_CONTAINER", "selfcert"),
|
||||
)
|
||||
if target == "blob":
|
||||
return BlobExporter(
|
||||
container=os.getenv("BENCHMARK_BLOB_CONTAINER", "benchmark-selfcert"),
|
||||
connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING"),
|
||||
account_url=os.getenv("BENCHMARK_BLOB_ACCOUNT_URL"),
|
||||
)
|
||||
except Exception as exc:
|
||||
# Misconfigured target (e.g. missing creds) must not break the others or the run.
|
||||
logger.warning("benchmark export target '%s' disabled: %s", target, exc)
|
||||
return None
|
||||
logger.warning("unknown BENCHMARK_EXPORT_TARGET entry '%s'; ignored", target)
|
||||
return None
|
||||
|
||||
|
||||
def get_exporters() -> List[MetricsExporter]:
|
||||
"""Parse BENCHMARK_EXPORT_TARGET (comma list) → exporters. Default → [NoopExporter]."""
|
||||
raw = (os.getenv("BENCHMARK_EXPORT_TARGET") or "none").strip().lower()
|
||||
targets = [t.strip() for t in raw.split(",") if t.strip()] or ["none"]
|
||||
exporters = [e for e in (_build_target(t) for t in targets) if e is not None]
|
||||
return exporters or [NoopExporter()]
|
||||
|
||||
Reference in New Issue
Block a user