当前位置：首页 > article >正文

ELK系列-（六）Redis也能作为消息队列？（下）

article 2024/12/23 4:41:23

一、前文回顾 🔍

在前面的ELK系列中，我们已经搭建了ELK的核心组件，包括：

ELK系列-（一）Docker部署ELK核心组件
ELK系列-（二）LogStash数据处理的瑞士军刀
ELK系列-（三）Kibana 数据可视化的艺术家
ELK系列-（四）轻量级的日志收集助手-Beat家族
ELK系列-（五）指标收集-MetricBeat（上）
ELK系列-（六）Redis也能作为消息队列？（上）

系统架构图 📊

Screenshot 2024-12-10 at 13.46.34

有关整个系统架构的部署，您可以回顾前面的文章。今天，我们将继续探讨Redis在ELK系统中的落地

关于Redis的选型、Beats的部署使用，请翻阅往期博客哦

二、Beat发送消息到Redis

第一步就是将Beats的消息发送到Redis，因此我们需要修改Beats的配置如下：

output.redis:
  # 启用或禁用输出模块的布尔标志。
  enabled: true

  # 要连接的 Redis 服务器列表。如果启用了负载均衡，则事件将分发到列表中的服务器。如果某个服务器无法访问，则事件仅分发到可访问的服务器。
  # hosts 设置支持带有自定义密码的 redis 和 rediss URLs，例如 redis://:password@localhost:6379。
  hosts: ["121.**.60.**:6379"]

  # 用于认证 Redis 的密码。默认不使用认证。
  password: your_password

  # 事件发布到的 Redis 数据库号。默认值为 0。
  db: 9

  # 用于发布事件的 Redis 数据类型。如果数据类型为 list，则使用 Redis RPUSH 命令。如果数据类型为 channel，则使用 Redis PUBLISH 命令。默认值为 list。
  datatype: list
  
  # 配置的每个主机用于发布事件到 Redis 的工作线程数。将此设置与负载均衡选项一起使用。例如，如果你有 2 个主机和 3 个工作线程，总共启动 6 个工作线程（每个主机 3 个）。
  worker: 1

三、 LogStash从Redis获取消息

改为多Pipeline

以前单管道，现在要引入Beats消息，要改为多管道

# 原本
/usr/share/logstash/pipeline/logstash.conf
# 改造后
/usr/share/logstash/pipeline/conf.d/metricbeat.conf
/usr/share/logstash/pipeline/conf.d/application_log.conf

配置文件(logstash.yml)也挂载出来:

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]
path.config: "/usr/share/logstash/pipeline/conf.d/*.conf"

metricbeat配置

其配置存在特殊性，输出如果和普通的一样配置，不会纳入Kibana的Observability视图

经过反复阅读文档，调试，发现Metricbeat使用的是数据流，即多个索引的特殊组合。所以此处要使用数据流作为输出

数据流会在下一篇讲哦

修改metricbeat.conf，使用如下配置：

# 从redis获取
input {
  redis {
    host => "121.**.60.**"
    port => 6379
    password => "**"
    data_type => "channel"
    key => "metricbeat"
    db => 9
    threads => 1
    tags => ["metricbeat"]
  }
}

# metricbeat收集的数据打时间标签
filter {
  if "metricbeat" in [tags] {
    date {
      match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
      timezone => "Asia/Shanghai"
      target => "@timestamp"
    }
  }
}

# 指定输出，数据流相关信息
output {
  if "metricbeat" in [tags] {
    elasticsearch {
      hosts => ["es:9200"]
      data_stream => true
      data_stream_type => "metrics"
      data_stream_dataset => "8.15.0"
      data_stream_namespace => "metricbeat"
    }
  }
}