feat(observability): add file exporter writing standard OTLP JSON-agent-core-AtomGit

已合并

feat(observability): add file exporter writing standard OTLP JSON #1762

夏景创建于 6月27日从hwxiajing/agent-core： obsfile 合入到 openJiuwen/agent-core： develop

feat(observability): add file exporter writing standard OTLP JSON #1762

已合并

夏景创建于 6月27日

从hwxiajing/agent-core obsfile合入到openJiuwen/agent-core develop

共 6 个文件变更+785-6

Mdeploy/observability/README.md+87-0

	@@ -29,6 +29,93 @@ docker-compose logs -f otel-collector
29		29
30	Wait for all services to become healthy (~30-60s on first start).	30	Wait for all services to become healthy (~30-60s on first start).
31		31
		32	+## File Exporter (offline / two-phase)
		33	+
		34	+The `file` exporter writes traces as standard OTLP JSON files on disk
		35	+instead of streaming to the collector in real time. Files can be uploaded
		36	+later with `upload_traces_to_langfuse.py`.
		37	+
		38	+### Phase 1 — configure the exporter
		39	+
		40	+```python
		41	+from openjiuwen.agent_teams.observability import (
		42	+ ObservabilityConfig,
		43	+ init_observability,
		44	+)
		45	+
		46	+obs_config = ObservabilityConfig(
		47	+ enabled=True,
		48	+ exporter="file",
		49	+ traces_dir="./traces_run_001",
		50	+ file_retention_days=7, # optional, default 7
		51	+ sample_rate=1.0,
		52	+)
		53	+init_observability(obs_config)
		54	+```
		55	+
		56	+- Flat layout — all trace files are written directly under `traces_dir`,
		57	+ no per-session sub-folders.
		58	+- One append-only file per calendar day, named `traces-<YYYY-MM-DD>.jsonl`.
		59	+ Spans from all traces are interleaved in it; each line is a standalone
		60	+ single-span OTLP JSON (`resourceSpans` → `scopeSpans` → `spans`),
		61	+ directly ingestible by the collector at `/v1/traces` — replaying is
		62	+ just POSTing each line in turn. The collector splits traces by the
		63	+ `traceId` carried on every span, so interleaving is irrelevant for
		64	+ ingestion.
		65	+- `export()` appends straight to disk with no in-memory buffer; paired
		66	+ with `BatchSpanProcessor` (the default for the `file` exporter) so
		67	+ span-end does not block the business thread — spans land on disk when
		68	+ the processor flushes (default every 5s / 512 spans) and on shutdown.
		69	+- `session.id` (if present) is read by Langfuse from span attributes,
		70	+ not the filename — the filename carries only the date.
		71	+
		72	+### Phase 2 — upload to Langfuse
		73	+
		74	+The upload script accepts either a single trace file or a directory.
		75	+When given a directory, it uploads every `*.jsonl` directly under it
		76	+(flat — no sub-folder walking). It POSTs each line of each file as a
		77	+standalone OTLP request to the collector. After the run it prints the
		78	+unique trace IDs that were ingested, parsed from each uploaded line's
		79	+`resourceSpans[].scopeSpans[].spans[].traceId`.
		80	+
		81	+```bash
		82	+# Start the collector stack
		83	+docker-compose up -d
		84	+
		85	+# Upload a whole directory of trace files:
		86	+python upload_traces_to_langfuse.py ./traces_run_001
		87	+# or, equivalently:
		88	+python upload_traces_to_langfuse.py --dir ./traces_run_001
		89	+
		90	+# Upload a single trace file:
		91	+python upload_traces_to_langfuse.py --file ./traces_run_001/traces-2026-06-29.jsonl
		92	+
		93	+# Use a non-default collector endpoint:
		94	+python upload_traces_to_langfuse.py ./traces_run_001 --endpoint http://localhost:4318/v1/traces
		95	+```
		96	+
		97	+The script POSTs each `.jsonl` line to the collector's OTLP HTTP
		98	+endpoint (`localhost:4318/v1/traces` by default, no auth), which forwards
		99	+to Langfuse. Exit code is `0` on full success, `1` if any line failed,
		100	+`2` if the input path was not found or contained no `.jsonl` files.
		101	+
		102	+Sample output:
		103	+
		104	+```
		105	+[upload] source=./traces_run_001 files=3 endpoint=http://localhost:4318/v1/traces
		106	+[upload] total_lines=128 ok=128 fail=0 elapsed=0.6s
		107	+[upload] trace_ids (3):
		108	+ 4f3c1a8b9d2e4f6081a3c5b7d9e1f2a3
		109	+ 8e1c0b2d4f6a8c9e1d3b5a7c9e1f0a2b
		110	+ 1a2b3c4d5e6f70819203a4b5c6d7e8f9
		111	+```
		112	+
		113	+### Cleanup
		114	+
		115	+Trace files (`*.jsonl`) whose mtime predates `file_retention_days` are
		116	+lazily deleted by the exporter itself (a sweep runs at most every 64
		117	+exports). No manual cleanup is required.
		118	+
32	### Access Langfuse UI	119	### Access Langfuse UI
33		120
34	- URL: http://localhost:3000	121	- URL: http://localhost:3000

Adeploy/observability/upload_traces_to_langfuse.py+199-0

@@ -0,0 +1,199 @@
	1	+# coding: utf-8
	2	+"""Upload ``.jsonl`` trace files to Langfuse via the local OTel collector.
	3	+
	4	+Accepts either a single file or a directory:
	5	+
	6	+ python upload_traces_to_langfuse.py <path>
	7	+ python upload_traces_to_langfuse.py --file traces-2026-06-29.jsonl
	8	+ python upload_traces_to_langfuse.py --dir <traces_dir>
	9	+
	10	+Where ``<path>`` is:
	11	+ - a ``.jsonl`` file → upload every line in it
	12	+ - a directory → upload every ``*.jsonl`` directly under it (flat,
	13	+ no sub-folder walking)
	14	+
	15	+The file exporter writes one per-day ``traces-<YYYY-MM-DD>.jsonl`` whose
	16	+lines are spans from potentially many traces, interleaved. Every line
	17	+is a standalone OTLP JSON ``ExportTraceServiceRequest`` carrying a
	18	+single span — just POST each line to the collector. No reconstruction,
	19	+no merging. The collector splits traces by the ``traceId`` carried on
	20	+each span, so interleaving is irrelevant for ingestion.
	21	+``session.id`` (if present) is read by Langfuse from span attributes,
	22	+not the filename.
	23	+
	24	+After upload, the script prints the unique trace IDs ingested, parsed
	25	+from each uploaded line's ``resourceSpans[].scopeSpans[].spans[].traceId``.
	26	+
	27	+Prerequisites:
	28	+ docker-compose up -d # from deploy/observability/
	29	+
	30	+The collector listens on :4318 (OTLP HTTP, no auth).
	31	+"""
	32	+
	33	+from __future__ import annotations
	34	+
	35	+import argparse
	36	+import glob
	37	+import json
	38	+import os
	39	+import sys
	40	+import time
	41	+import urllib.error
	42	+import urllib.request
	43	+
	44	+_COLLECTOR = "http://localhost:4318/v1/traces"
	45	+
	46	+
	47	+def _iter_lines(path: str):
	48	+ """Yield non-empty stripped lines from a .jsonl file."""
	49	+ with open(path, "r", encoding="utf-8") as f:
	50	+ for line in f:
	51	+ line = line.strip()
	52	+ if line:
	53	+ yield line
	54	+
	55	+
	56	+def _extract_trace_id(line_body: bytes) -> str \| None:
	57	+ """Parse one OTLP JSON line and return its traceId, or None."""
	58	+ try:
	59	+ data = json.loads(line_body)
	60	+ except (ValueError, json.JSONDecodeError):
	61	+ return None
	62	+ for rs in data.get("resourceSpans", []):
	63	+ for ss in rs.get("scopeSpans", []):
	64	+ for sp in ss.get("spans", []):
	65	+ tid = sp.get("traceId")
	66	+ if isinstance(tid, str):
	67	+ return tid
	68	+ return None
	69	+
	70	+
	71	+def _post_line(body: bytes, endpoint: str) -> bool:
	72	+ """POST one OTLP JSON line to the collector. Returns True on success."""
	73	+ req = urllib.request.Request(endpoint, data=body, method="POST")
	74	+ req.add_header("Content-Type", "application/json")
	75	+ try:
	76	+ urllib.request.urlopen(req, timeout=15)
	77	+ return True
	78	+ except urllib.error.HTTPError as e:
	79	+ snippet = e.read()[:200]
	80	+ print(f"[upload] HTTP {e.code}: {snippet}", flush=True)
	81	+ except urllib.error.URLError as e:
	82	+ print(f"[upload] url error: {e}", flush=True)
	83	+ except OSError as e:
	84	+ print(f"[upload] net error: {e}", flush=True)
	85	+ return False
	86	+
	87	+
	88	+def _upload_one(path: str, endpoint: str) -> tuple[int, int, list[str]]:
	89	+ """Upload every line of one ``.jsonl`` file.
	90	+
	91	+ Returns (lines_ok, lines_fail, trace_ids) where trace_ids are the
	92	+ unique traceIds seen in successfully uploaded lines.
	93	+ """
	94	+ ok = 0
	95	+ fail = 0
	96	+ seen: set[str] = set()
	97	+ trace_ids: list[str] = []
	98	+ for line in _iter_lines(path):
	99	+ body = line.encode("utf-8")
	100	+ tid = _extract_trace_id(body)
	101	+ if _post_line(body, endpoint):
	102	+ ok += 1
	103	+ if tid and tid not in seen:
	104	+ seen.add(tid)
	105	+ trace_ids.append(tid)
	106	+ else:
	107	+ fail += 1
	108	+ return ok, fail, trace_ids
	109	+
	110	+
	111	+def _collect_files(path: str) -> list[str] \| None:
	112	+ """Return list of .jsonl files to upload from a file/dir path.
	113	+
	114	+ Returns None if path doesn't exist; empty list if dir has no .jsonl.
	115	+ Only top-level ``*.jsonl`` under a directory are picked up (flat
	116	+ layout — matches the file exporter's per-trace output).
	117	+ """
	118	+ if os.path.isfile(path):
	119	+ return [path]
	120	+ if os.path.isdir(path):
	121	+ return sorted(glob.glob(os.path.join(path, "*.jsonl")))
	122	+ return None
	123	+
	124	+
	125	+def _resolve_input_path(args: argparse.Namespace) -> str \| None:
	126	+ """Pick the input path from positional / --dir / --file (in that order)."""
	127	+ if args.path:
	128	+ return args.path
	129	+ if args.dir:
	130	+ return args.dir
	131	+ if args.file:
	132	+ return args.file
	133	+ return None
	134	+
	135	+
	136	+def main() -> int:
	137	+ parser = argparse.ArgumentParser(description="Upload per-trace .jsonl trace files to collector")
	138	+ parser.add_argument(
	139	+ "path",
	140	+ nargs="?",
	141	+ help="trace .jsonl file or directory containing .jsonl files",
	142	+ )
	143	+ parser.add_argument(
	144	+ "--dir",
	145	+ help="directory containing .jsonl trace files (alternative to positional path)",
	146	+ )
	147	+ parser.add_argument(
	148	+ "--file",
	149	+ help="single .jsonl trace file (alternative to positional path)",
	150	+ )
	151	+ parser.add_argument("--endpoint", default=_COLLECTOR)
	152	+ args = parser.parse_args()
	153	+
	154	+ path = _resolve_input_path(args)
	155	+ if not path:
	156	+ parser.error("provide a trace file/dir as a positional argument, or use --dir / --file")
	157	+
	158	+ files = _collect_files(path)
	159	+ if files is None:
	160	+ print(f"[upload] path not found: {path}", flush=True)
	161	+ return 2
	162	+ if not files:
	163	+ print(f"[upload] no *.jsonl found under {path}", flush=True)
	164	+ return 2
	165	+
	166	+ print(
	167	+ f"[upload] source={path} files={len(files)} endpoint={args.endpoint}",
	168	+ flush=True,
	169	+ )
	170	+
	171	+ total_ok = 0
	172	+ total_fail = 0
	173	+ uploaded_trace_ids: list[str] = []
	174	+ t0 = time.time()
	175	+ for fpath in files:
	176	+ ok, fail, trace_ids = _upload_one(fpath, args.endpoint)
	177	+ total_ok += ok
	178	+ total_fail += fail
	179	+ uploaded_trace_ids.extend(trace_ids)
	180	+
	181	+ elapsed = time.time() - t0
	182	+ print(
	183	+ f"[upload] total_lines={total_ok + total_fail} ok={total_ok} fail={total_fail} elapsed={elapsed:.1f}s",
	184	+ flush=True,
	185	+ )
	186	+
	187	+ unique_ids = list(dict.fromkeys(uploaded_trace_ids))
	188	+ if unique_ids:
	189	+ print(f"[upload] trace_ids ({len(unique_ids)}):", flush=True)
	190	+ for tid in unique_ids:
	191	+ print(f" {tid}", flush=True)
	192	+ else:
	193	+ print("[upload] no trace ids parsed from uploaded files", flush=True)
	194	+
	195	+ return 0 if total_fail == 0 and total_ok > 0 else 1
	196	+
	197	+
	198	+if __name__ == "__main__":
	199	+ sys.exit(main())

Mopenjiuwen/agent_teams/observability/config.py+16-2

	@@ -16,18 +16,29 @@ class ObservabilityConfig(BaseModel):
16	Attributes:	16	Attributes:
17	enabled: Master switch. When False, init_observability is a no-op.	17	enabled: Master switch. When False, init_observability is a no-op.
18	service_name: OTel resource attribute service.name.	18	service_name: OTel resource attribute service.name.
19	- exporter: Exporter backend type.	19	+ exporter: Exporter backend type. ``file`` writes OTLP JSON directly
		20	+ to ``traces_dir`` without a collector.
20	endpoint: OTLP endpoint URL (gRPC default localhost:4317; HTTP 4318).	21	endpoint: OTLP endpoint URL (gRPC default localhost:4317; HTTP 4318).
		22	+ Ignored when ``exporter`` is ``file``.
21	sample_rate: Parent-based ratio sampler rate (0.0 - 1.0).	23	sample_rate: Parent-based ratio sampler rate (0.0 - 1.0).
22	redact_prompts: When True, hash/truncate prompt contents.	24	redact_prompts: When True, hash/truncate prompt contents.
23	redact_completions: When True, hash/truncate completion contents.	25	redact_completions: When True, hash/truncate completion contents.
24	attribute_value_max_length: Hard cap on string attribute length.	26	attribute_value_max_length: Hard cap on string attribute length.
25	export_timeout_ms: Span exporter shutdown timeout.	27	export_timeout_ms: Span exporter shutdown timeout.
		28	+ traces_dir: Root directory for the ``file`` exporter. One
		29	+ append-only ``traces-<YYYY-MM-DD>.jsonl`` file per calendar
		30	+ day, written directly under this dir; each line is a
		31	+ standalone single-span OTLP JSON request. Spans from all
		32	+ traces share the file — the collector splits them by
		33	+ ``traceId`` on ingest. Paired with BatchSpanProcessor so
		34	+ span-end does not block the business thread.
		35	+ file_retention_days: Trace files older than this (by mtime) are
		36	+ lazily deleted by the ``file`` exporter. Default 7 days.
26	"""	37	"""
27		38
28	enabled: bool = True	39	enabled: bool = True
29	service_name: str = "openjiuwen-agent-teams"	40	service_name: str = "openjiuwen-agent-teams"
30	- exporter: Literal["otlp_grpc", "otlp_http", "console"] = "otlp_grpc"	41	+ exporter: Literal["otlp_grpc", "otlp_http", "console", "file"] = "otlp_grpc"
31	endpoint: str = "http://localhost:4317"	42	endpoint: str = "http://localhost:4317"
32	sample_rate: float = Field(default=1.0, ge=0.0, le=1.0)	43	sample_rate: float = Field(default=1.0, ge=0.0, le=1.0)
33	redact_prompts: bool = False	44	redact_prompts: bool = False
	@@ -37,3 +48,6 @@ class ObservabilityConfig(BaseModel):
37	# Langfuse authentication (for OTLP export via Langfuse OTLP endpoint)	48	# Langfuse authentication (for OTLP export via Langfuse OTLP endpoint)
38	langfuse_public_key: str = ""	49	langfuse_public_key: str = ""
39	langfuse_secret_key: str = ""	50	langfuse_secret_key: str = ""
		51	+ # file exporter
		52	+ traces_dir: str = "./traces"
		53	+ file_retention_days: int = 7

Aopenjiuwen/agent_teams/observability/file_exporter.py+184-0

@@ -0,0 +1,184 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	3	+
	4	+"""File-based SpanExporter that appends OTLP JSON lines, no collector.
	5	+
	6	+All spans are appended to a per-day file ``traces-<YYYY-MM-DD>.jsonl``
	7	+directly under ``root_dir``. Each line is a standalone OTLP JSON
	8	+``ExportTraceServiceRequest`` (``resourceSpans`` → ``scopeSpans`` →
	9	+``spans`` with a single span) carrying ``traceId``/``spanId``/
	10	+``parentSpanId`` as hex strings — the format Collector/Langfuse ingest
	11	+directly via ``POST /v1/traces``. Spans from different traces are
	12	+interleaved in the same file; the collector rebuilds each trace from
	13	+the ``traceId``/``parentSpanId`` carried on every span, so physical
	14	+ordering or per-trace file separation is irrelevant for ingestion.
	15	+Replaying is just POSTing each line in turn.
	16	+
	17	+Pair this exporter with ``BatchSpanProcessor`` (see ``setup.py``) so
	18	+span-end does not block the business thread: the processor batches
	19	+ended spans and calls :meth:`export` asynchronously (default every 5s
	20	+or 512 spans). ``export()`` appends straight to disk — no in-memory
	21	+buffer, so there is nothing to flush; ``force_flush`` / ``shutdown``
	22	+are no-ops.
	23	+
	24	+Trace files (``*.jsonl``) whose mtime predates ``retention_days`` are
	25	+lazily pruned at most every ``_CLEANUP_INTERVAL`` exports; cleanup never
	26	+raises.
	27	+"""
	28	+
	29	+from __future__ import annotations
	30	+
	31	+import base64
	32	+import binascii
	33	+import json
	34	+import os
	35	+import threading
	36	+import time
	37	+from typing import Any, Sequence
	38	+
	39	+from google.protobuf import json_format
	40	+from opentelemetry.exporter.otlp.proto.common._internal.trace_encoder import encode_spans
	41	+from opentelemetry.sdk.trace import ReadableSpan
	42	+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
	43	+
	44	+from openjiuwen.core.common.logging import team_logger
	45	+
	46	+# Cleanup runs at most every N export cycles to keep latency low.
	47	+_CLEANUP_INTERVAL = 64
	48	+_SECONDS_PER_DAY = 86400
	49	+# OTLP JSON requires traceId/spanId/parentSpanId as hex strings, but
	50	+# google.protobuf.json_format renders protobuf bytes as base64.
	51	+_HEX_ID_KEYS = frozenset({"traceId", "spanId", "parentSpanId"})
	52	+
	53	+
	54	+def _b64_to_hex(value: str) -> str:
	55	+ """Convert a base64-encoded id (MessageToDict output) to lower hex."""
	56	+ try:
	57	+ return binascii.hexlify(base64.b64decode(value)).decode()
	58	+ except Exception:
	59	+ return value
	60	+
	61	+
	62	+def _fix_hex_ids(node: Any) -> Any:
	63	+ """Recursively rewrite id fields from base64 to hex in an OTLP JSON dict."""
	64	+ if isinstance(node, dict):
	65	+ for key, val in list(node.items()):
	66	+ if key in _HEX_ID_KEYS and isinstance(val, str):
	67	+ node[key] = _b64_to_hex(val)
	68	+ else:
	69	+ _fix_hex_ids(val)
	70	+ elif isinstance(node, list):
	71	+ for item in node:
	72	+ _fix_hex_ids(item)
	73	+
	74	+
	75	+def _encode_span_line(span: ReadableSpan) -> str:
	76	+ """Encode a single ended span as one OTLP JSON line (hex ids, no indent)."""
	77	+ req = encode_spans([span])
	78	+ otlp_dict = json_format.MessageToDict(req)
	79	+ _fix_hex_ids(otlp_dict)
	80	+ return json.dumps(otlp_dict, ensure_ascii=False)
	81	+
	82	+
	83	+class TraceFileExporter(SpanExporter):
	84	+ """Append OTLP JSON lines to ``<root_dir>/traces-<YYYY-MM-DD>.jsonl``.
	85	+
	86	+ One append-only file per calendar day; spans from every trace share
	87	+ it. The collector splits traces by ``traceId`` on ingest, so no
	88	+ per-trace file separation is needed.
	89	+ """
	90	+
	91	+ def __init__(self, root_dir: str = "./traces", retention_days: int = 7) -> None:
	92	+ self.root_dir = root_dir
	93	+ self.retention_days = max(0, int(retention_days))
	94	+ # Serialize appends: BatchSpanProcessor may call export() from its
	95	+ # worker thread while shutdown runs on the main thread.
	96	+ self._lock = threading.Lock()
	97	+ self._write_count = 0
	98	+ try:
	99	+ os.makedirs(self.root_dir, exist_ok=True)
	100	+ except Exception as exc:
	101	+ team_logger.warning("file_exporter: cannot create traces_dir={} - {}", self.root_dir, exc)
	102	+
	103	+ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
	104	+ """Append every ended span (one OTLP JSON line each) to today's file.
	105	+
	106	+ Writes straight to disk — no buffering, so nothing is held back
	107	+ for a later flush. Called asynchronously by BatchSpanProcessor.
	108	+ """
	109	+ lines: list[str] = []
	110	+ for span in spans or ():
	111	+ # on_end only fires after end, but drop any not-yet-ended defensively.
	112	+ if getattr(span, "end_time", None) is None:
	113	+ continue
	114	+ lines.append(_encode_span_line(span))
	115	+
	116	+ if lines:
	117	+ file_path = os.path.join(self.root_dir, f"traces-{time.strftime('%Y-%m-%d')}.jsonl")
	118	+ try:
	119	+ with self._lock:
	120	+ with open(file_path, "a", encoding="utf-8") as f:
	121	+ for line in lines:
	122	+ f.write(line)
	123	+ f.write("\n")
	124	+ except OSError as exc:
	125	+ team_logger.warning("file_exporter: append failed to {} - {}", file_path, exc)
	126	+
	127	+ self._maybe_cleanup()
	128	+ return SpanExportResult.SUCCESS
	129	+
	130	+ def force_flush(self, timeout_millis: int = 30000) -> bool:
	131	+ """No-op: export() writes straight to disk, nothing buffered to flush."""
	132	+ return True
	133	+
	134	+ def shutdown(self) -> None:
	135	+ """No-op: nothing buffered; the last export() already hit disk."""
	136	+ return
	137	+
	138	+ # ------------------------------------------------------------------
	139	+ # internals
	140	+ # ------------------------------------------------------------------
	141	+
	142	+ def _maybe_cleanup(self) -> None:
	143	+ self._write_count += 1
	144	+ if self._write_count % _CLEANUP_INTERVAL != 0:
	145	+ return
	146	+ try:
	147	+ self._cleanup_old_files()
	148	+ except Exception as exc:
	149	+ team_logger.warning("file_exporter: cleanup failed - {}", exc)
	150	+
	151	+ def _cleanup_old_files(self) -> None:
	152	+ """Delete trace files whose mtime predates the retention cutoff.
	153	+
	154	+ ``FileNotFoundError`` is silently ignored in both the listdir and
	155	+ per-file steps (the dir may not exist yet; a file may have been
	156	+ removed by another process between listdir and stat/remove). Any
	157	+ other OSError is logged once and skipped so a single bad file
	158	+ can't abort the whole sweep.
	159	+ """
	160	+ if self.retention_days <= 0:
	161	+ return
	162	+ try:
	163	+ entries = os.listdir(self.root_dir)
	164	+ except FileNotFoundError:
	165	+ return
	166	+ except OSError as exc:
	167	+ team_logger.warning("file_exporter: cannot list {} - {}", self.root_dir, exc)
	168	+ return
	169	+
	170	+ cutoff = time.time() - self.retention_days * _SECONDS_PER_DAY
	171	+ for entry in entries:
	172	+ if not entry.endswith(".jsonl"):
	173	+ continue
	174	+ file_path = os.path.join(self.root_dir, entry)
	175	+ if not os.path.isfile(file_path):
	176	+ continue
	177	+ try:
	178	+ if os.path.getmtime(file_path) < cutoff:
	179	+ os.remove(file_path)
	180	+ except FileNotFoundError:
	181	+ # removed by another process between listdir and now
	182	+ continue
	183	+ except OSError as exc:
	184	+ team_logger.warning("file_exporter: cannot prune {} - {}", file_path, exc)

Mopenjiuwen/agent_teams/observability/setup.py+14-4文件内容审核中，请稍后刷新重试

Atests/unit_tests/agent_teams/observability/test_file_exporter.py+285-0

@@ -0,0 +1,285 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	3	+
	4	+"""Tests for the file-based SpanExporter (per-day OTLP JSON lines, no collector).
	5	+
	6	+The exporter appends every ended span (one OTLP JSON line each) straight
	7	+to a per-day ``traces-<YYYY-MM-DD>.jsonl`` file on ``export()`` — no
	8	+in-memory buffer, nothing deferred to flush. Each line is a standalone
	9	+``ExportTraceServiceRequest`` with hex traceId/spanId, ingestible by
	10	+Collector/Langfuse via POST /v1/traces. Spans from different traces
	11	+share the file; the collector splits traces by traceId on ingest.
	12	+"""
	13	+
	14	+from __future__ import annotations
	15	+
	16	+import json
	17	+import os
	18	+import time
	19	+from pathlib import Path
	20	+
	21	+import pytest
	22	+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
	23	+from opentelemetry.sdk.trace.export import SpanExportResult
	24	+from opentelemetry.trace import SpanKind, Status, StatusCode
	25	+
	26	+from openjiuwen.agent_teams.observability import ObservabilityConfig
	27	+from openjiuwen.agent_teams.observability.file_exporter import TraceFileExporter
	28	+
	29	+
	30	+def _make_span(
	31	+ name: str = "test.span",
	32	+ *,
	33	+ session_id: str \| None = "sess-1",
	34	+) -> ReadableSpan:
	35	+ """Build a finished ReadableSpan with the given attributes."""
	36	+ tracer = TracerProvider().get_tracer("ut")
	37	+ span = tracer.start_span(name, kind=SpanKind.INTERNAL)
	38	+ if session_id is not None:
	39	+ span.set_attribute("session.id", session_id)
	40	+ span.set_status(Status(StatusCode.OK))
	41	+ span.end()
	42	+ return span
	43	+
	44	+
	45	+def _day_file(tmp_path: Path) -> Path:
	46	+ """The per-day file path the exporter writes to today."""
	47	+ return tmp_path / f"traces-{time.strftime('%Y-%m-%d')}.jsonl"
	48	+
	49	+
	50	+def _read_spans(path: Path) -> list[dict]:
	51	+ """Read a ``.jsonl`` file and return the union of spans across all lines.
	52	+
	53	+ Each line is its own ``ExportTraceServiceRequest``; spans from every
	54	+ line are concatenated in file order.
	55	+ """
	56	+ spans: list[dict] = []
	57	+ for line in path.read_text(encoding="utf-8").splitlines():
	58	+ line = line.strip()
	59	+ if not line:
	60	+ continue
	61	+ data = json.loads(line)
	62	+ for rs in data.get("resourceSpans", []):
	63	+ for ss in rs.get("scopeSpans", []):
	64	+ spans.extend(ss.get("spans", []))
	65	+ return spans
	66	+
	67	+
	68	+# ---------------------------------------------------------------------------
	69	+# Exporter in isolation
	70	+# ---------------------------------------------------------------------------
	71	+
	72	+
	73	+def test_export_writes_immediately(tmp_path: Path) -> None:
	74	+ """export() appends straight to today's file — no buffering, no flush needed."""
	75	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	76	+ span = _make_span("llm.call", session_id="abc")
	77	+
	78	+ result = exporter.export([span])
	79	+ assert result == SpanExportResult.SUCCESS
	80	+ # file exists right after export() — no force_flush required
	81	+ f = _day_file(tmp_path)
	82	+ assert f.exists()
	83	+ spans = _read_spans(f)
	84	+ assert len(spans) == 1
	85	+ assert spans[0]["name"] == "llm.call"
	86	+
	87	+
	88	+def test_each_line_is_valid_otlp_json_with_hex_ids(tmp_path: Path) -> None:
	89	+ """Each appended line is OTLP JSON with hex traceId/spanId (not base64)."""
	90	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	91	+ span = _make_span("llm.call", session_id="abc")
	92	+ exporter.export([span])
	93	+
	94	+ lines = [ln for ln in _day_file(tmp_path).read_text("utf-8").splitlines() if ln.strip()]
	95	+ assert len(lines) == 1
	96	+ data = json.loads(lines[0])
	97	+ assert "resourceSpans" in data
	98	+ sp = data["resourceSpans"][0]["scopeSpans"][0]["spans"][0]
	99	+ assert sp["name"] == "llm.call"
	100	+ assert len(sp["traceId"]) == 32
	101	+ assert all(c in "0123456789abcdef" for c in sp["traceId"]), "traceId must be hex"
	102	+ assert len(sp["spanId"]) == 16
	103	+ assert all(c in "0123456789abcdef" for c in sp["spanId"]), "spanId must be hex"
	104	+
	105	+
	106	+def test_spans_of_same_trace_interleaved_in_one_file(tmp_path: Path) -> None:
	107	+ """Two spans of the same trace land as two lines in today's file;
	108	+ parentSpanId still links child to parent."""
	109	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	110	+ provider = TracerProvider()
	111	+ tracer = provider.get_tracer("ut")
	112	+ from opentelemetry.trace import set_span_in_context
	113	+
	114	+ parent = tracer.start_span("parent", kind=SpanKind.INTERNAL)
	115	+ parent.set_attribute("session.id", "abc")
	116	+ parent.set_status(Status(StatusCode.OK))
	117	+ child = tracer.start_span("child", kind=SpanKind.INTERNAL, context=set_span_in_context(parent))
	118	+ child.set_attribute("session.id", "abc")
	119	+ child.set_status(Status(StatusCode.OK))
	120	+ parent.end()
	121	+ child.end()
	122	+
	123	+ exporter.export([parent])
	124	+ exporter.export([child])
	125	+
	126	+ spans = _read_spans(_day_file(tmp_path))
	127	+ assert len(spans) == 2
	128	+ names = {s["name"] for s in spans}
	129	+ assert names == {"parent", "child"}
	130	+ parent_sp = next(s for s in spans if s["name"] == "parent")
	131	+ child_sp = next(s for s in spans if s["name"] == "child")
	132	+ assert child_sp.get("parentSpanId") == parent_sp["spanId"]
	133	+
	134	+
	135	+def test_spans_of_different_traces_share_one_file(tmp_path: Path) -> None:
	136	+ """Spans from different traces are interleaved in the same per-day file;
	137	+ each carries its own traceId so the collector can split them."""
	138	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	139	+ s1 = _make_span("a.span", session_id="s1")
	140	+ s2 = _make_span("b.span", session_id="s2")
	141	+ exporter.export([s1])
	142	+ exporter.export([s2])
	143	+
	144	+ spans = _read_spans(_day_file(tmp_path))
	145	+ assert len(spans) == 2
	146	+ trace_ids = {s["traceId"] for s in spans}
	147	+ assert len(trace_ids) == 2, "two distinct traces in one file"
	148	+
	149	+
	150	+def test_no_session_attribute_still_written(tmp_path: Path) -> None:
	151	+ """A span without session.id is appended like any other — no fallback needed."""
	152	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	153	+ span = _make_span("orphan.span", session_id=None)
	154	+ exporter.export([span])
	155	+ assert _read_spans(_day_file(tmp_path))[0]["name"] == "orphan.span"
	156	+
	157	+
	158	+def test_repeated_export_appends_no_duplication(tmp_path: Path) -> None:
	159	+ """Two exports of the same span produce two distinct lines (no dedup needed
	160	+ — append semantics). Two exports of different spans just accumulate."""
	161	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	162	+ span = _make_span("x.span", session_id="abc")
	163	+ exporter.export([span])
	164	+ exporter.force_flush() # no-op; must not drop or duplicate the line
	165	+ spans = _read_spans(_day_file(tmp_path))
	166	+ assert len(spans) == 1
	167	+
	168	+
	169	+def test_shutdown_is_noop_does_not_lose_data(tmp_path: Path) -> None:
	170	+ """shutdown is a no-op (nothing buffered); data already on disk stays."""
	171	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	172	+ span = _make_span("late.span", session_id="abc")
	173	+ exporter.export([span])
	174	+ exporter.shutdown()
	175	+ assert len(_read_spans(_day_file(tmp_path))) == 1
	176	+
	177	+
	178	+def test_cleanup_deletes_old_trace_files(tmp_path: Path) -> None:
	179	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=1)
	180	+ span = _make_span("old.span", session_id="old")
	181	+ exporter.export([span])
	182	+
	183	+ old_file = _day_file(tmp_path)
	184	+ assert old_file.is_file()
	185	+ old_time = time.time() - 2 * 86400
	186	+ os.utime(old_file, (old_time, old_time))
	187	+
	188	+ exporter._cleanup_old_files()
	189	+ assert not old_file.exists()
	190	+
	191	+
	192	+def test_cleanup_keeps_recent_trace_files(tmp_path: Path) -> None:
	193	+ exporter = TraceFileExporter(root_dir=str(tmp_path), retention_days=7)
	194	+ span = _make_span("fresh.span", session_id="fresh")
	195	+ exporter.export([span])
	196	+ exporter._cleanup_old_files()
	197	+ assert _day_file(tmp_path).is_file()
	198	+
	199	+
	200	+# ---------------------------------------------------------------------------
	201	+# End-to-end through init_observability / _build_exporter
	202	+# ---------------------------------------------------------------------------
	203	+
	204	+
	205	+@pytest.fixture
	206	+def file_config(tmp_path: Path) -> ObservabilityConfig:
	207	+ return ObservabilityConfig(
	208	+ enabled=True,
	209	+ exporter="file",
	210	+ traces_dir=str(tmp_path / "traces"),
	211	+ file_retention_days=7,
	212	+ sample_rate=1.0,
	213	+ )
	214	+
	215	+
	216	+def test_build_exporter_returns_trace_file_exporter(file_config: ObservabilityConfig) -> None:
	217	+ from openjiuwen.agent_teams.observability.setup import _build_exporter
	218	+
	219	+ exporter = _build_exporter(file_config)
	220	+ assert isinstance(exporter, TraceFileExporter)
	221	+ assert exporter.root_dir == file_config.traces_dir
	222	+ assert exporter.retention_days == 7
	223	+
	224	+
	225	+def test_init_observability_writes_per_day_jsonl(file_config: ObservabilityConfig) -> None:
	226	+ """init_observability with exporter=file lands a per-day ``.jsonl`` on
	227	+ disk after shutdown; each line is a hex-id OTLP JSON request. Pair it
	228	+ with BatchSpanProcessor (setup.py default) — spans flush to disk on
	229	+ provider shutdown."""
	230	+ import asyncio
	231	+
	232	+ asyncio.run(_e2e_async(file_config))
	233	+
	234	+
	235	+async def _e2e_async(file_config: ObservabilityConfig) -> None:
	236	+ from openjiuwen.agent_teams.observability import init_observability, shutdown_observability
	237	+ from openjiuwen.agent_teams.observability.setup import get_tracer
	238	+ from openjiuwen.agent_teams.observability.span_context import get_or_create_team_span, remove_team_span
	239	+ from openjiuwen.core.runner import Runner
	240	+ from openjiuwen.core.runner.callback.events import LLMCallEvents
	241	+
	242	+ class _FakeUsage:
	243	+ input_tokens = 12
	244	+ output_tokens = 7
	245	+ total_tokens = 19
	246	+ model_name = "fake-llm-1"
	247	+
	248	+ class _FakeAssistantMessage:
	249	+ def __init__(self) -> None:
	250	+ self.content = "hello"
	251	+ self.reasoning_content = ""
	252	+ self.finish_reason = "stop"
	253	+ self.tool_calls = None
	254	+ self.usage_metadata = _FakeUsage()
	255	+
	256	+ init_observability(file_config)
	257	+ try:
	258	+ get_or_create_team_span("e2e_team", get_tracer("openjiuwen.agent_teams.observability"))
	259	+ fw = Runner.callback_framework
	260	+ messages = [{"role": "user", "content": "hi"}]
	261	+ await fw.trigger(LLMCallEvents.LLM_INVOKE_INPUT, messages=messages, model="fake-llm-1")
	262	+ await fw.trigger(
	263	+ LLMCallEvents.LLM_INVOKE_OUTPUT,
	264	+ messages=messages,
	265	+ result=_FakeAssistantMessage(),
	266	+ )
	267	+ remove_team_span("e2e_team")
	268	+ finally:
	269	+ shutdown_observability()
	270	+
	271	+ traces_root = Path(file_config.traces_dir)
	272	+ jsonl_files = list(traces_root.glob("*.jsonl"))
	273	+ assert jsonl_files, "no .jsonl trace file written"
	274	+ for jf in jsonl_files:
	275	+ for line in jf.read_text("utf-8").splitlines():
	276	+ line = line.strip()
	277	+ if not line:
	278	+ continue
	279	+ data = json.loads(line)
	280	+ assert "resourceSpans" in data
	281	+ for rs in data["resourceSpans"]:
	282	+ for ss in rs["scopeSpans"]:
	283	+ for sp in ss["spans"]:
	284	+ tid = sp.get("traceId", "")
	285	+ assert len(tid) == 32 and all(c in "0123456789abcdef" for c in tid)