创业团队的可观测性体系Metrics、Traces与Logs的三位一体设计一、引言可观测性是技术团队的眼睛。没有可观测性线上故障靠用户投诉才知道排查问题靠猜。创业团队尤其需要低成本高价值的可观测性方案。可观测性不等于监控。监控告诉你出问题了可观测性告诉你为什么出问题。本文拆解一个适合创业团队的三位一体方案核心组件选型免费开源。行业案例某AI创业公司的可观测性建设一家做AI推理服务的创业公司3人技术团队。产品上线后用户偶尔投诉响应慢但团队找不到原因。前期做法没有可观测性靠用户投诉发现问题。每次排查靠翻应用日志耗时约30-60分钟。问题日志只有应用层错误没有链路信息。不知道慢在哪里是模型推理慢还是数据库查询慢还是网络慢。后来搭建了基础可观测性Metrics记录每次推理的耗时、Token消耗、成功率。Traces每次请求分配trace_id串联API网关、推理服务、数据库。Logs结构化日志关联trace_id。上线后第一周就发现了问题数据库查询在某些时段耗时过高原因是共享数据库实例被其他服务占用。模型推理耗时正常但API网关到推理服务的网络延迟不稳定。修复后P99延迟从2.1秒降到0.8秒。用户投诉减少约70%。数据说明没有可观测性优化靠猜。有了可观测性优化靠数据。这个差距在用户量少时不明显用户量多后就是口碑差距。二、原理三位一体模型三种信号的职责Metrics回答系统有没有问题。聚合数据时序存储。Traces回答请求走了哪些服务。全链路跟踪。Logs回答具体发生了什么。详细上下文。三者通过trace_id关联形成完整视图。三、代码OpenTelemetry集成实现import time import logging from contextlib import contextmanager from typing import Optional, Dict, Any from dataclasses import dataclass # 1. Metrics 模块 dataclass class MetricPoint: name: str value: float labels: Dict[str, str] timestamp: float 0.0 def __post_init__(self): if self.timestamp 0.0: self.timestamp time.time() class MetricsRegistry: Prometheus风格的指标注册表轻量实现 def __init__(self): self._counters: Dict[str, float] {} self._histograms: Dict[str, list] {} self._gauges: Dict[str, float] {} def counter_inc(self, name: str, value: float 1, labels: Dict None): 计数器增加 key self._label_key(name, labels) self._counters[key] self._counters.get(key, 0) value def gauge_set(self, name: str, value: float, labels: Dict None): 仪表盘设置 key self._label_key(name, labels) self._gauges[key] value def histogram_observe(self, name: str, value: float, labels: Dict None): 直方图记录 key self._label_key(name, labels) if key not in self._histograms: self._histograms[key] [] self._histograms[key].append(value) def _label_key(self, name: str, labels: Optional[Dict]) - str: if not labels: return name label_str ,.join( f{k}{v} for k, v in sorted(labels.items()) ) return f{name}{{{label_str}}} def export_prometheus(self) - str: 导出Prometheus格式可直接暴露/metrics端点 lines [] for key, val in self._counters.items(): lines.append(f# TYPE {key.split({)[0]} counter) lines.append(f{key} {val}) for key, val in self._gauges.items(): lines.append(f# TYPE {key.split({)[0]} gauge) lines.append(f{key} {val}) for key, vals in self._histograms.items(): name key.split({)[0] lines.append(f# TYPE {name} histogram) if vals: lines.append( f{name}_avg{{}} {sum(vals)/len(vals):.2f} ) lines.append(f{name}_count{{}} {len(vals)}) return \n.join(lines) # 2. Traces 模块 import uuid class TraceContext: 链路追踪上下文 def __init__(self, trace_id: str None, span_id: str None): self.trace_id trace_id or str(uuid.uuid4())[:16] self.span_id span_id or str(uuid.uuid4())[:8] self.parent_span_id: Optional[str] None class Span: 追踪跨度 def __init__(self, name: str, context: TraceContext, attributes: Dict None): self.name name self.context context self.attributes attributes or {} self.start_time time.time() self.end_time: Optional[float] None self.events: list [] self.children: list [] def add_event(self, name: str, attributes: Dict None): 添加事件 self.events.append({ name: name, timestamp: time.time(), attributes: attributes or {}, }) def finish(self): 结束Span self.end_time time.time() property def duration_ms(self) - float: if self.end_time and self.start_time: return (self.end_time - self.start_time) * 1000 return 0.0 class Tracer: 追踪器 def __init__(self, service_name: str default): self.service_name service_name self._active_span: Optional[Span] None def start_span(self, name: str, attributes: Dict None) - Span: 开始一个新的Span if self._active_span: ctx TraceContext( trace_idself._active_span.context.trace_id ) ctx.parent_span_id self._active_span.context.span_id else: ctx TraceContext() span Span(name, ctx, attributes) if self._active_span: self._active_span.children.append(span) self._active_span span return span def end_span(self, span: Span): 结束一个Span span.finish() if span.context.parent_span_id: # 回到父Span for child in self._active_span.children: if child is span: self._active_span span return contextmanager def span(self, name: str, **attributes): 上下文管理器方式使用Span span self.start_span(name, attributes) try: yield span finally: self.end_span(span) def to_jaeger_format(self) - Dict: 导出Jaeger兼容格式 # 简化版导出 spans [] def collect(s: Span): spans.append({ traceID: s.context.trace_id, spanID: s.context.span_id, operationName: s.name, startTime: int(s.start_time * 1_000_000), duration: int(s.duration_ms * 1000), tags: [ {key: k, value: str(v)} for k, v in s.attributes.items() ], }) for child in s.children: collect(child) if self._active_span: collect(self._active_span) return { data: [{ traceID: self._active_span.context.trace_id if self._active_span else , spans: spans, processes: { p1: {serviceName: self.service_name} } }] } # 3. Logs 模块 class StructuredLogger: 结构化日志关联trace_id def __init__(self, name: str, tracer: Optional[Tracer] None): self.name name self.tracer tracer self._logger logging.getLogger(name) self._logger.setLevel(logging.DEBUG) def _build_context(self) - Dict: 构建日志上下文 ctx {service: self.name} if self.tracer and self.tracer._active_span: span self.tracer._active_span ctx[trace_id] span.context.trace_id ctx[span_id] span.context.span_id return ctx def info(self, message: str, **kwargs): Info 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.info(message, extractx) def error(self, message: str, exc_infoFalse, **kwargs): Error 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.error(message, extractx, exc_infoexc_info) def warn(self, message: str, **kwargs): Warning 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.warning(message, extractx) # 4. 统一门面 class Observability: 可观测性统一入口 def __init__(self, service_name: str my-service): self.metrics MetricsRegistry() self.tracer Tracer(service_name) self.logger StructuredLogger(service_name, self.tracer) self.service_name service_name def instrument(self, name: str, labels: Dict None): 装饰器自动记录指标和链路 labels labels or {} def decorator(func): def wrapper(*args, **kwargs): start time.time() self.metrics.counter_inc( f{name}_total, labelslabels ) self.logger.info(f{name} started) with self.tracer.span(name): try: result func(*args, **kwargs) duration time.time() - start self.metrics.histogram_observe( f{name}_duration_seconds, duration, labels ) self.metrics.counter_inc( f{name}_success_total, labelslabels ) self.logger.info( f{name} completed, duration_msint(duration * 1000) ) return result except Exception as e: self.metrics.counter_inc( f{name}_error_total, labelslabels ) self.logger.error( f{name} failed, exc_infoTrue, errorstr(e) ) raise return wrapper return decorator # 使用示例 obs Observability(llm-service) obs.instrument(inference, labels{model: gemini-2.0}) def do_inference(prompt: str) - str: 模拟LLM推理 with obs.tracer.span(call_llm_api, modelgemini-2.0): time.sleep(0.5) obs.metrics.gauge_set( llm_tokens_used, 150, labels{model: gemini-2.0} ) obs.logger.info(inference complete, tokens150) return 推理结果 # 执行 result do_inference(什么是分布式系统) print(obs.metrics.export_prometheus()[:200])关键设计点零外部依赖自实现Prometheus指标格式可直接暴露。trace_id贯穿Logs自动关联当前Span的trace_id。装饰器集成obs.instrument一行代码接入。分级导出轻量场景用内置导出正式环境接OpenTelemetry SDK。四、权衡投入产出的平衡点自建 vs SaaS。本文方案适合日活万级以下的产品。日活超过10万后建议迁移到Grafana Cloud或Datadog。采样率设定。全量采集Trace成本高。生产环境建议10%-50%采样重要接口保留100%。日志存储成本。结构化日志增长快。设置30天保留策略超出自动归档到对象存储。告警阈值调优。错误率告警设太高漏故障设太低频繁误报。建议从5%开始根据两周数据调优。取舍决策可观测性建设的优先级框架可观测性建设从哪里开始根据团队规模和产品阶段决定团队规模判断1-3人团队先做Metrics成本最低价值最高。Traces和Logs可以暂缓。3-8人团队Metrics Traces。能定位大部分问题。8人以上团队Metrics Traces Logs三位一体。产品阶段判断产品验证期用户少出问题影响小。做基础Metrics即可。增长期用户变多稳定性开始重要。加Traces能定位跨服务问题。规模化期用户多稳定性要求高。加Logs能定位具体错误原因。告警策略选择初期只告警P0问题服务不可用、错误率10%。避免告警疲劳。中期加P1告警延迟过高、成功率下降。开始主动发现问题。成熟期完整的告警体系覆盖所有关键指标。成本考量自建方案本文方案成本低适合日活万级以下。SaaS方案Datadog、Grafana Cloud成本高但功能全、维护省心。适合日活10万级以上。决策输出根据团队规模和产品阶段选择可观测性建设的深度和优先级。不要一开始就做最完整的方案按需逐步建设。五、总结可观测性三位一体Metrics看整体、Traces看链路、Logs看细节。本文给出了一个零依赖的Python实现可快速集成到现有服务。落地建议先在1-2个核心服务接入验证指标采集和告警。稳定后推广到所有服务。先告警后Dashboard先知道出问题再建大盘看趋势。
创业团队的可观测性体系：Metrics、Traces与Logs的三位一体设计
创业团队的可观测性体系Metrics、Traces与Logs的三位一体设计一、引言可观测性是技术团队的眼睛。没有可观测性线上故障靠用户投诉才知道排查问题靠猜。创业团队尤其需要低成本高价值的可观测性方案。可观测性不等于监控。监控告诉你出问题了可观测性告诉你为什么出问题。本文拆解一个适合创业团队的三位一体方案核心组件选型免费开源。行业案例某AI创业公司的可观测性建设一家做AI推理服务的创业公司3人技术团队。产品上线后用户偶尔投诉响应慢但团队找不到原因。前期做法没有可观测性靠用户投诉发现问题。每次排查靠翻应用日志耗时约30-60分钟。问题日志只有应用层错误没有链路信息。不知道慢在哪里是模型推理慢还是数据库查询慢还是网络慢。后来搭建了基础可观测性Metrics记录每次推理的耗时、Token消耗、成功率。Traces每次请求分配trace_id串联API网关、推理服务、数据库。Logs结构化日志关联trace_id。上线后第一周就发现了问题数据库查询在某些时段耗时过高原因是共享数据库实例被其他服务占用。模型推理耗时正常但API网关到推理服务的网络延迟不稳定。修复后P99延迟从2.1秒降到0.8秒。用户投诉减少约70%。数据说明没有可观测性优化靠猜。有了可观测性优化靠数据。这个差距在用户量少时不明显用户量多后就是口碑差距。二、原理三位一体模型三种信号的职责Metrics回答系统有没有问题。聚合数据时序存储。Traces回答请求走了哪些服务。全链路跟踪。Logs回答具体发生了什么。详细上下文。三者通过trace_id关联形成完整视图。三、代码OpenTelemetry集成实现import time import logging from contextlib import contextmanager from typing import Optional, Dict, Any from dataclasses import dataclass # 1. Metrics 模块 dataclass class MetricPoint: name: str value: float labels: Dict[str, str] timestamp: float 0.0 def __post_init__(self): if self.timestamp 0.0: self.timestamp time.time() class MetricsRegistry: Prometheus风格的指标注册表轻量实现 def __init__(self): self._counters: Dict[str, float] {} self._histograms: Dict[str, list] {} self._gauges: Dict[str, float] {} def counter_inc(self, name: str, value: float 1, labels: Dict None): 计数器增加 key self._label_key(name, labels) self._counters[key] self._counters.get(key, 0) value def gauge_set(self, name: str, value: float, labels: Dict None): 仪表盘设置 key self._label_key(name, labels) self._gauges[key] value def histogram_observe(self, name: str, value: float, labels: Dict None): 直方图记录 key self._label_key(name, labels) if key not in self._histograms: self._histograms[key] [] self._histograms[key].append(value) def _label_key(self, name: str, labels: Optional[Dict]) - str: if not labels: return name label_str ,.join( f{k}{v} for k, v in sorted(labels.items()) ) return f{name}{{{label_str}}} def export_prometheus(self) - str: 导出Prometheus格式可直接暴露/metrics端点 lines [] for key, val in self._counters.items(): lines.append(f# TYPE {key.split({)[0]} counter) lines.append(f{key} {val}) for key, val in self._gauges.items(): lines.append(f# TYPE {key.split({)[0]} gauge) lines.append(f{key} {val}) for key, vals in self._histograms.items(): name key.split({)[0] lines.append(f# TYPE {name} histogram) if vals: lines.append( f{name}_avg{{}} {sum(vals)/len(vals):.2f} ) lines.append(f{name}_count{{}} {len(vals)}) return \n.join(lines) # 2. Traces 模块 import uuid class TraceContext: 链路追踪上下文 def __init__(self, trace_id: str None, span_id: str None): self.trace_id trace_id or str(uuid.uuid4())[:16] self.span_id span_id or str(uuid.uuid4())[:8] self.parent_span_id: Optional[str] None class Span: 追踪跨度 def __init__(self, name: str, context: TraceContext, attributes: Dict None): self.name name self.context context self.attributes attributes or {} self.start_time time.time() self.end_time: Optional[float] None self.events: list [] self.children: list [] def add_event(self, name: str, attributes: Dict None): 添加事件 self.events.append({ name: name, timestamp: time.time(), attributes: attributes or {}, }) def finish(self): 结束Span self.end_time time.time() property def duration_ms(self) - float: if self.end_time and self.start_time: return (self.end_time - self.start_time) * 1000 return 0.0 class Tracer: 追踪器 def __init__(self, service_name: str default): self.service_name service_name self._active_span: Optional[Span] None def start_span(self, name: str, attributes: Dict None) - Span: 开始一个新的Span if self._active_span: ctx TraceContext( trace_idself._active_span.context.trace_id ) ctx.parent_span_id self._active_span.context.span_id else: ctx TraceContext() span Span(name, ctx, attributes) if self._active_span: self._active_span.children.append(span) self._active_span span return span def end_span(self, span: Span): 结束一个Span span.finish() if span.context.parent_span_id: # 回到父Span for child in self._active_span.children: if child is span: self._active_span span return contextmanager def span(self, name: str, **attributes): 上下文管理器方式使用Span span self.start_span(name, attributes) try: yield span finally: self.end_span(span) def to_jaeger_format(self) - Dict: 导出Jaeger兼容格式 # 简化版导出 spans [] def collect(s: Span): spans.append({ traceID: s.context.trace_id, spanID: s.context.span_id, operationName: s.name, startTime: int(s.start_time * 1_000_000), duration: int(s.duration_ms * 1000), tags: [ {key: k, value: str(v)} for k, v in s.attributes.items() ], }) for child in s.children: collect(child) if self._active_span: collect(self._active_span) return { data: [{ traceID: self._active_span.context.trace_id if self._active_span else , spans: spans, processes: { p1: {serviceName: self.service_name} } }] } # 3. Logs 模块 class StructuredLogger: 结构化日志关联trace_id def __init__(self, name: str, tracer: Optional[Tracer] None): self.name name self.tracer tracer self._logger logging.getLogger(name) self._logger.setLevel(logging.DEBUG) def _build_context(self) - Dict: 构建日志上下文 ctx {service: self.name} if self.tracer and self.tracer._active_span: span self.tracer._active_span ctx[trace_id] span.context.trace_id ctx[span_id] span.context.span_id return ctx def info(self, message: str, **kwargs): Info 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.info(message, extractx) def error(self, message: str, exc_infoFalse, **kwargs): Error 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.error(message, extractx, exc_infoexc_info) def warn(self, message: str, **kwargs): Warning 级别日志 ctx self._build_context() ctx.update(kwargs) self._logger.warning(message, extractx) # 4. 统一门面 class Observability: 可观测性统一入口 def __init__(self, service_name: str my-service): self.metrics MetricsRegistry() self.tracer Tracer(service_name) self.logger StructuredLogger(service_name, self.tracer) self.service_name service_name def instrument(self, name: str, labels: Dict None): 装饰器自动记录指标和链路 labels labels or {} def decorator(func): def wrapper(*args, **kwargs): start time.time() self.metrics.counter_inc( f{name}_total, labelslabels ) self.logger.info(f{name} started) with self.tracer.span(name): try: result func(*args, **kwargs) duration time.time() - start self.metrics.histogram_observe( f{name}_duration_seconds, duration, labels ) self.metrics.counter_inc( f{name}_success_total, labelslabels ) self.logger.info( f{name} completed, duration_msint(duration * 1000) ) return result except Exception as e: self.metrics.counter_inc( f{name}_error_total, labelslabels ) self.logger.error( f{name} failed, exc_infoTrue, errorstr(e) ) raise return wrapper return decorator # 使用示例 obs Observability(llm-service) obs.instrument(inference, labels{model: gemini-2.0}) def do_inference(prompt: str) - str: 模拟LLM推理 with obs.tracer.span(call_llm_api, modelgemini-2.0): time.sleep(0.5) obs.metrics.gauge_set( llm_tokens_used, 150, labels{model: gemini-2.0} ) obs.logger.info(inference complete, tokens150) return 推理结果 # 执行 result do_inference(什么是分布式系统) print(obs.metrics.export_prometheus()[:200])关键设计点零外部依赖自实现Prometheus指标格式可直接暴露。trace_id贯穿Logs自动关联当前Span的trace_id。装饰器集成obs.instrument一行代码接入。分级导出轻量场景用内置导出正式环境接OpenTelemetry SDK。四、权衡投入产出的平衡点自建 vs SaaS。本文方案适合日活万级以下的产品。日活超过10万后建议迁移到Grafana Cloud或Datadog。采样率设定。全量采集Trace成本高。生产环境建议10%-50%采样重要接口保留100%。日志存储成本。结构化日志增长快。设置30天保留策略超出自动归档到对象存储。告警阈值调优。错误率告警设太高漏故障设太低频繁误报。建议从5%开始根据两周数据调优。取舍决策可观测性建设的优先级框架可观测性建设从哪里开始根据团队规模和产品阶段决定团队规模判断1-3人团队先做Metrics成本最低价值最高。Traces和Logs可以暂缓。3-8人团队Metrics Traces。能定位大部分问题。8人以上团队Metrics Traces Logs三位一体。产品阶段判断产品验证期用户少出问题影响小。做基础Metrics即可。增长期用户变多稳定性开始重要。加Traces能定位跨服务问题。规模化期用户多稳定性要求高。加Logs能定位具体错误原因。告警策略选择初期只告警P0问题服务不可用、错误率10%。避免告警疲劳。中期加P1告警延迟过高、成功率下降。开始主动发现问题。成熟期完整的告警体系覆盖所有关键指标。成本考量自建方案本文方案成本低适合日活万级以下。SaaS方案Datadog、Grafana Cloud成本高但功能全、维护省心。适合日活10万级以上。决策输出根据团队规模和产品阶段选择可观测性建设的深度和优先级。不要一开始就做最完整的方案按需逐步建设。五、总结可观测性三位一体Metrics看整体、Traces看链路、Logs看细节。本文给出了一个零依赖的Python实现可快速集成到现有服务。落地建议先在1-2个核心服务接入验证指标采集和告警。稳定后推广到所有服务。先告警后Dashboard先知道出问题再建大盘看趋势。
