ELK - Logstash - Data Resiliency

当数据流过事件处理管道时，Logstash可能会遇到阻止它向配置的输出传递事件的情况。例如，数据可能包含意外的数据类型，或者Logstash可能异常终止。

为了防止数据丢失并确保事件流过管道而不中断，Logstash提供了以下数据弹性特性。

持久队列通过在磁盘上的内部队列中存储事件来防止数据丢失。

Dead Letter Queues 为Logstash无法处理的事件提供磁盘存储。通过使用 dead_letter_queue 输入插件，可以轻松地重新处理 dead letter queue 中的事件。

默认情况下禁用这些弹性特性。要打开这些特性，必须在Logstash设置文件中显式启用它们。

Persistent Queues

默认情况下，Logstash在pipeline 阶段（inputs → pipeline workers）之间使用内存中的有界队列来缓冲事件。这些内存中队列的大小是固定的，无法配置。如果Logstash遇到临时机器故障，内存队列的内容将丢失。临时机器故障是指Logstash或其主机异常终止，但能够重新启动的情况。

为了防止异常终止期间的数据丢失，Logstash具有将消息队列存储在磁盘上的持久队列特性。持久队列提供Logstash中的数据的持久性。

持久队列对于需要大缓冲区的Logstash部署也很有用。代替部署和管理消息代理（如Redis、RabbitMQ或Apache Kafka）以促进缓冲的 publish-subscriber 模型，您可以启用持久队列来缓冲磁盘上的事件并删除消息代理。

总之，启用持久队列的好处：

解决突发事件而不需要外部缓冲机制，如Redis或Apache Kafka。

提供至少一次传递保证，以防止在正常关闭期间以及当Logstash异常终止时消息丢失。如果在事件进行中时重新启动Logstash，则Logstash将尝试传递存储在持久队列中的消息，直到传递成功至少一次。

Persistent Queues 如何工作

input → queue → filter + output

当一个输入准备好处理事件时，它将它们写入队列。当对队列的写入成功时，输入可以向其数据源发送确认。

当处理来自队列的事件时，Logstash仅在过滤器和输出完成之后才确认队列中的事件已完成。队列保存由管道处理的事件的记录。如果并且仅当日志存储管道已经完全处理了事件，则记录为已处理（在本文档中，称为“确认”或“ACKed”）。

被承认是什么意思？这意味着事件已经由所有配置的过滤器和输出处理。例如，如果只有一个输出Elasticsearch，则当Elasticsearch输出将此事件成功发送到Elasticsearch时，事件为ACKed。

在正常关机（CTRL+C或SIGTERM）期间，Logstash将停止从队列中读取，并将完成对过滤器和输出正在处理中的事件的处理。重新启动后，Logstash将恢复处理持久队列中的事件以及从输入接受新事件。

如果Logstash异常终止，则任何飞行中的事件都不会被ACKed，并且在重新启动Logstash时将由过滤器和输出重新处理。Logstash以批处理事件，所以对于任何给定的批，当发生异常终止时，该批中的一些可能已经成功完成，但是没有记录为ACKed。

配置 Persistent Queues

queue.type: 指定 persisted 启用持久队列。默认情况下，禁用持久队列 (default: queue.type: memory).

path.queue: 存储数据文件的目录路径。默认情况下，文件存储在 path.data/queue 中。

queue.page_capacity: 以字节为单位的队列页的最大大小。队列数据由名为“pages”的仅附加的文件组成。默认大小是64mb。更改此值不太可能有性能好处。

queue.drain: 如果希望Logstash在关闭之前等待持久队列耗尽，请指定true。耗尽队列所需的时间量取决于队列中累积的事件的数量。因此，应该避免使用此设置，除非队列（即使已满）相对较小并且可以很快耗尽。

queue.max_events: 队列中允许的最大事件数。默认值为0（无限制）。

queue.max_bytes: 队列总容量的字节数。默认值为1024mb（1gb）。确保磁盘驱动器的容量大于这里指定的值。

Handling Back Pressure

当队列已满时，Logstash对输入施加压力，以停止流入Logstash的数据。这种机制帮助Logstash在输入阶段控制数据流的速率，而不会像Elasticsearch那样压倒输出。

使用queue.max_bytes设置来配置磁盘上队列的总容量。

通过指定这些设置，Logstash将缓冲磁盘上的事件，直到队列的大小达到8gb。当队列中充满了unACKed事件，并且已经达到大小限制时，Logstash将不再接受新的事件。

每个输入独立处理 Back Pressure 。例如，当beats输入遇到反压力时，它不再接受新的连接并等待，直到持久队列有空间接受更多的事件。在过滤器和输出阶段完成队列中现有事件的处理并ACK之后，Logstash自动开始接受新事件。

Controlling Durability

持久性是存储写入的属性，它确保数据在写入之后可用。

当启用持久队列特性时，Logstash将在磁盘上存储事件。Logstash在称为检查点的机制中提交到磁盘。

首先，队列本身是一组页面。有两种页面：head pages and tail pages.。head page 是编写新事件的地方。只有一个head page 。当 head page 具有一定大小时（queue.page_capacity），它变成一个 tail page，并创建一个新的 head page 。tail page 是不可变的，而 head page 是只附加的。其次，队列在称为检查点文件的单独文件中记录关于其自身（页面、确认等）的细节。

记录检查点时，Logstash将：

• 在head page 调用fsync。
• 将队列的当前状态自动写入磁盘。

检查点的过程是原子性的，这意味着如果成功，将保存对文件的任何更新。

通过设置queue.checkpoint.writes，可以更频繁地强制Logstash到检查点。此设置指定强制检查点之前可能写入磁盘的事件的最大数量。默认值为1024。为了确保最大的持久性并避免在持久队列中丢失数据，可以设置queue.checkpoint.writes:1以在写入每个事件之后强制检查点。请记住，磁盘写入具有资源成本。将这个值设置为1，会严重影响性能。

Disk Garbage Collection

在磁盘上，队列作为一组页存储，其中每个页都是一个文件。每个页面的大小最多可以是queue.page_capacity 。在该页中的所有事件都被ACKed之后，将删除（垃圾收集）页。如果旧页面至少有一个事件尚未ACKed，则整个页面将保留在磁盘上，直到成功处理该页面中的所有事件。包含未处理事件的每个页面将根据queue.max_bytes 字节大小进行计数。

Dead Letter Queues

dead letter queue 特性目前只支持弹性搜索输出。此外， dead letter queue 只在响应代码是400或404的情况下使用，这两者都指示不能重试的事件。

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

Configuring Logstash to Use Dead Letter Queues

默认是关闭的。如果想要打开，要在logstash.yml配置文件中，设置：

dead_letter_queue.enable: true

存储的文件路径。默认存在 path.data/dead_letter_queue

path.dead_letter_queue

File rotation

Dead letter queues 具有管理队列文件大小的内置文件旋转策略。当文件大小达到预先配置的阈值时，将自动创建新文件。

默认情况下，每个 Dead letter queues 的最大大小设置为1024mb。要更改此设置，请使用 dead_letter_queue.max_bytes 选项。如果条目将增加超出此设置的 Dead letter queues 的大小，则会删除这些条目.

Processing Events in the Dead Letter Queue

当准备处理 dead letter queue 中的事件时，您将创建一个管道，该管道使用 dead_letter_queue 输入插件从 dead letter queue 中读取。当然，您使用的管道配置取决于您需要做什么。例如，如果dead letter queue 包含由Elasticsearch中的映射错误导致的事件，则可以创建读取“dead”事件的管道，移除导致映射问题的字段，并将干净的事件重新索引到Elasticsearch中。

下面的示例显示了一个简单的管道，该管道从死信队列读取事件，并将事件（包括元数据）写入标准输出：

input { dead_letter_queue { path => "/path/to/data/dead_letter_queue" commit_offsets => true pipeline_id => "main" } } output { stdout { codec => rubydebug { metadata => true } } }

当管道已完成对dead letter queue 中的所有事件的处理时，它将在流入队列时继续运行和处理新事件。这意味着您不需要停止生产系统来处理dead letter queue 中的事件。

Reading From a Timestamp

当从dead letter queue 读取数据时，您可能不希望处理队列中的所有事件，特别是如果队列中有很多旧事件。可以使用start_timestamp选项在队列中的特定点开始处理事件。此选项将管道配置为基于事件进入队列时的时间戳开始处理事件：

input { dead_letter_queue { path => "/path/to/data/dead_letter_queue" start_timestamp => 2017-06-06T23:40:37 pipeline_id => "main" } }

对于这个示例，管道开始读取这个时间之后的传递到dead letter queue 的所有事件。