guke
129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""接口指标后台上报 worker:批量 drain 事件队列 → POST 到 OpenObserve。
|
|
|
|
对齐 heartbeat_monitor_worker 等的 start_*/stop_* 形态。best-effort 遥测:catch 全部
|
|
异常,上报失败直接丢批不重试。未配置观测 → start 返回 None(不启动),整套 no-op。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
from app.core.observe import get_queue, take_dropped
|
|
|
|
logger = logging.getLogger("shagua.observe")
|
|
|
|
# 上报用的 httpx client,start 时建、stop 时关。
|
|
_client: httpx.AsyncClient | None = None
|
|
|
|
|
|
async def _collect_batch() -> list[dict]:
|
|
"""等到 ≥1 条(或到 flush 间隔)后,连抽到 BATCH_MAX 条或抽空。超时且空 → 返回 []。"""
|
|
queue = get_queue()
|
|
batch: list[dict] = []
|
|
try:
|
|
first = await asyncio.wait_for(
|
|
queue.get(), timeout=settings.OBSERVE_FLUSH_INTERVAL_SEC
|
|
)
|
|
except asyncio.TimeoutError: # noqa: UP041 - 3.10 兼容:该版 wait_for 抛的 asyncio.TimeoutError ≠ 内置 TimeoutError
|
|
return batch
|
|
batch.append(first)
|
|
while len(batch) < settings.OBSERVE_BATCH_MAX:
|
|
try:
|
|
batch.append(queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
return batch
|
|
|
|
|
|
async def _post_batch(client: httpx.AsyncClient, batch: list[dict]) -> None:
|
|
"""POST 一批事件到 OpenObserve 的 _json ingest 端点。非 2xx 仅告警。"""
|
|
url = f"/api/{settings.OBSERVE_ORG}/{settings.OBSERVE_STREAM}/_json"
|
|
resp = await client.post(url, json=batch)
|
|
if resp.status_code >= 300:
|
|
logger.warning(
|
|
"observe ingest failed status=%s body=%s",
|
|
resp.status_code,
|
|
resp.text[:200],
|
|
)
|
|
|
|
|
|
async def _run_loop(client: httpx.AsyncClient) -> None:
|
|
try:
|
|
while True:
|
|
batch = await _collect_batch()
|
|
dropped = take_dropped()
|
|
if dropped:
|
|
logger.warning("observe dropped %d events (queue full)", dropped)
|
|
if not batch:
|
|
continue
|
|
try:
|
|
await _post_batch(client, batch)
|
|
except Exception: # noqa: BLE001 - best-effort 遥测,失败丢批不重试、不退出
|
|
logger.warning(
|
|
"observe post batch failed, dropped %d events",
|
|
len(batch),
|
|
exc_info=True,
|
|
)
|
|
except asyncio.CancelledError:
|
|
logger.info("observe worker stopped")
|
|
raise
|
|
|
|
|
|
def start_observe_worker() -> asyncio.Task | None:
|
|
"""启动上报 worker。未配置观测 → 返回 None(no-op)。约定每进程只调一次(lifespan)。"""
|
|
global _client
|
|
if not settings.observe_configured:
|
|
return None
|
|
if _client is not None:
|
|
# 约定 start 每进程只调一次;已启动则不重复建 client(避免泄漏旧连接池)。
|
|
logger.warning("observe worker already started; ignoring duplicate start")
|
|
return None
|
|
_client = httpx.AsyncClient(
|
|
base_url=settings.OBSERVE_ENDPOINT,
|
|
auth=(settings.OBSERVE_USER, settings.OBSERVE_PASSWORD),
|
|
timeout=settings.OBSERVE_TIMEOUT_SEC,
|
|
)
|
|
logger.info(
|
|
"observe worker started endpoint=%s org=%s stream=%s",
|
|
settings.OBSERVE_ENDPOINT,
|
|
settings.OBSERVE_ORG,
|
|
settings.OBSERVE_STREAM,
|
|
)
|
|
return asyncio.create_task(_run_loop(_client), name="observe-worker")
|
|
|
|
|
|
async def stop_observe_worker(task: asyncio.Task | None) -> None:
|
|
"""收尾:cancel worker → best-effort 发最后一批 → 关 client。"""
|
|
global _client
|
|
if task is None:
|
|
return
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|
|
dropped = take_dropped() # 收口:补记最后一个 flush 窗口累计的丢弃数,不让账丢在关停期
|
|
if dropped:
|
|
logger.warning("observe dropped %d events (queue full) before shutdown", dropped)
|
|
if _client is not None:
|
|
# worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停);
|
|
# 超过一批(BATCH_MAX)的剩余直接丢,不做多轮 flush(best-effort,关停从速)。
|
|
try:
|
|
queue = get_queue()
|
|
final: list[dict] = []
|
|
while len(final) < settings.OBSERVE_BATCH_MAX:
|
|
try:
|
|
final.append(queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
if final:
|
|
await asyncio.wait_for(
|
|
_post_batch(_client, final), timeout=settings.OBSERVE_TIMEOUT_SEC
|
|
)
|
|
except Exception: # noqa: BLE001 - 关停期尽力而为,失败忽略
|
|
pass
|
|
await _client.aclose()
|
|
_client = None
|