ELK Stack集成：同步Logstash、Elasticsearch和Kibana

当Logstash、Elasticsearch和Kibana不同步时，日志会消失、仪表盘显示为空，或者字段类型错误。ELK Stack集成不仅仅是启动三个服务，更重要的是确保索引名称、映射、时间戳、凭据和数据视图保持一致。

本指南将介绍一个实用的日志管道：Logstash接收事件、解析事件、发送到Elasticsearch，Kibana读取生成的索引或数据流。示例使用经典的Logstash管道语法和Elasticsearch API，您可以从Kibana Dev Tools运行。

理解数据流

追踪一个事件在堆栈中的流转：

Logstash 从Beats、TCP、syslog、文件、队列或其他输入接收数据。
Logstash过滤器 解析、丰富、重命名和规范化字段。
Elasticsearch 使用模板、映射和生命周期策略对事件进行索引。
Kibana 通过数据视图查询Elasticsearch，并在Discover、仪表盘、Lens或告警中显示事件。

大多数集成问题发生在边界处。Logstash无法连接、Elasticsearch拒绝文档，或者Kibana查看的是错误的数据视图或时间范围。

确保数据流清晰的Logstash配置

Logstash管道包含三个主要部分：input、filter和output。保持每个部分简单且可测试。

输入插件

常见的输入插件包括：

beats：接收来自Filebeat、Metricbeat和其他Beats的事件。
tcp / udp：通过网络套接字接收事件。
file：读取本地文件。这对于小型部署和测试很有用，但对于分布式生产主机，通常使用代理更好。
syslog：接收syslog消息。

带TLS的Beats输入示例：

input {
  beats {
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/etc/pki/tls/certs/logstash.crt"
    ssl_key => "/etc/pki/tls/private/logstash.key"
  }
}

确保端口已打开，证书与客户端连接方式匹配，并且选项名称与您安装的插件版本一致。最近的Beats输入插件版本使用ssl_enabled。

过滤器插件

过滤器将原始事件转换为有用的字段。顺序很重要，因为Logstash会按顺序运行过滤器。

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }

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

使用grok处理非结构化文本，date设置@timestamp，mutate清理字段，当需要基于IP的位置丰富时使用geoip。在生产环境中使用之前，请针对实际日志行测试grok模式。一个小的解析错误可能会导致数千个事件以缺失字段的方式发送到Elasticsearch。

输出插件

对于ELK堆栈，Elasticsearch输出是通常的目标。

output {
  elasticsearch {
    hosts => ["https://elasticsearch-node1:9200", "https://elasticsearch-node2:9200"]
    index => "my-logs-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${LOGSTASH_ES_PASSWORD}"
    ssl_enabled => true
    cacert => "/etc/logstash/certs/http_ca.crt"
  }
}

index值是Elasticsearch模板和Kibana数据视图的契约。如果Logstash写入my-logs-2026.05.23，您的模板和数据视图应匹配my-logs-*。

对于较大的环境，考虑使用数据流和索引生命周期管理，而不是手动管理的每日索引。如果使用数据流，请遵循Elastic当前关于Logstash输出中data_stream设置的指南，而不是混合使用数据流和经典索引选项。

Elasticsearch模板和映射

Elasticsearch需要在文档到达之前具有一致的映射。否则，第一个文档可能会设置一个字段类型，从而破坏后续事件。首先以"200"形式到达的状态码可能变成text或keyword，而不是数字。

可组合索引模板示例：

PUT _index_template/my_log_template
{
  "index_patterns": ["my-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": {
      "properties": {
        "@timestamp": {"type": "date"},
        "message": {"type": "text"},
        "host.name": {"type": "keyword"},
        "log.level": {"type": "keyword"},
        "http.response.status_code": {"type": "integer"}
      }
    }
  }
}

对于精确匹配和聚合使用keyword，全文搜索使用text，指标和状态码使用数字类型，时间字段使用date。除非有测量理由需要增加，否则保持分片数量适中。太多小分片可能会损害集群性能。

Kibana数据视图

Kibana当前UI使用数据视图。旧版本称之为索引模式。创建一个与Elasticsearch实际拥有的索引名称或数据流匹配的数据视图。

典型设置：

转到Stack Management -> Kibana -> Data Views。
创建一个数据视图，例如my-logs-*。
选择@timestamp作为时间字段。
打开Discover并在测试时扩大时间选择器。

如果Discover为空，不要假设Logstash失败。检查时间范围、数据视图模式以及@timestamp是否正确解析。

常见集成问题排查

数据未出现在Kibana中

检查每个环节：

GET _cat/indices/my-logs-*?v
GET my-logs-*/_search?size=1&sort=@timestamp:desc

然后检查：

Logstash日志中的连接、认证、TLS或映射错误。
Elasticsearch日志中的拒绝文档和安全失败。
Kibana数据视图模式和选定的时间范围。
事件时间戳是否在未来、过去或缺失。

文档被Elasticsearch拒绝

映射冲突很常见。例如，一个事件发送http.response.status_code为200，而另一个发送"OK"。Elasticsearch无法将两者都存储在integer字段中。

修复Logstash过滤器，使字段类型一致，或者将错误事件路由到单独的索引以供审查。不要在没有修复产生错误文档的管道的情况下，反复删除和重新创建索引。

Logstash CPU使用率过高

昂贵的grok模式、高事件量和大型多行事件可能会迅速推高Logstash CPU。首先测量哪个管道繁忙，然后简化模式、锚定正则表达式，并在操作更简单时将简单解析移到Beats或Elasticsearch摄取管道。

Kibana查询缓慢

缓慢的仪表盘通常源于宽时间范围、高基数聚合、过多分片，或者字段映射为text而Kibana需要keyword。使用更窄的仪表盘默认值、ILM滚动和与可视化匹配的字段映射。

总结

将ELK Stack集成视为三个层之间的契约。Logstash必须生成可预测的字段，Elasticsearch必须正确映射和存储它们，Kibana必须在正确的时间范围内查询正确的数据视图。当出现问题时，跟踪一个示例事件从输入到索引再到仪表盘的完整路径。