From 27739af4469dde8b553d076ad2be2cc54e7b01bc Mon Sep 17 00:00:00 2001 From: Fasthei <167957975+Fasthei@users.noreply.github.com> Date: Fri, 12 Jun 2026 18:12:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(benchmark):=20=E5=AF=BC=E5=87=BA=E5=99=A8?= =?UTF-8?q?=E6=8E=A5=E5=85=A5=20Cosmos=20NoSQL=20+=20Blob=20=E5=AD=98?= =?UTF-8?q?=E5=82=A8=E8=B4=A6=E6=88=B7(=E5=8F=AF=E7=BB=84=E5=90=88?= =?UTF-8?q?=E5=A4=9A=20target)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 阶段3 数据落库：BENCHMARK_EXPORT_TARGET 支持逗号组合 cosmos,blob（默认 none）。 - CosmosExporter：Cosmos DB for NoSQL，一 run 一文档(id=swarm_id，分区键 /swarm_id)。 - BlobExporter：存储账户归档 /.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) --- benchmark/collectors/capture.py | 17 +++-- benchmark/export/__init__.py | 111 +++++++++++++++++++++++-------- orchestrator/requirements.txt | 4 ++ scripts/test-benchmark-export.py | 85 +++++++++++++++++++++++ 4 files changed, 181 insertions(+), 36 deletions(-) create mode 100644 scripts/test-benchmark-export.py diff --git a/benchmark/collectors/capture.py b/benchmark/collectors/capture.py index 12a3be5..c3a5820 100644 --- a/benchmark/collectors/capture.py +++ b/benchmark/collectors/capture.py @@ -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 diff --git a/benchmark/export/__init__.py b/benchmark/export/__init__.py index 85cee5b..e84c8ad 100644 --- a/benchmark/export/__init__.py +++ b/benchmark/export/__init__.py @@ -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`。 -归档路径：`/.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）。归档路径 /.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()] diff --git a/orchestrator/requirements.txt b/orchestrator/requirements.txt index 3e3f19b..09f6622 100644 --- a/orchestrator/requirements.txt +++ b/orchestrator/requirements.txt @@ -15,3 +15,7 @@ opentelemetry-exporter-otlp==1.24.0 opentelemetry-instrumentation-redis==0.45b0 opentelemetry-instrumentation-requests==0.45b0 opentelemetry-instrumentation-logging==0.45b0 +# benchmark metrics export targets (lazy-imported; only used when BENCHMARK_EXPORT_TARGET set) +azure-cosmos==4.7.0 +azure-storage-blob==12.23.1 +azure-identity==1.19.0 diff --git a/scripts/test-benchmark-export.py b/scripts/test-benchmark-export.py new file mode 100644 index 0000000..d94aa11 --- /dev/null +++ b/scripts/test-benchmark-export.py @@ -0,0 +1,85 @@ +"""Test benchmark export target selection + doc shaping (no network). + +Verifies: + - BENCHMARK_EXPORT_TARGET parsing: default→noop, comma list→multiple, unknown ignored. + - a misconfigured target (missing creds) is skipped, not fatal, and never empties the list. + - CosmosExporter.build_doc shapes a Cosmos doc (string id = swarm_id; partition field present). + - constructors do NOT connect (no creds touched until export()). + +Run: python scripts/test-benchmark-export.py +""" +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmark.export import get_exporters, NoopExporter, CosmosExporter, BlobExporter + +failures = [] +_ENV = ("BENCHMARK_EXPORT_TARGET", "BENCHMARK_COSMOS_CONNECTION_STRING", + "BENCHMARK_COSMOS_DATABASE", "BENCHMARK_COSMOS_CONTAINER", + "AZURE_STORAGE_CONNECTION_STRING", "BENCHMARK_BLOB_ACCOUNT_URL", "BENCHMARK_BLOB_CONTAINER") + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +def reset(**env): + for k in _ENV: + os.environ.pop(k, None) + os.environ.update(env) + + +def names(exps): + return [e.name for e in exps] + + +def main(): + # default → noop + reset() + check("default → [noop]", names(get_exporters()) == ["noop"]) + + # explicit none + reset(BENCHMARK_EXPORT_TARGET="none") + check("none → [noop]", names(get_exporters()) == ["noop"]) + + # cosmos + blob, both with creds present → both built + reset(BENCHMARK_EXPORT_TARGET="cosmos,blob", + BENCHMARK_COSMOS_CONNECTION_STRING="AccountEndpoint=https://x.documents.azure.com:443/;AccountKey=FAKE==;", + AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=x;AccountKey=FAKE==;") + check("cosmos,blob → [cosmos, blob]", names(get_exporters()) == ["cosmos", "blob"]) + + # cosmos requested but NO connection string → skipped (not fatal); blob still built + reset(BENCHMARK_EXPORT_TARGET="cosmos,blob", + AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=x;AccountKey=FAKE==;") + check("misconfigured cosmos skipped, blob kept", names(get_exporters()) == ["blob"]) + + # all targets misconfigured → never empty, falls back to noop + reset(BENCHMARK_EXPORT_TARGET="cosmos") + check("all misconfigured → [noop] fallback", names(get_exporters()) == ["noop"]) + + # unknown target ignored → noop fallback + reset(BENCHMARK_EXPORT_TARGET="clickhouse") + check("unknown target → [noop] fallback", names(get_exporters()) == ["noop"]) + + # CosmosExporter.build_doc: string id = swarm_id, partition field swarm_id retained + doc = CosmosExporter.build_doc({"swarm_id": "swarm-123", "scenario": "coding", "metrics": {"s_gain": None}}) + check("cosmos doc id == swarm_id (str)", doc["id"] == "swarm-123" and isinstance(doc["id"], str)) + check("cosmos doc keeps swarm_id partition field", doc["swarm_id"] == "swarm-123") + check("cosmos doc preserves null metric (no fabrication)", doc["metrics"]["s_gain"] is None) + + reset() + print() + if failures: + print(f"{len(failures)} export check(s) FAILED: {failures}") + return 1 + print("all benchmark export checks passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main())