Como evitar a perda de mensagens no Logstash com Dead letter queue (DLQ)

O Logstash é um pipeline de processamento de dados do lado do servidor de código aberto que ingere dados de várias fontes, transforma-os simultaneamente e os envia para seu "stash" favorito, o mais comum é o Elasticsearch.

A coleta de dados é realizada por meio de plugins de entrada configuráveis. Depois que um plugin de entrada coleta dados, ele pode ser processado por plugins de filtro que modificam e anotam os dados do evento. Por fim, o Logstash roteia eventos para plugins de saída que podem encaminhar os eventos para uma variedade de ferramentas, incluindo Elasticsearch.

Qaundo o Logstash encontra um evento que não pode processar (erro de mapeamento ou algum outro problema) a pipeline do Logstash trava ou descarta o evento malsucedido. Para proteger contra perda de dados nessa situação, podemos configurar o Logstash para gravar eventos malsucedidos em uma fila de mensagens em vez de descartá-los. Esta fila é chamada de Dead letter queue (DLQ)

Neste artigo, quero mostrar um exemplo prático de como configurar essa fila!

Laboratório

Crie o diretório:

sudo mkdir -p /opt/dlq
sudo chown -R logstash:logstash /opt/dlq

Edite o arquivo logstash.yml

dead_letter_queue.enable: true
path.dead_letter_queue: /opt/dlq

Crie um arquivo json com o seguinte conteúdo:

sudo vim /opt/dlq/teste-dql.json

{"age":39,"full_name":"Shelley Bangs","gender":"Female"}
{"age":32,"full_name":"Sally Penylton","gender":"Female"}
{"age":39,"full_name":"Janot Maxfield","gender":"Female"}
{"age":28,"full_name":"Isaak Taynton","gender":"Male"}
{"age":36,"full_name":"Pavel Braund","gender":"Male"}
{"age":43,"full_name":"Eleanore Seaton","gender":"Female"}
{"age":46,"full_name":"Ely Fullilove","gender":"Male"}
{"age":23,"full_name":"Deeann Moon","gender":"Female"}
{"age":27,"full_name":"Niels Milam","gender":"Male"}
{"age":23,"full_name":"Lorne Cuxson","gender":"Female"}
{"age":false,"full_name":"Robbyn Narrie","gender":"Female"}
{"age":true,"full_name":"Clyde Marney","gender":"Male"}
{"age":false,"full_name":"Frankie Semble","gender":"Male"}
{"age":true,"full_name":"Aggy Reditt","gender":"Female"}
{"age":true,"full_name":"Fionna Dozdill","gender":"Female"}
{"age":true,"full_name":"Erroll Hallut","gender":"Male"}
{"age":true,"full_name":"Gayle Terrell","gender":"Female"}
{"age":true,"full_name":"Lucita Garthside","gender":"Female"}
{"age":true,"full_name":"Renaud Djurkovic","gender":"Male"}
{"age":false,"full_name":"Joellen Strick","gender":"Female"}

Esse documento json é o que vamos utilizar para simular a perda de dados, é possível observar que as mensagens finais contém o campo age com dado do tipo booleano e não inteiro.

Crie a pipeline:

sudo vim /etc/logstash/conf.d/exemplo-dlq.conf

input {
  file {
    codec => "json"
    start_position => "beginning"
    path => "/opt/dlq/teste-dql.json"

  }
}
output {
  elasticsearch {
    hosts => [ "https://node-master:9200", "https://node-data:9200" ]
    index => "dlq-exemplo-dados"
    user => "sysadmin"
    password => "4linux"
    cacert => "/etc/logstash/certs/elasticsearch-ca.pem"
  }

  stdout {
    codec => "rubydebug"
  }
}

Execute o logstash:

sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash/ -f /etc/logstash/conf.d/exemplo-dlq.conf

A pipeline será executada, o logstash vai retornar o erro que ocorreu ao tentar anexas as ultimas mensagens mas a pipeline não será encerrada. Ao verificar a quantidade de documentos no elasticsearch, pode-se observar que algumas não foram enviadas:

GET /_cat/indices/dlq*?v

Resultado:

health status index             uuid                   pri rep docs.count docs.deleted store.size pri.store.size
green  open   dlq-exemplo-dados 7i3c7U1yRCaGt4_QESE36g   1   1         10            0     31.7kb         19.7kb

Na pasta /opt/dlq, é possível observar que o logstash criou uma pasta com o nome da pipeline:

root@monitoring:/opt/dlq# ls
main  teste-dql.json
root@monitoring:/opt/dlq# ls -lha main/
total 28K
drwxrwxr-x 2 logstash logstash 4.0K Dec  3 11:15 .
drwxr-xr-x 3 logstash logstash 4.0K Dec  3 11:15 ..
-rw-rw-r-- 1 logstash logstash  14K Dec  3 11:15 1.log
-rw-rw-r-- 1 logstash logstash    1 Dec  3 11:15 2.log.tmp
-rw-rw-r-- 1 logstash logstash    0 Dec  3 11:15 .lock

Para recuperar esses dados, vamos criar uma pipeline e utilizar o módulo dead_letter_queue:

sudo vim /etc/logstash/conf.d/dlq.conf

input {
  dead_letter_queue {
    path => "/opt/dlq/"
  }
}
output {
  elasticsearch {
    hosts => [ "https://node-master:9200", "https://node-data:9200" ]
    index => "dlq-dados-perdidos"
    user => "sysadmin"
    password => "4linux"
    cacert => "/etc/logstash/certs/elasticsearch-ca.pem"
  }

  stdout {
    codec => "rubydebug"
  }
}

Execute a pipeline:

sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash/ -f /etc/logstash/conf.d/dlq.conf

Agora temos 2 indices com 10 documentos:

health status index              uuid                   pri rep docs.count docs.deleted store.size pri.store.size
green  open   dlq-dados-perdidos D4rLwdaZTSWXF-ne79DaOw   1   1         10            0       18kb          8.9kb
green  open   dlq-exemplo-dados  uJTSqP5RSHypoxCdZLYHLw   1   1         10            0       18kb            9kb

Neste exemplo eu enviei os dados para outro índice porquê neste caso, não conseguimos converter booleano para inteiro. Mas dependendo do seu tipo de erro, é possível tratá-lo utilizando os plugins de filtro do logstash e enviar para o mesmo índice.