Logstash ：从日志采集到数据处理与 Elasticsearch 集成

在后端开发和运维工作中，日志是排查问题、分析系统运行状态和追踪业务行为的重要依据。随着服务数量增加，日志可能分散在不同服务器、容器和应用中。如果每次排查问题都需要登录服务器查看日志，不仅效率低，也很难进行跨服务的关联分析。

ELK Stack 就是为解决这类问题而广泛使用的一套技术组合：

• Elasticsearch：负责存储、索引和检索数据。
• Logstash：负责采集、解析、转换和输出数据。
• Kibana：负责查询、可视化和分析数据。

实际项目中，还经常会使用 Filebeat 负责轻量级日志采集，使用 Kafka 缓冲和分发数据，形成更加完整的日志处理链路。

本文从 Logstash 的基本原理出发，结合配置示例，介绍它的核心组件、常用插件、Kafka 集成、Elasticsearch 输出和生产环境优化。

一、Logstash 是什么

Logstash 是 Elastic 生态中的数据处理引擎。它能够从不同数据源读取数据，对数据进行解析、清洗、转换和丰富，最后将处理结果发送到一个或多个目标系统。

例如，一个 NestJS 应用输出如下日志：

2026-10-10 08:30:15 INFO GET /api/users 200 32ms

对于人来说，这条日志很容易理解，但如果直接存储为普通文本，就不方便按照请求路径、状态码和响应时间进行统计。

通过 Logstash，可以将它解析成结构化数据：

{
  "@timestamp": "2026-10-10T00:30:15.000Z",
  "level": "INFO",
  "method": "GET",
  "path": "/api/users",
  "status": 200,
  "duration_ms": 32
}

结构化之后，就可以在 Elasticsearch 中执行查询，例如统计某个接口的请求量、分析不同状态码的分布，或者筛选响应时间超过 500ms 的请求。

Logstash 的核心价值并不只是转发数据，而是把来源各异、格式不统一的数据转换为适合后续存储和分析的形式。

二、Logstash 的核心工作流程

Logstash 的处理流程由三个核心阶段组成：

数据源
  │
  ▼
Input 输入
  │
  ▼
Queue 队列
  │
  ▼
Filter 过滤与转换
  │
  ▼
Output 输出
  │
  ▼
Elasticsearch / Kafka / 文件 / 其他系统

这是理解 Logstash 最重要的基础。每一个阶段都由对应的插件实现，开发者通过配置文件将这些插件组合起来，形成一条 Pipeline（数据处理管道）。官方文档也将 Input、Filter、Output 作为 Logstash Pipeline 的核心组成部分。

1. Input：数据从哪里来

Input 负责接入数据。常见插件包括：

不同 Input 的适用场景并不相同。例如，应用日志通常可以由 Filebeat 采集后发送给 Logstash；已经进入 Kafka 的业务事件则可以由 Logstash 的 Kafka Input 消费。

2. Filter：数据如何处理

Filter 是 Logstash 的核心处理环节，主要用于解析文本、调整字段、转换数据类型和补充信息。

常见插件包括：

例如，日志中的响应时间最初是字符串，经过处理后可以转换为整数，便于在 Elasticsearch 中进行数值比较和统计。

需要注意，Filter 并不是必须的。如果输入数据已经是符合要求的 JSON，且不需要额外转换，可以直接输出到目标系统。

3. Output：数据最终去哪里

Output 负责将处理后的事件发送到目标位置。

一条 Pipeline 可以配置多个 Output。例如，处理后的日志既写入 Elasticsearch 用于检索，也可以发送到 Kafka 供其他消费者使用。

不过，多 Output 并不意味着各目标之间天然具备事务一致性。如果其中一个输出失败，需要结合重试策略、持久化队列和业务要求设计可靠的数据处理机制。

三、编写第一个 Logstash 配置

Logstash 通常通过 .conf 配置文件定义 Pipeline。一个最简单的配置包含 Input 和 Output 两部分。

创建 logstash.conf：

input {
  stdin {}
}

filter {
  mutate {
    add_field => {
      "service" => "demo-service"
    }
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

启动 Logstash：

bin/logstash -f logstash.conf

Windows 环境下可以使用对应的 bin\logstash.bat 启动脚本。

启动后，在控制台输入一行文本，Logstash 就会接收该输入，经过 Filter 增加 service 字段，再通过 rubydebug 格式输出事件。

这个例子虽然简单，却包含了完整的处理逻辑：

1. Input 从标准输入读取数据。
2. Filter 为事件增加字段。
3. Output 将最终事件打印到控制台。

rubydebug 是调试时很有用的 Codec，它可以把事件中的字段以便于阅读的形式展示出来。生产环境中则通常将数据输出到 Elasticsearch、Kafka 或其他存储系统。

四、常用 Filter 插件详解

在实际项目中，日志解析和字段处理往往占据了 Logstash 配置的大部分内容。

1. 使用 JSON 插件解析日志

假设应用输出的日志本身就是 JSON 字符串：

{"level":"INFO","method":"GET","path":"/api/users","status":200,"duration":32}

可以使用 JSON Filter 解析：

input {
  stdin {
    codec => plain
  }
}

filter {
  json {
    source => "message"
    target => "log"
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

这里的 source => "message" 表示从事件的 message 字段读取 JSON 字符串，target => "log" 表示将解析后的字段放到 log 对象下。

解析后，字段路径类似于：

log.level
log.method
log.path
log.status
log.duration

如果不设置 target，解析出的字段通常会直接合并到事件顶层。

在生产环境中，建议根据统一的日志规范决定是否使用 target，避免不同来源的字段发生冲突。还应注意 JSON 解析失败时的处理方式，可以利用 _jsonparsefailure 等标签识别异常事件。

2. 使用 Grok 解析普通文本

并非所有应用都会输出 JSON。有些日志格式更接近传统文本，例如：

2026-10-10 08:30:15 INFO GET /api/users 200 32ms

此时可以使用 Grok：

input {
  stdin {}
}

filter {
  grok {
    match => {
      "message" => "%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:level} %{WORD:method} %{URIPATHPARAM:path} %{NUMBER:status:int} %{NUMBER:duration:int}ms"
    }
  }

  date {
    match => ["log_time", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  }

  mutate {
    remove_field => ["log_time"]
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

Grok 模式由多个部分组成，例如：

• %{TIMESTAMP_ISO8601:log_time}：提取时间并保存到 log_time。
• %{LOGLEVEL:level}：提取日志级别。
• %{WORD:method}：提取 HTTP 方法。
• %{NUMBER:status:int}：提取状态码并尝试转换为整数。
• %{NUMBER:duration:int}：提取响应时间并转换为整数。

随后，date 插件将业务日志时间写入 @timestamp。这里指定了上海时区，是因为示例日志使用的是北京时间；如果原始日志已经带有明确的时区信息，就应根据实际格式调整解析规则。

这里有一个容易忽略的细节：Logstash 事件默认会带有采集时生成的 @timestamp。如果不解析原始日志时间，那么查询结果反映的可能是采集时间，而不是请求实际发生的时间。对于延迟到达、批量补传的日志，这两者可能相差很大。

3. 使用 Mutate 转换字段

Mutate 用于执行常见的数据整理操作：

filter {
  mutate {
    rename => {
      "duration" => "duration_ms"
    }

    convert => {
      "status" => "integer"
      "duration_ms" => "integer"
    }

    add_field => {
      "environment" => "production"
    }

    remove_field => [
      "host",
      "agent"
    ]
  }
}

上面的配置会重命名字段、转换数据类型、增加环境标识，并删除不需要的字段。

但删除 host、agent 之类的字段之前，需要先确认后续是否还要使用它们。这些字段可能包含主机标识、采集端信息等排障所需的数据。

字段类型最好在写入 Elasticsearch 前就确定。否则，同一个索引中的某个字段可能因为首次写入的数据类型不符合预期，导致后续文档出现映射冲突。

五、Logstash 与 Elasticsearch 集成

Logstash 经常被用作 Elasticsearch 的数据接入层。它能够把原始日志处理为结构化事件，再通过 Elasticsearch Output 写入索引。

例如，下面的配置可以将解析后的日志写入 Elasticsearch：

input {
  beats {
    port => 5044
  }
}

filter {
  if [message] {
    mutate {
      add_field => {
        "[@metadata][target_index]" => "app-logs"
      }
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "%{[@metadata][target_index]}-%{+yyyy.MM.dd}"
  }
}

这个例子使用 Beats Input 接收数据，并将其写入按日期区分的索引，例如：

app-logs-2026.10.10
app-logs-2026.10.11
app-logs-2026.10.12

[@metadata] 是 Logstash 事件的元数据区域，适合保存路由、索引名称等内部处理信息。元数据字段不会像普通事件字段那样直接写入最终文档。

上面的配置只是展示索引路由方式。实际部署时，需要根据 Elasticsearch 版本、索引生命周期管理策略和数据量确定索引设计。对于生产集群，不建议只因为日期变化就无限制创建大量小索引。

如果 Elasticsearch 开启了身份认证，应通过受支持的认证配置和安全的凭据管理方式连接，不要把真实密码直接提交到代码仓库。

Logstash 与 Elasticsearch Ingest Pipeline 的区别

Elasticsearch 本身也支持 Ingest Pipeline，可以在数据进入索引前执行字段转换和处理。

两者的定位有所不同：

• Logstash：适合承担独立的数据接入、复杂转换、多数据源汇聚、Kafka 消费和多目标输出。
• Ingest Pipeline：适合在 Elasticsearch 内部完成相对直接的入库前处理。
• 两者配合：Logstash 负责采集与复杂的数据整理，Elasticsearch Ingest Pipeline 负责最后阶段的轻量处理。

并不是所有日志都需要经过 Logstash。如果数据本身已经结构化，而且只需要少量处理，直接通过 Beats 或其他客户端写入 Elasticsearch，可能更加简单。

六、Logstash 与 Kafka 的集成

在数据量较大、生产者和消费者需要解耦的系统中，Kafka 经常被放在日志处理链路中。

例如：

NestJS 服务
    │
    ▼
Filebeat / 日志生产者
    │
    ▼
Kafka Topic
    │
    ▼
Logstash
    │
    ▼
Elasticsearch
    │
    ▼
Kibana

Kafka 在这里承担消息缓冲和分发的职责。即使 Elasticsearch 暂时无法接收数据，Kafka 仍可以在保留策略允许的范围内保存消息，等待消费者恢复。

Logstash 的 Kafka Input 示例：

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["app-logs"]
    group_id => "logstash-app-logs"

    codec => json

    auto_offset_reset => "earliest"
  }
}

filter {
  mutate {
    add_field => {
      "pipeline_source" => "kafka"
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "app-logs-%{+yyyy.MM.dd}"
  }
}

这段配置表达了几个关键概念：

• bootstrap_servers：Kafka 集群的连接地址。
• topics：需要消费的 Topic。
• group_id：消费者组标识，同一消费者组内的多个消费者可以分担分区。
• codec => json：将 Kafka 消息按照 JSON 格式解析。
• auto_offset_reset：在没有有效提交位点时，指定从何处开始消费。

auto_offset_reset => "earliest" 并不意味着每次重启都会从头消费，它主要作用于没有可用有效偏移量的情况。

生产环境还需要根据 Kafka 插件版本和配置，明确消息确认、偏移量提交、重试及失败处理行为。不能仅凭一条 Kafka Input 配置，就认定整个链路已经具备端到端的可靠交付能力。

为什么要使用 Kafka，而不是让 Logstash 直接接收所有日志？

当日志量不大、链路简单时，Filebeat 直接发送给 Logstash 是合理的选择：

应用日志 → Filebeat → Logstash → Elasticsearch

但如果有大量应用、多个消费目标，或者希望将日志采集与 Elasticsearch 的写入速度解耦，就可以引入 Kafka：

应用日志 → 采集端 → Kafka → Logstash → Elasticsearch

这种架构的优势是缓冲能力、消费者扩展能力和数据分发能力更强，但也会引入 Kafka 集群的部署、容量规划、Topic 管理和监控成本。

需要特别注意：Kafka 的数据保留时间、Logstash 的消费能力、Elasticsearch 的写入能力必须共同规划。如果 Logstash 长时间跟不上生产速度，Kafka 消费延迟仍会持续增长。

七、生产环境中的可靠性与性能优化

Logstash 的运行效果不仅取决于配置是否正确，还与事件大小、Filter 复杂度、Pipeline 并发度、磁盘性能和下游写入速度密切相关。

1. 启用 Persistent Queue

默认情况下，Logstash 的事件队列主要使用内存。如果进程异常终止，尚未处理完成的内存事件可能丢失。

Persistent Queue（持久化队列）可以将队列中的事件保存到磁盘，降低 Logstash 异常退出时的数据丢失风险。

在 logstash.yml 中配置：

queue.type: persisted
queue.max_bytes: 4gb
path.queue: /usr/share/logstash/data/queue

其中，queue.max_bytes 用于限制持久化队列容量。实际容量应根据日志速率、平均事件大小、下游故障时长以及可用磁盘空间确定。

持久化队列并不是 Kafka 的完全替代品。它主要保护 Logstash 内部排队的事件，不能替代跨机器的数据复制，也无法解决磁盘损坏等所有故障。持久化队列写满后，Logstash 会对输入施加反压，暂停或减缓接收数据。

2. 合理配置 Pipeline 并发度

Logstash 通过 Pipeline Worker 并行处理事件批次。可以在 logstash.yml 中调整：

pipeline.workers: 4
pipeline.batch.size: 125
pipeline.batch.delay: 50

主要参数的含义：

• pipeline.workers：处理事件的 Worker 数量。
• pipeline.batch.size：单个批次尝试处理的最大事件数。
• pipeline.batch.delay：等待批次填充时允许的等待时间。

增加 Worker 或 Batch Size 不一定会带来线性提升。如果 Filter 使用大量正则匹配、Ruby 脚本或复杂的地理信息查询，CPU 和堆内存可能先成为瓶颈。批次过大也会增加单次处理的内存占用。

建议先观察吞吐量、事件延迟、CPU、堆内存和下游写入速度，再逐步调整参数，而不是直接套用固定配置。

3. 控制 Grok 的复杂度

Grok 对非结构化日志很有用，但复杂模式可能带来较高的 CPU 消耗。

优化思路包括：

• 优先使用应用直接输出的 JSON，减少重复解析。
• 尽量使用清晰、可预测的匹配模式。
• 为不同日志格式拆分处理逻辑，避免所有日志都经过一组复杂的 Grok。
• 对不需要的日志进行条件过滤，减少后续处理量。
• 使用真实样本验证匹配规则，并覆盖异常格式。

如果日志格式固定，也可以考虑 Dissect 插件。它适合按照明确的分隔结构拆分文本，在某些固定格式场景中比复杂的 Grok 更直接。

4. 减少不必要的字段

日志事件并非字段越多越好。

例如，完整的 HTTP 请求日志可能包含 Cookie、请求头、请求体、追踪信息和用户信息。并不是所有字段都需要进入 Elasticsearch。

减少不必要的字段，可以降低网络传输、序列化、堆内存和索引存储开销。对包含敏感信息的字段，还应在写入目标系统之前进行脱敏或删除。

但要避免为了节省少量存储空间而删除关键的 trace_id、request_id、服务名和环境信息。这些字段往往对跨服务排查问题非常重要。

5. 监控队列和下游写入

生产环境至少应关注以下指标：

如果输入量持续高于输出量，队列就可能不断增长。此时应该进一步判断是 Logstash 处理能力不足，还是 Elasticsearch 写入受限，而不是一味增加 Worker 数量。

八、常见问题排查

1. Logstash 启动失败

优先检查配置语法、插件是否安装、配置项是否与当前版本兼容，以及端口是否被占用。

可以使用配置检查模式：

bin/logstash -f logstash.conf --config.test_and_exit

该模式适合在启动前验证配置，但无法保证运行时的网络连接、认证和目标写入一定成功。

2. Elasticsearch 中没有数据

建议按链路逐层排查：

1. 确认 Input 是否收到数据。
2. 使用 stdout { codec => rubydebug } 检查 Filter 后的事件。
3. 检查 Elasticsearch Output 的错误日志。
4. 确认 Elasticsearch 地址、认证信息和网络连通性。
5. 检查目标索引是否正确，以及是否存在 Mapping 冲突或拒绝写入的情况。

如果使用 Kafka，还要检查 Topic 是否有消息、消费者组是否正常运行，以及消费延迟是否持续增长。

3. Elasticsearch 出现字段类型冲突

例如，某个索引的 status 字段已经被映射为整数，后来却写入了无法转换的字符串。

这类问题通常需要统一上游字段类型、明确 Elasticsearch Mapping，并对异常数据进行隔离或修正。

Logstash 可以帮助转换字段，但它无法自动消除所有 Mapping 冲突。已经建立的字段映射通常不能直接改成任意另一种类型，必要时需要创建新索引并重新写入数据。

4. 数据重复

Logstash、Kafka 和 Elasticsearch 组合起来，并不天然等于端到端的 exactly-once（恰好一次）处理。

例如，事件已经写入 Elasticsearch，但消费者在记录消费进度前发生故障，事件可能再次被处理。对于允许重复的数据，可以接受这种行为；对于需要幂等的业务事件，则应考虑稳定的事件 ID，并合理设计 Elasticsearch 文档 ID 或业务去重机制。

九、Logstash 在整个日志体系中的定位

Logstash 不是日志系统的全部，而是数据处理链路中的一个组件。理解它与周边技术的职责划分，才能合理设计系统。

在简单系统中，可以使用 Filebeat 直接将日志发送给 Elasticsearch；在需要复杂解析时引入 Logstash；在数据量大、消费端需要解耦或需要多个下游时，再考虑 Kafka。

一个好的日志平台，不是简单地把所有组件都部署一遍，而是根据数据量、可靠性要求、查询需求和运维成本，选择合适的架构。

总结来说，Logstash 最值得掌握的内容有三个：Pipeline 的处理模型、Filter 插件的数据转换能力，以及输入、队列和输出之间的可靠性设计。 掌握这些内容之后，无论是处理 NestJS 应用日志、消费 Kafka 消息，还是将数据写入 Elasticsearch，都能更清楚地判断应该在哪个环节处理问题，以及如何优化整个数据链路。

参考资料：

• Logstash 官方文档：How Logstash Works
• Logstash 官方文档：Execution Model
• Logstash 官方文档：Persistent Queues
• Logstash 官方文档：Pipeline-to-pipeline communication

注意：不同 Logstash 版本以及各插件版本的可用参数可能存在差异，正式部署前应以实际使用版本的官方文档为准。