logstash高可用之队列、死信队列

logstash高可用之队列、死信队列 1 队列

应对瞬间流量爆炸的通用机制就是使用队列，将瞬时流量先缓存起来再交由后台系统处理，后台系统能处理多少就从队列中取多少，从而避免了因流量爆炸导致的系统崩溃。使用队列，能够应对瞬间流量爆炸、提高系统可用性的机制，它利用了队列先进先出的机制平滑事件流量的峰谷，起到了削峰填谷的重要作用。

1.1 持久化队列

Logstash输入插件默认使用基于内存的事件队列，这就意味着如果logstash意外崩溃，队列中未处理的事件将全部丢失。不仅如此，基于内存的队列不仅容量小且不可以通过配置扩大容量，所以它能够起到的缓存作用也就非常有限。

为了应对这些问题，可以将事件队列配置为基于硬盘存储的持久化队列。

配置logstash.yml

http.host: "0.0.0.0"
# 发送系统数据到es所需的验证
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]
xpack.monitoring.elasticsearch.username: "logstash_system"
xpack.monitoring.elasticsearch.password: "123456"

# 基于硬盘的持久化队列
queue.type: persisted
path.queue: ${path.data}/queue

1.2 死信队列

仅当elasticsearch输出支持死信队列功能 。此外，死信队列仅用于响应代码为400或404的情况，两者都表示无法重试的事件。在将来的Logstash插件版本中将提供对其他输出的支持。在配置Logstash以使用此功能之前，请参阅输出插件文档以验证插件是否支持死信队列功能。

默认情况下，当Logstash遇到由于数据包含映射错误或其他问题而无法处理的事件时，Logstash管道会挂起或丢弃不成功的事件。为了防止在这种情况下丢失数据，您可以配置Logstash将不成功的事件写入死信队列而不是丢弃它们。

写入死信队列的每个事件都包括原始事件，描述无法处理事件的原因的元数据，有关编写事件的插件的信息以及事件进入死信队列的时间戳。

要处理死信队列中的事件，只需创建一个Logstash管道配置，该配置使用 dead_letter_queue输入插件从队列中读取。

配置logstash.yml(多管道时可以在pipline.yml中分别配置)

# 高可用配置
dead_letter_queue.enable: true

死信队列作为文件存储在 Logstash 实例的本地目录中。默认情况下，死信队列文件存储在 path.data/dead_letter_queue中. 每个管道都有一个单独的队列和死信队列。
例如，默认情况下main存储管道的死信队列。在LOGSTASH_HOME/data/dead_letter_queue/main队列文件按顺序编号：1.log、2.log等。

可以在文件中设置path.dead_letter_queue为logstash.yml文件指定不同的路径：(多管道时可以在pipline.yml中分别配置)

# 高可用配置
dead_letter_queue.enable: true
path.dead_letter_queue: "${path.data}/dead_letter_queue"

1.3 死信队列配置

死信队列配置文件，由于存到死信队列的事件是原始事件，所以需要进行filter

dlq.conf

input {
  dead_letter_queue {
    path => "/usr/share/filebeat/data/dead_letter_queue"
	commit_offsets => true 
    pipeline_id => "main" 
  }
}

filter {
    xxx
}

# 输出追加到日志中[上线时注释掉]
output {
	stdout { codec => rubydebug }
}

output {
	if [flag] == "elk" {
		elasticsearch {
			hosts => ["http://elasticsearch:9200"]
            user => "elastic"
            password => "xxx"
			manage_template => true
			index => "webserver-log-%{+YYYY.MM.dd}"
			#document_type=> "_doc"   
			template => "/usr/share/logstash/templates/webserver-log.json"
			template_overwrite => true
	  }
	}
}

path :包含死信队列的顶级目录的路径。此目录包含一个单独的文件夹，用于写入死信队列的每个管道。要查找此目录的路径，请查看logstash.yml 设置文件。默认情况下，Logstash dead_letter_queue在用于持久存储的位置 ( path.data) 下创建目录，例如LOGSTASH_HOME/data/dead_letter_queue. 但是，如果path.dead_letter_queue已设置，它将使用该位置。

commit_offsets:保存偏移量。当管道重新启动时，它将继续从它停止的位置读取，而不是重新处理队列中的所有项目。当您正在探索死信队列中的事件并想要多次迭代事件时，您可以设置commit_offsets为。false

pipeline_id :正在写入死信队列的管道的 ID。默认值为"main".

1.4 修改启动项启动死信队列管道

在启用死信队列前，使用
"./bin/logstash -f ./pipeline/logstash-to-es.conf"指定配置启动
在启用死信队列后，我们就有两个队列了，需要使用
./bin/logstash来进行多管道启动，同时修改pipline.yml配置文件来指定

pipline.yml

- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline/logstash-to-es.conf"
  #pipeline.workers: 3 默认cpu核心数
  config.reload.automatic: true
  
- pipeline.id: dlq
  path.config: "/usr/share/logstash/pipeline/dlq.conf"
  #pipeline.workers: 1
  config.reload.automatic: true

1.5 注意

死信队列无法在上游管道运行的情况下被清除。

死信队列是一个页面目录。要清除它，请停止管道并删除

${path.data}/dead_letter_queue/${pipeline_id}

管道在重新启动时会创建一个新的死信队列。

1.6 pipline.yml配置文件写法

3、配置

注意一下所有的配置文件请设置成utf-8格式，不然启动可能会报错

配置方式有三种
 
1、直接写input,output这样，使用config.string字段
 - pipeline.id: test
   pipeline.workers: 1
   pipeline.batch.size: 1
   config.string: "input { generator {} } filter { sleep { time => 1 } } output { stdout { codec => dots } }"
 
2、使用配置文件的路径 使用path.config字段
 
 - pipeline.id: another_test
   queue.type: persisted
   path.config: "/tmp/logstash/a.config"
 
3、使用通配符格式 path.config=/tmp/logstash/conf.d/*.conf
 
 - pipeline.id: another_test
   queue.type: persisted
   path.config: "/tmp/logstash/conf.d/*.conf"
 
 
#多个路径分开写
 
  - pipeline.id: kafka
    pipeline.workers: 2 #线程数默认与cpu核数一致
    pipeline.batch.size: 1 #批量处理的条数默认125
    path.config: "/usr/local/logstash/config/logstash-kafka.conf"
  - pipeline.id: es
    queue.type: persisted #队列持久化，防止丢失数据，默认不开启
    path.config: "/usr/local/logstash/config/logstash-es.conf"