AgentOps v4 API 日志端点解析任务设计、路由实现与测试验证【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops本篇技术文章围绕 AgentOps 仓库中 v4 API 的日志端点任务文档04_log_endpoints.md展开梳理其端点设计、查询参数与响应格式的规划并对照仓库中已落地的路由实现上传端点与按 Trace 读取端点、S3 兼容对象存储基类以及配套单元测试完整呈现从任务文档到可运行代码的演进过程。读完后你可以掌握 v4 日志端点的鉴权方式、文件大小与字符集约束、免费计划截断机制以及如何通过现有测试验证其正确性。一、背景v4 API 与日志数据的迁移v4 API 是 AgentOps 从 SupabasePostgres数据链路向 OpenTelemetry Clickhouse 数据链路迁移的一部分。v4 API 任务总览 明确说明v4 端点将取代查询 Supabase 的 v2 端点新端点从 Clickhouse 中查询 OpenTelemetry 的 trace、log 和 metric 数据同时保持与 v2 的向后兼容支持渐进式迁移。任务按依赖顺序排列为 10 个Clickhouse Client、Authentication Middleware、Trace Endpoints、Log Endpoints、Metric Endpoints、Session Endpoints、Computed Fields Migration、API Documentation、Testing and Validation、Deployment and Monitoring。日志端点处于第 4 位其前置依赖是 Clickhouse 客户端实现与鉴权中间件。与此相对v2 时代的日志链路在 v2 路由 中以PUT /update_logs实现请求携带 Bearer JWT服务端从 Supabasesessions表查出project_id将日志内容追加到 Supabase Storage 中session-logs桶的{project_id}/{session_id}.txt再把 public URL 写回sessions.logs_url字段。v4 的设计目标是把日志按Trace 维度组织并统一走 S3 兼容的对象存储接口。二、任务文档中的端点规划与查询参数任务文档 给出了明确的需求清单与端点规划以下是其核心内容的完整继承2.1 需求与规划端点实现日志数据查询端点与现有 v2 端点保持兼容针对查询性能做优化实现过滤filtering、排序sorting与分页pagination处理错误场景。规划实现的三个端点端点用途GET /v4/logs获取日志列表GET /v4/logs/{trace_id}获取指定 trace 的日志GET /v4/logs/search基于条件搜索日志2.2 规划中的查询参数任务文档列出了 10 个查询参数完整列表如下project_id按项目 ID 过滤trace_id按 trace ID 过滤start_time/end_time按起止时间过滤service_name按服务名过滤severity_text/severity_number按日志严重级别文本/数值过滤——这两个字段对应 OpenTelemetry 日志数据模型中的SeverityText与SeverityNumber属性body_contains按日志正文内容做包含匹配limit限制返回结果数量offset分页偏移量。响应格式要求返回带日志数据的 JSON 响应附带分页元数据pagination metadata并包含指向相关资源的链接。测试要求包括为端点编写单元测试、用 Clickhouse 中的真实数据测试、用大数据集做性能测试、测试错误处理。预估工时 6-8 小时。2.3 落地形态规划与实现的差异从源码结构看当前仓库中 v4 日志端点的实际落地与任务文档的规划存在明确分工GET /v4/logs/{trace_id}已实现日志的写入通过POST /v4/logs/upload/完成而从 Clickhouse 做列表/搜索查询的GET /v4/logs与GET /v4/logs/search在 v4 路由注册表中尚未出现见下文路由清单。实际的日志存储采用 Supabase Storage 的 S3 兼容接口bucket 按{trace_id}.log命名而非直接对 Clickhouse 的 OTel logs 表做 SQL 查询——任务文档中用 Clickhouse 客户端查询日志的实现路线在日志链路上被对象存储 trace 维度取回的方案替代。这一点在阅读代码时需要特别注意避免被任务文档的原始措辞误导。三、已实现的端点之一POST /v4/logs/upload/3.1 端点定位LogsUploadView 是 SDK 侧日志上报端点。文件头注释说明其职责接收 AgentOps SDK 发来的日志文件内容JWT 鉴权存储后返回可公开访问的 URL底层使用 Supabase Bucket 的 S3 接口存储。该视图以public_route装饰器声明为公开路由继承自 BaseObjectUploadViewpublic_route class LogsUploadView(BaseObjectUploadView): bucket_name: str SUPABASE_S3_LOGS_BUCKET property def filename(self) - str: Generate a unique filename for the object trace_id self.request.headers.get(Trace-Id) if not trace_id: raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailNo trace ID provided, ) # only allow alphanumeric characters, underscores, dashes, and dots if re.search(r[^a-zA-Z0-9_.-], trace_id): raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailTrace ID contains invalid characters, ) return f{trace_id}.log关键设计点文件名即 trace ID对象以{trace_id}.log命名这使得后续按 trace 读取日志只需一次 S3get_object无需索引表——这是 v4 日志链路按 trace 维度组织的直接体现字符集白名单Trace-Id请求头只允许字母、数字、下划线、连字符和点号防止路径穿越类注入S3 Key 被直接拼进对象路径存储桶SUPABASE_S3_LOGS_BUCKET从环境变量读取定义在 api/environment.py。3.2 上传流程与大小限制BaseObjectUploadViewstorage.py 中的BaseObjectUploadView提供通用的分块读取与限流框架LogsUploadView只需覆写filenameclass BaseObjectUploadView(BaseView, ABC): bucket_name: str max_size: int 25 * 1024 * 1024 # 25 MB async def __call__(self, token: JWTPayload Depends(get_jwt_token)) - ObjectUploadResponse: ... body BytesIO() total_size 0 # read the body in chunks so we dont ever load an entire oversized file into memory async for chunk in self.request.stream(): total_size len(chunk) if total_size self.max_size: logger.error(Uploaded file exceeds maximum size limit) raise HTTPException( status_codestatus.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detailfFile size exceeds the maximum limit of {self.max_size} bytes, ) body.write(chunk) ... return ObjectUploadResponse(urlself.public_url, sizetotal_size)要点默认 25 MB 上限以代码中的max_size 25 * 1024 * 1024为准类 docstring 中Defaults to 10 MB的注释与代码不一致属过期文档分块流式读取async for chunk in self.request.stream()边读边累计字节数超限立即抛 413避免超大文件整体载入内存——这正对应任务文档处理错误场景中超大请求一类错误用例S3 客户端单例get_s3_client()用 boto3 指向{SUPABASE_URL}/storage/v1/s3即 Supabase Storage 的 S3 兼容端点签名版本 s3v4响应模型ObjectUploadResponse返回url{SUPABASE_URL}/storage/v1/object/public/{bucket}/{filename}与size与任务文档返回数据可访问的公共 URL的响应要求一致。四、已实现的端点之二GET /v4/logs/{trace_id}4.1 端点代码与鉴权链get_trace_logs 实现按 trace 读取日志全文其处理顺序严格对应任务文档验证用户有权访问后才返回日志的要求add_cors_headers(origins[APP_URL], methods[GET, OPTIONS]) async def get_trace_logs( *, request: Request, orm: Session Depends(get_orm_session), trace_id: str, ) - LogContentResponse: Retrieve logs for a specific trace ID. Verifies that the user has access to the trace before returning the logs. trace_id_int convert_trace_id(trace_id) trace await TraceModel.select(filters{trace_id: trace_id}) if not trace.spans: # trace does not exist raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detailYou do not have access to this trace, ) project ProjectModel.get_by_id(orm, trace.project_id) if not project or not project.org.is_user_member(request.state.session.user_id): raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detailYou do not have access to this trace, ) try: s3_client get_s3_client() response s3_client.get_object( BucketSUPABASE_S3_LOGS_BUCKET, Keyf{trace_id_int}.log, ) content response[Body].read().decode(utf-8) return LogContentResponse( contentcontent, trace_idtrace_id, freeplan_truncatedproject.is_freeplan, ) except s3_client.exceptions.NoSuchKey: raise HTTPException( status_codestatus.HTTP_404_NOT_FOUND, detailfNo logs found for trace ID: {trace_id}, )处理链可以拆解为四步trace 存在性检查先查 SupabaseTraceModel若该 trace 无 spans 则视为不存在返回 403而非 404避免暴露资源是否存在组织成员鉴权取 trace 所属project_id确认当前会话用户属于该项目的组织否则同样 403S3 取回日志Key 使用经convert_trace_id转换后的{trace_id_int}.logNoSuchKey时返回 404响应构造返回LogContentResponse含content、trace_id字段。注意 CORS 装饰器add_cors_headers只放行APP_URL官方 dashboard 域名的 GET/OPTIONS 请求即此端点主要服务于 dashboard 的 trace 日志查看页面而不是对外开放的查询 API。4.2convert_trace_id十六进制 Trace ID 到十进制 Key 的转换def convert_trace_id(trace_id: str) - str: Convert hex trace_id to int if in hex format (contains at least one letter a-f). try: # Only convert if string contains hex letters and is valid hex if any(c in abcdefABCDEF for c in trace_id) and all( c in 0123456789abcdefABCDEF for c in trace_id ): return str(int(trace_id, 16)) return trace_id except ValueError: return trace_idOpenTelemetry 的trace_id是 128-bit 值在 W3C Trace Context 中以 32 位十六进制字符串传输。SDK 上传侧可能以十六进制或十进制整数形式命名对象 Key此函数保证两种形式都能命中同一个对象含字母的合法 hex 串转十进制纯数字串原样保留非法 hex 原样返回。该规则有对应的边界用例覆盖hex 转换、纯数字不变、非法串不变、空串、混合大小写、长 hex见 test_convert_trace_id_hex_to_int。4.3 免费计划行截断响应模型继承了FreePlanFilteredResponse并声明了按字段截断规则class LogContentResponse(FreePlanFilteredResponse): _freeplan_maxlines { content: FREEPLAN_LOGS_LINE_LIMIT, } content: str trace_id: strFREEPLAN_LOGS_LINE_LIMIT在 common/environment.py 中定义默认值为100 行环境变量可调。即免费计划项目取回的日志内容最多保留 100 行响应中携带freeplan_truncated标志位表明是否发生了截断。这是对任务文档响应格式一节中JSON 响应 元数据要求在商业层的一个具体化实现。五、路由注册v4 日志端点在哪里挂接v4 路由入口 展示了当前已注册的完整 v4 端点清单日志相关的两条是router APIRouter(prefix/v4) route_config: list[RouteConfig] [ # Metrics RouteConfig(nameget_project_metrics, path/meterics/project/{project_id}, endpointProjectMetricsView, methods[GET]), # Traces RouteConfig(nameget_project_traces, path/traces/list/{project_id}, endpointTraceListView, methods[GET]), RouteConfig(nameget_trace, path/traces/detail/{trace_id}, endpointTraceDetailView, methods[GET]), # Objects RouteConfig(nameupload_object, path/objects/upload/, endpointObjectUploadView, methods[POST]), # Logs RouteConfig(nameupload_logs, path/logs/upload/, endpointLogsUploadView, methods[POST]), RouteConfig(nameget_trace_logs, path/logs/{trace_id}, endpointget_trace_logs, methods[GET]), ] api_router APIRouter(route_classAuthenticatedRoute) register_routes(api_router, route_config, prefix/v4) router.include_router(api_router)两点值得注意全部业务路由挂在AuthenticatedRouteJWT 鉴权中间件之下对应任务总览中v4 端点使用与 v3 相同的鉴权机制API key 换 JWT再用 JWT 访问 v4的设计对照任务文档规划GET /v4/logs列表与GET /v4/logs/search搜索未出现在route_config中——从源码结构看日志的过滤/排序/分页能力在当前实现中尚未通过 HTTP 端点暴露后续可按任务文档的查询参数表在 Clickhouse 查询链路上补齐。六、错误处理全景四种错误码的触发条件综合上传端点与读取端点任务文档要求的错误场景处理在实现上对应如下错误码矩阵均已在测试中验证状态码触发条件来源400 Bad Request上传请求缺少Trace-Id头或 Trace-Id 含白名单外字符LogsUploadView.filename403 Forbiddentrace 不存在无 spans或用户不属于该 trace 所属项目的组织get_trace_logs404 Not FoundS3 中不存在{trace_id}.log对象NoSuchKeyget_trace_logs413 Payload Too Large上传内容超过max_size默认 25 MBBaseObjectUploadView.__call__403 同时用于trace 不存在与无权限两种情形是一种有意避免信息泄露的错误归并而 404 仅用于trace 有效但日志对象缺失这一数据缺失场景。七、测试验证tests/v4/test_logs.py单元测试文件 覆盖了两类端点的关键路径可作为验证实现是否符合任务文档的对照清单LogsUploadView 侧TestLogsUploadView桶配置断言view.bucket_name SUPABASE_S3_LOGS_BUCKET成功上传mock 请求流写入测试内容后断言响应为ObjectUploadResponse、size等于内容字节数、public URL 形如{SUPABASE_URL}/storage/v1/object/public/{bucket}/{trace_id}.log并验证 S3upload_fileobj以正确的 bucket 与{trace_id}.log被调用一次缺少Trace-Id断言 400 且 detail 含 No trace ID provided非法字符trace123、trace#456、trace 789、trace/abc、trace\def五个用例均断言 400 Trace ID contains invalid characters合法字符trace123、trace-456、trace_789、trace.abc、混合大小写等断言生成{id}.log大小限制将view.max_size设为 100 字节后上传 150 字节内容断言 413 File size exceeds the maximum limit。get_trace_logs 侧TestGetGetTraceLogs对应类convert_trace_id的 hex→int、纯数字、非法串、空串、混合大小写、长 hex 各边界用例成功取回mock S3 返回日志内容后断言 JSON 响应中content、trace_id正确且freeplan_truncated为False并验证get_object以转换后的 Key 被调用trace 不存在spans为空断言 403用户非组织成员断言 403S3NoSuchKey断言 404 No logs found for trace ID: ...。八、小结任务文档与落地实现的映射关系把 04_log_endpoints.md 的规划与当前仓库代码逐条对照可以得到如下映射端点规划GET /v4/logs/{trace_id}已落地POST /v4/logs/upload/作为写入侧补齐了规划中缺失的日志上报链路GET /v4/logs列表与GET /v4/logs/search搜索暂未注册查询参数表project_id、severity_text、body_contains等 10 项属于 Clickhouse 查询链路的规划当前 v4 日志读取端点仅以trace_id路径参数取回全文过滤/分页参数尚未启用响应格式上传端点返回{url, size}读取端点返回{content, trace_id, freeplan_truncated}免费计划按 100 行FREEPLAN_LOGS_LINE_LIMIT默认值截断错误处理400/403/404/413 四类错误均有实现与测试覆盖依赖项Clickhouse 客户端依赖在日志链路上被 S3 兼容存储storage.py 的 boto3 单例替代鉴权依赖由AuthenticatedRoute与get_jwt_token依赖注入满足。对后续维护者而言这份任务文档的价值在于保留了日志端点的完整参数规划——当 Clickhouse OTel logs 查询链路任务 1、9成熟后GET /v4/logs与GET /v4/logs/search可以按第二节的参数表直接补齐而无需重新设计命名与语义。【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考