使用 Logstash 可靠性和可用性记录到 elasticsearch，处理丢失的日志？

Question

我正在关注 ELK 上的这篇文章：Building Logging System in Microservice Architecture with ELK Stack and Serilog .NET Core，在这个架构中，serilog 记录到 Logstash，然后将 logstash 推送到 Elastic 搜索，如果无法访问弹性搜索或者如果 logstash 服务关闭或者如果网络宕机或系统宕机，简而言之，我要去哪里，如何确保我的日志在所有情况下都不会丢失，日志也应该可以离线使用，我想先存储在日志文件中然后处理从文件到弹性搜索的日志，但是文件会在一段时间内增长，然后我必须注意没有重复的消息并且消息也需要删除，最重要的是没有死锁类型关于读写文件的情况，能否请您帮我了解一下ELK是否会照顾，如果我使用fluend或fluentbit而不是logstash，它们会更好吗？？

代码：

var log = new LoggerConfiguration()
         .WriteTo.Console()
         .WriteTo.Http("http://localhost:8080")
         .CreateLogger();


while (true)
{
    var customer = Customer.Generate();
    log.Information("{@customer} registered", customer);
    Thread.Sleep(1000);
}

输出：

[13:56:02 INF] {"FirstName": "Lourdes", "LastName": "Kreiger", "SSNumber": "350-11-7869", "$type": "Customer"} registered
[13:56:03 INF] {"FirstName": "Desmond", "LastName": "Balistreri", "SSNumber": "929-58-1854", "$type": "Customer"} registered
...

使用 ELK 发送日志

Http输入监听端口8080

input {
    http {
        #default host 0.0.0.0:8080
        codec => json
    }
}

# Separate the logs
filter {
    split {
        field => "events"
        target => "e"
        remove_field => "events"
    }
}

# Send the logs to Elasticsearch
output {
    elasticsearch {
        hosts => "elasticsearch:9200"
        index=>"customer-%{+xxxx.ww}"
    }
}

Answer 1

您有两个可能导致数据丢失的主要故障点，logstash 和 elasticsearch 之间的通信以及您的服务和 logstash 之间的通信。

logstash 和 elasticsearch 之间的通信

默认情况下，将数据发送到 elasticsearch 时，logstash 使用管道的 input 块和 filter 块之间的内存队列，此队列用于在 logstash 无法与 elasticsearch 通信的情况下存储事件。

这个内存队列有一个固定大小的1000个事件，所以如果你每秒有很多事件，它并没有多大帮助。您可以更改管道以使用persisted queue，此队列将执行与内存中队列相同的操作，但它会写入logstash 服务器中的文件，您可以更改文件大小以存储更多事件。

如果持久化队列满了而elasticsearch仍然宕机，logstash将停止接受新事件，队列何时填满完全取决于队列的大小、事件的大小和事件的速率，但是持久化队列是您可以做的事情之一，以避免在 logstash 和 elasticsearch 之间丢失数据。

您的服务和 logstash 之间的通信

如果您的服务无法与 logstash 通信，那么您需要在其上实现一些逻辑以避免数据丢失。如何做到这一点完全掌握在您的手中。

您可以复制 logstash 使用的持久队列并将未发送到 logstash 的事件写入文件，然后在 logstash 返回时重播这些事件。

这会增加很多额外的东西，你需要自己去实现。

替代方案

我想说最好的方法是将日志写入日志文件并使用filebeat 将这些日志发送到logstash，如果您不想在logstash、filebeat 中使用任何过滤器，甚至直接发送到elasticsearch可以在输出服务不可达的情况下自动重试发送日志，并且它还可以跟踪已经发送的内容。

由于您使用的是 dotnet，您可以使用 log4net 进行日志记录，它会处理日志记录部分并在达到指定大小时轮换您的日志。