日志结构化与索引优化：从 grep 到 Elasticsearch 的演进之路

一、"grep 大法"的黄昏时代

实习第一个月，排查线上问题的标准操作是grep ERROR app.log | tail -100。当每天的日志量从 500MB 增长到 50GB 时，grep 从 2 秒变成了 200 秒。更要命的是，grep 只能做简单的字符串匹配：它无法回答"昨天下午 3 点到 4 点之间，Judge Worker 超时的请求中有多少是 Docker 沙箱资源耗尽导致的？"

非结构化日志有四个致命缺陷：无法按字段查询（必须全量扫描）、无法聚合统计（不能 COUNT/AVG/SUM）、不知道日志的"主语"是谁（缺少关联 key）、以及乱序写入导致时序混乱。

日志结构化的核心目标是把日志从"一段文字"变成"一组字段"，让日志变成可检索、可聚合、可关联的时间序列事件数据。

flowchart LR A[应用产生非结构化日志] --> B[日志采集 Agent] B --> C[日志解析 & 结构化] C --> D[写入 Elasticsearch] D --> E[索引构建] E --> F[Kibana 可视化查询] subgraph 解析流水线 C --> C1[正则提取字段] C --> C2[JSON 解析] C --> C3[关联 TraceID] end subgraph 查询能力 F --> F1[全文搜索: 'ERROR timeout'] F --> F2[字段过滤: service:judge AND duration>1000] F --> F3[时序聚合: 每小时 P99 延迟趋势] end style A fill:#ffcccc style F fill:#ccffcc

二、结构化日志的设计原则

2.1 一条好的日志应该包含什么

每条日志至少应包含以下 8 个字段：

2.2 JSON Lines 格式：结构化日志的标准载体

每行一条 JSON 的格式（JSON Lines）是目前最流行也最实用的结构化日志格式：

{"timestamp":"2025-07-15T03:12:45.123Z","level":"INFO","service":"judge-scheduler","trace_id":"abc123","message":"评测开始","context":{"problem_id":42,"language":"python"}} {"timestamp":"2025-07-15T03:12:45.456Z","level":"ERROR","service":"judge-scheduler","trace_id":"abc123","message":"编译超时","error":{"type":"TimeoutError","timeout_ms":5000}}

这种格式既可以直接grep，又可以批量导入 Elasticsearch，是向后兼容结构化升级的最佳选择。

三、结构化日志系统与索引优化的完整实现

以下代码展示了结构化日志生成器和Elasticsearch 索引模板与查询的完整方案。生成器确保每条日志都带 TraceID 和结构化上下文；索引模板优化了映射和分片策略。

""" 结构化日志系统 功能：JSON Lines 日志生成 + 索引模板 + ES 查询辅助 """ import json import time import uuid import os import logging from datetime import datetime, timezone from typing import Dict, Any, Optional from dataclasses import dataclass, field, asdict from logging.handlers import RotatingFileHandler # ==================== 结构化日志模型 ==================== @dataclass class LogEntry: """一条结构化日志的核心字段""" timestamp: str # ISO 8601 时间戳 level: str # 日志级别：DEBUG/INFO/WARN/ERROR service: str # 服务名称 trace_id: str # 关联的 TraceID span_id: str = "" # 关联的 SpanID message: str = "" # 人类可读描述 context: Dict[str, Any] = field(default_factory=dict) # 业务上下文 error: Optional[Dict[str, Any]] = None # 错误详情 duration_ms: Optional[float] = None # 操作耗时 host: str = "" # 主机名 def to_json(self) -> str: """序列化为 JSON Lines 格式""" entry_dict = asdict(self) # 移除 None 值的字段，减少日志体积 return json.dumps( {k: v for k, v in entry_dict.items() if v is not None}, ensure_ascii=False, ) class StructuredLogger: """ 结构化日志生成器 每次日志输出都包含 TraceID 和 structured context """ def __init__(self, service_name: str = "unknown", log_dir: str = "/var/log/app", max_bytes: int = 100 * 1024 * 1024, # 100MB backup_count: int = 10): self.service_name = service_name self.host = os.uname().nodename self._ensure_log_dir(log_dir) # 设置日志处理器：写入 JSON Lines 文件 self.log_path = os.path.join(log_dir, "app.jsonl") handler = RotatingFileHandler( self.log_path, maxBytes=max_bytes, backupCount=backup_count, ) formatter = logging.Formatter('%(message)s') handler.setFormatter(formatter) self.python_logger = logging.getLogger(service_name) self.python_logger.setLevel(logging.DEBUG) self.python_logger.addHandler(handler) def _ensure_log_dir(self, log_dir: str): """确保日志目录存在""" os.makedirs(log_dir, exist_ok=True) def _create_entry(self, level: str, message: str, trace_id: str = "", context: Optional[Dict] = None, error: Optional[Exception] = None, duration_ms: Optional[float] = None, span_id: str = "") -> LogEntry: """创建一条结构化日志条目""" entry = LogEntry( timestamp=datetime.now(timezone.utc).isoformat(), level=level, service=self.service_name, trace_id=trace_id or uuid.uuid4().hex[:16], span_id=span_id, message=message, context=context or {}, host=self.host, duration_ms=duration_ms, ) if error: entry.error = { "type": type(error).__name__, "message": str(error), } return entry def info(self, message: str, **kwargs): """记录 INFO 级别日志""" entry = self._create_entry("INFO", message, **kwargs) self.python_logger.info(entry.to_json()) def warn(self, message: str, **kwargs): """记录 WARN 级别日志""" entry = self._create_entry("WARN", message, **kwargs) self.python_logger.warning(entry.to_json()) def error(self, message: str, error: Optional[Exception] = None, **kwargs): """记录 ERROR 级别日志""" entry = self._create_entry( "ERROR", message, error=error, **kwargs ) self.python_logger.error(entry.to_json()) def debug(self, message: str, **kwargs): """记录 DEBUG 级别日志""" entry = self._create_entry("DEBUG", message, **kwargs) self.python_logger.debug(entry.to_json()) # ==================== Elasticsearch 索引模板 ==================== ELASTICSEARCH_INDEX_TEMPLATE = { "index_patterns": ["app-logs-*"], "template": { "settings": { # 分片策略：按天建索引，每天一个主分片 "number_of_shards": 1, "number_of_replicas": 1, # 刷新间隔：调大以减少写入压力 "refresh_interval": "5s", # 索引生命周期：7天后自动删除 "index.lifecycle.name": "logs-retention-7d", "index.lifecycle.rollover_alias": "app-logs", # 写入优化 "translog.durability": "async", "translog.sync_interval": "5s", # 排序字段优化 "sort.field": "timestamp", "sort.order": "desc", }, "mappings": { "dynamic": "strict", "properties": { # 时间戳字段 "timestamp": { "type": "date", "format": "strict_date_optional_time", }, # 日志级别（用于聚合统计） "level": { "type": "keyword", }, # 服务名（keyword 类型支持精确过滤） "service": { "type": "keyword", }, # TraceID（keyword 支持精确匹配） "trace_id": { "type": "keyword", }, # 消息（text 类型支持全文搜索） "message": { "type": "text", "fields": { # 同时索引 keyword 版本，支持排序 "keyword": {"type": "keyword", "ignore_above": 256} }, }, # 业务上下文（flattened 类型：不限制字段数量） "context": { "type": "flattened", }, # 错误信息 "error": { "type": "object", "properties": { "type": {"type": "keyword"}, "message": {"type": "text"}, }, }, # 耗时 "duration_ms": { "type": "float", }, # 主机名 "host": { "type": "keyword", }, # SpanID "span_id": { "type": "keyword", "index": False, # 不需要单独索引 }, }, }, }, } # 常用 ES 查询模板 ES_QUERIES = { "errors_last_hour": { "query": { "bool": { "must": [ {"term": {"level": "ERROR"}}, {"range": {"timestamp": {"gte": "now-1h"}}}, ], } }, "sort": [{"timestamp": "desc"}], "size": 100, }, "trace_by_id": { "query": { "bool": { "must": [ {"term": {"trace_id": "{{TRACE_ID}}"}}, ], } }, "sort": [{"timestamp": "asc"}], "size": 1000, }, "p99_by_service": { "size": 0, "query": { "range": {"timestamp": {"gte": "now-1h"}}, }, "aggs": { "by_service": { "terms": {"field": "service", "size": 10}, "aggs": { "p99_latency": { "percentiles": { "field": "duration_ms", "percents": [99], }, }, "error_rate": { "filter": {"term": {"level": "ERROR"}}, }, }, }, }, }, } # ==================== 使用示例 ==================== def demo_structured_logging(): """演示结构化日志的使用""" import time as _time # 创建多个服务的日志器 api_logger = StructuredLogger(service_name="api-gateway") judge_logger = StructuredLogger(service_name="judge-scheduler") # 模拟一个请求链路 trace_id = uuid.uuid4().hex # API Gateway 日志 api_logger.info( "接收提交请求", trace_id=trace_id, context={"problem_id": 42, "user_id": "user_123"}, ) # Judge Scheduler 日志 start = _time.time() judge_logger.info( "开始调度评测", trace_id=trace_id, context={"problem_id": 42, "language": "python"}, ) # 模拟执行 _time.sleep(0.5) duration = (_time.time() - start) * 1000 judge_logger.info( "评测完成", trace_id=trace_id, duration_ms=duration, context={ "problem_id": 42, "result": "accepted", "memory_mb": 128, "cpu_time_ms": 45, }, ) # 模拟一个错误 try: raise TimeoutError("Docker 沙箱启动超时") except TimeoutError as e: judge_logger.error( "编译阶段失败", error=e, trace_id=trace_id, context={"problem_id": 42, "phase": "compile"}, ) print(f"日志已写入: {api_logger.log_path}") print(f"日志已写入: {judge_logger.log_path}") if __name__ == "__main__": demo_structured_logging()

四、日志索引优化的实战要诀

按天建索引（Time-based Index）：不要把所有日志写入一个索引，而是按app-logs-2025-07-15的格式每天建一个。好处：旧数据可以按天删除（直接 DROP 索引，O(1) 操作）、查询可以限定日期范围（减少扫描的分片数）。

字段映射的谨慎设计：避免dynamic: true（ES 默认），因为这会为每个新字段自动创建映射。在高 QPS 场景下可能导致映射爆炸——成千上万个不同命名的context.xxx字段。使用flattened类型处理动态上下文字段。

控制分词策略：text类型会进行分词（消耗 CPU 和存储），keyword类型不会。TraceID、服务名、日志级别都应该用keyword；只有message需要text做全文搜索。

采样写入高基数日志：对于访问日志（每个请求一条），不必全量写入。设置 10-20% 的采样率能大幅降低存储和索引压力，同时保留足够的分析样本。

五、总结

从grep到Elasticsearch >= 7.x，日志治理的演进路径清晰：

1. 结构化是第一步：把日志从"一段文字"变成"一组字段"，JSON Lines 是当前最佳载体。
2. TraceID 是日志的"黏合剂"：没有 TraceID 的日志就是散落的拼图碎片。
3. 索引设计是性能的天花板：按天建索引、keyword 映射、flattened 上下文——这三招能让写入吞吐和查询延迟都保持健康。
4. 只索引需要搜索的字段：日志体积的大头（stack trace、长输出）可以存在对象存储中，只在日志中留一个链接。

结构化日志的投资回报率极高——一次配置，终身受益。下次排查问题时，不再需要记住那个grep -A 10 -B 5的复杂命令，只需要一条 Kibana 查询。

本文从日志结构化的设计原则出发，提供了完整的 JSON Lines 日志生成器和 ES 索引模板。代码中的 LogEntry 数据模型和 ES_QUERIES 查询模板可以直接用于生产环境的日志治理。