日志采集与查询：配置日志采集与批处理

本篇以日志多行合并与采集为例；前置是源文件身份、编码、采集偏移和索引时间窗。源码与输入与下文配置共用同一份文件，示例未执行。

核对日期2026-10-17；适用范围：Filebeat 8.19 filestream教学配置；SLS按2026-10-17在线中文文档的查询/索引概念核对，不固定云端后端版本。8.19归档文档不是当前安装建议。示例未执行；真实环境仍需验证产品配置、权限、连通与恢复结果。先读起点理解前置，再按章节完成推演。

文件及输入合同

日志批处理文件包括normalize.py、raw.log、orphan.log、requirements.txt与filebeat.yml，全部源码配置在下方共用仓库文件。示例未执行；Python标准库3.11–3.13是读者前置。一次性程序先把所有事件整理完才输出，非法UTF-8、孤立续行或非法级别会退出2，防止半批正常输出造成错误成功判断。它不扫描目录、不连接存储、不持续轮询，也不替代真正采集器。

成功和失败推演

日志成功推演为3个JSON事件，第二条message含两行堆栈，邮箱变为[EMAIL]、token变为[REDACTED]，offset指原始文件字节位置。orphan.log第一行没有时间头，预期stderr报孤立续行且stdout无事件；不能把这个失败样本追加到正常输出并继续统计。读者若更改分隔符、空格、编码或事件头格式，应先重新定义解析合同，再修改规则，不能默认任何日志都适用。

自跑与清理顺序

日志自跑清理须在隔离目录中先读README和依赖声明，再用自己的Python执行normalize.py raw.log。重定向输出时先用新文件名，避免Shell先截断旧证据；确认脱敏和事件数后才共享报告。若读者将来使用Filebeat，先替换路径、创建专用状态目录、审核权限，再按受维护发行包文档配置。清理时先停止专用采集器，保留registry和原文件用于恢复，直接删registry可能重复，删源文件可能永久丢失。

完整源码、配置与依赖

下列文件与README共用同一份仓库内容，保持输入、失败与清理合同一致。示例未执行；以下命令仅供读者在满足前置后自行运行。

requirements.txt

# Python >=3.11,<3.14; standard library only; no pip packages

raw.log

2026-10-17T08:00:00Z INFO request service=checkout request_id=r1 email=alice@example.invalid token=FAKE_TOKEN_ONLY status=200
2026-10-17T08:00:01Z ERROR request service=checkout request_id=r2 email=bob@example.invalid token=FAKE_TOKEN_ONLY status=500
  Traceback: teaching exception
  ValueError: invalid cart
2026-10-17T08:00:02Z INFO request service=checkout request_id=r3 email=carol@example.invalid token=FAKE_TOKEN_ONLY status=200

normalize.py

"""Offline one-shot log normalization. No collector/service/network is started."""
import argparse
import json
import re
import sys
from pathlib import Path

START = re.compile(r"^\d{4}-\d{2}-\d{2}T")
EMAIL = re.compile(r"[^\s=]+@[^\s]+")
TOKEN = re.compile(r"token=[^\s]+")

def normalize(source):
    events = []
    current = None
    offset = 0
    for raw in source.read_bytes().splitlines(keepends=True):
        line = raw.decode("utf-8", errors="strict").rstrip("\r\n")
        clean = TOKEN.sub("token=[REDACTED]", EMAIL.sub("[EMAIL]", line))
        if START.match(line):
            parts = clean.split(" ", 2)
            if len(parts) != 3 or parts[1] not in {"INFO", "WARN", "ERROR"}:
                raise ValueError("invalid event header")
            current = {"offset": offset, "timestamp": parts[0], "level": parts[1], "message": parts[2]}
            events.append(current)
        elif current is None:
            raise ValueError("orphan continuation line")
        else:
            current["message"] += "\n" + clean
        offset += len(raw)
    return events

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("source", type=Path)
    args = parser.parse_args()
    try:
        events = normalize(args.source)
    except (OSError, ValueError, UnicodeError) as error:
        print(str(error), file=sys.stderr)
        return 2
    # Buffer before printing: malformed input must not emit a partial batch.
    for event in events:
        print(json.dumps(event, ensure_ascii=False))
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

filebeat.yml

# Filebeat 8.19 teaching configuration. Linux-only paths; reader must replace them.
filebeat.inputs:
  - type: filestream
    id: m5-next-java-log
    enabled: true
    paths: ['/srv/m5-next-logs/raw.log']
    parsers:
      - multiline:
          type: pattern
          pattern: '^\d{4}-\d{2}-\d{2}T'
          negate: true
          match: after
          max_lines: 100
          timeout: 5s
# Remove complete free-text payload instead of attempting unsafe partial secret masking.
processors:
  - drop_fields:
      fields: [message]
      ignore_missing: true
  - add_fields:
      target: ''
      fields:
        message: '[RAW_MESSAGE_REMOVED]'
        service: checkout
output.file:
  path: /srv/m5-next-logs/sanitized
  filename: events
path.data: /srv/m5-next-logs/state

orphan.log

  orphan stack line

README.md

# 离线日志规范化与采集配置

示例未执行。依赖见 requirements.txt；Python 3.11–3.13标准库，不安装第三方包。normalize.py是一次性读文件程序，没有监听端口，不是采集服务。raw.log中邮箱为保留域，token为明显假值，禁止换入真实秘密。Filebeat 8.19配置只是另一条采集思路，不与normalize.py串成已验证流水线。

读者自跑：进入此目录，用自己的Python解释器执行 `python normalize.py raw.log`，预期退出0并输出3个JSON事件；r2的两行堆栈合并在message中，邮箱变成[EMAIL]、token变成[REDACTED]。offset是输入UTF-8字节偏移，不是事件序号；程序每次从头读，没有持久断点和去重能力。它只识别演示字段，不能作为通用生产脱敏器。输出到独立新文件时先确认权限，避免Shell在失败前截断已有结果。

失败推演：`python normalize.py orphan.log`预期退出2、stderr提示orphan continuation line且stdout无事件；不存在路径或非法UTF-8同样退出2。第一条记录前没有合法时间头即拒绝，不把孤立堆栈归入下一请求。

Filebeat读者前置：受支持Linux实验机、已自行安装与核对的8.19发行包；把paths/path.data/output.file.path三个绝对路径替换到独立实验目录，原日志只读、状态与输出可写。配置移除整个原始message后写固定脱敏标记，保留源文件路径等元数据仍须按组织规则处理。它故意不保留r2内容，避免把Python演示替换误称Filebeat原生脱敏。不要采集自己的输出目录。读者以后可自选Filebeat配置检查及专用实例操作，本轮未执行任何Filebeat命令。

恢复清理：一次性Python无状态，只保留人工审核的脱敏输出；删除新建输出副本即可。Filebeat若以后由读者启用，先停止专用进程，再保存registry用于恢复；直接删除state会重读并可能重复，删除raw.log可能造成数据永久缺失。不得清理生产registry。演示正则不能识别所有个人信息，线上应白名单结构化日志并审核原始落盘权限。

本篇练习与参考解答

练习：Python脱敏器能否代替Filebeat配置保证任何秘密都不会出站？

参考解答

不能。Python只识别演示格式，Filebeat配置则主动删除自由文本message；二者职责不同，其他元数据仍需审核。

核对资料与相邻阅读

• Filebeat事件合并
• Filebeat filestream
• SLS中文入口

继续阅读机制、完整实践、运维、诊断与复习。