DEV Community: Leandro Pereira

logstash: melhorando o desempenho trocando o grok pelo dissect.

Leandro Pereira — Sun, 25 Oct 2020 22:45:03 +0000

Já tem um tempo que utilizo a pilha da elastic pra coletar e armazenar logs de aplicações, serviços e dispositivos que preciso monitorar. Sendo o Logstash o responsável em receber, tratar e publicar os logs, garantir que ele tenha um bom desempenho é extremamente importante.

Boa parte dos problemas de desempenho que enfrentei estavam relacionados a erros de configuração, implantação ou uso incorreto de alguns filtros, principalmente o filtro grok.

Embora o grok seja extremamente útil, ele é um filtro baseado em expressões regulares e que precisa validar cada expressão configurada, dependendo do número de eventos por segundo essas validações podem acabar gerando uma grande carga na CPU da máquina do Logstash, o que pode impactar em todo o processo de monitoramento.

Grok ou Dissect?

Uma forma simples de otimizar o desempenho de um pipeline é verificar a possibilidade de utilizar outro filtro no lugar do grok. Um filtro que eu venho utilizando bastante como substituto é o dissect.

A principal diferença entre o grok e o dissect é que o dissect não utiliza expressões regulares para analisar a mensagem, os campos são definidos a partir de sua posição, tudo que estiver entre um %{ e um } é considerado um campo e todo o resto é considerado um delimitador, isso torna o dissect mais rápido que o grok e também exige menos processamento durante a análise.

Considere o exemplo de mensagem a seguir:

2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message

Podemos extrair dessa mensagem os seguintes campos:

timestamp, ipaddr, thread, logger, loglevel, msg

Para analisarmos essa mensagem com o grok e dissect usamos as seguintes configurações.

grok

grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{IP:ipaddr} \[%{INT:thread}\] %{DATA:logger} %{WORD:loglevel} - %{GREEDYDATA:msg}"}
}

dissect

dissect {
    mapping => {
        "message" => "%{timestamp} %{ipaddr} [%{thread}] %{logger} %{loglevel} - %{msg}"
    }
}

Perceba que a estrutura é bem semelhante, mas no grok precisamos especificar contra qual expressão regular iremos validar o campo, como por exemplo TIMESTAMP_ISO8601 para o campo timestamp ou IP para o campo ipaddr, algo que não temos no dissect , já que o filtro irá armazenar como valor do campo o que estiver naquela posição na mensagem e isso acaba sendo um dos fatores que influencia na escolha de um ou outro.

Se as mensagens coletadas em um pipeline possuem sempre a mesma estrutura, ou até mesmo poucas variações, o uso do dissect no lugar do grok se faz possível e se mostra vantajoso, já que a análise será mais rápida e precisará de menos processamento.

Mas será que é mesmo mais rápido e usa menos processamento?

Grok vs Dissect

Para comparar o desempenho dos dois filtros utilizei um pipeline simples usando como input o filtro generator e como output o filtro stdout, e alternei o uso dos filtros grok e dissect exemplificados anteriormente.

input {
    generator {
        lines => ["2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message"]
        count => 10000000
    }
}

filter {
    #
    # filtro de grok ou dissect
    #
    if [sequence] != 0 and [sequence] != 9999999 {
         drop { }
     }
}
output {
    stdout { }
}

Esse pipeline basicamente gera 10 milhões de mensagens, aplica o filtro especificado no momento, grok ou dissect, e faz um drop na mensagem, exceto a primeira e a última, que são exibidas apenas para verificar o início e fim do processamento.

A saída desse pipeline é a seguinte.

{
          "host" => "elk",
            "ipaddr" => "127.0.0.1",
       "message" => "2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message",
      "sequence" => 0,
      "loglevel" => "INFO",
        "logger" => "program.Logger",
      "@version" => "1",
     "timestamp" => "2020-08-20T21:45:50",
           "msg" => "sample message",
        "thread" => "1",
    "@timestamp" => 2020-08-26T02:00:38.127Z
}
{
          "host" => "elk",
            "ipaddr" => "127.0.0.1",
       "message" => "2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message",
      "sequence" => 9999999,
      "loglevel" => "INFO",
        "logger" => "program.Logger",
      "@version" => "1",
     "timestamp" => "2020-08-20T21:45:50",
           "msg" => "sample message",
        "thread" => "1",
    "@timestamp" => 2020-08-26T02:02:34.018Z
}

Como estava interessado em comparar o processamento da máquina durante a execução dos filtros, utilizei o nmon rodando em background e coletando métricas a cada 1 segundo, durante 5 minutos, o que resulta em 300 medidas, uma quantidade mais do que suficiente para esse caso.

Para a execução do pipeline utilizei uma máquina virtual com 4 vCPUs e 4 GB de RAM rodando o CentOS 7.8 e o Logstash na versão 7.9.

Cada medida gerada pelo nmon tem o seguinte formato.

ZZZZ,T0010,00:16:49,21-AUG-2020
CPU001,T0010,50.5,2.0,0.0,46.5,1.0
CPU002,T0010,99.0,0.0,0.0,1.0,0.0
CPU003,T0010,97.0,1.0,0.0,1.0,1.0
CPU004,T0010,96.0,1.0,0.0,3.0,0.0
CPU_ALL,T0010,86.4,0.8,0.0,12.8,0.0,,4
MEM,T0010,3789.0,-0.0,-0.0,1024.0,2903.2,-0.0,-0.0,1024.0,-0.0,281.9,577.3,-1.0,2.0,0.0,181.5
VM,T0010,30,0,0,2384,12760,-1,0,0,0,0,27,0,0,2541,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0
PROC,T0010,5,0,2524.7,-1.0,-1.0,-1.0,1.0,-1.0,-1.0,-1.0
NET,T0010,2.6,0.1,2.8,0.1
NETPACKET,T0010,10.9,2.0,12.9,2.0
JFSFILE,T0010,59.0,0.0,0.5,59.0,24.3,4.5,4.5,4.5
DISKBUSY,T0010,0.0,0.0,0.0,0.0
DISKREAD,T0010,0.0,0.0,0.0,0.0
DISKWRITE,T0010,0.0,0.0,0.0,0.0
DISKXFER,T0010,0.0,0.0,0.0,0.0
DISKBSIZE,T0010,0.0,0.0,0.0,0.0

De todas essas linhas, somente a linha CPU_ALL importa, mais especificamente a terceira coluna que corresponde ao percentual de uso médio de todas as CPUs no momento da coleta.

Fazendo uma manipulação dos dados coletados durante a execução do pipeline com o grok e depois com o dissect podemos visualizar e comparar o consumo de CPU e o tempo de execução.

O pico inicial de processamento que vemos no gráfico é causado pela inicialização do Logstash, o platô seguinte corresponde ao processamento durante a análise das mensagens.

Analisando o gráfico vemos que o tempo para processar as 10 milhões de mensagens é basicamente o mesmo para os dois filtros testados nesse caso específico, mas temos uma diferença grande no processamento. Uma média de uso de CPU de 40% quando utilizamos o filtro dissect em comparação com uma média de uso de CPU de mais de 60% quando utilizamos o filtro grok.

Conclusão e Links

Dependendo do número de pipelines e da quantidade de eventos por segundo de cada um deles, dedicar um tempo para validar a possibilidade de troca do filtro grok pelo dissect é algo que pode otimizar bastante o desempenho durante o processo de análise e ingestão de dados, além de poder também refletir em redução de custos quando se tem uma infraestrutura em nuvem já que possibilita tanto o uso de máquinas menores, quanto menos consumo de créditos de cpu em alguns casos.

Nos casos onde os logs coletados seguem sempre a mesma estrutura, como acontece com logs de servidores web, servidores de aplicação, firewalls e roteadores, ou onde se tem controle sobre o formato dos logs gerados, o dissect é o filtro ideal quando se utiliza o Logstash no processo de coleta e análise de logs.

logstash: geolocation with geoip

Leandro Pereira — Sun, 25 Oct 2020 22:08:44 +0000

An important point in monitoring devices, applications or systems exposed on the internet, is to include information about the origin of the requests or attempts we receive.

This is useful on both the infrastructure and security side as well as the business side. For example, knowing the origin and number of requests or attempts, we can better plan the geographic distribution of servers, detect compromised credentials or attack attempts and identify new business opportunities in different locations.

A very simple way to do this when using the elastic stack as one of the monitoring tools is to use the geoip filter in logstash.

the geoip filter

The geoip filter has a very simple function, it queries an IP address in an internal database, identify its geolocation and returns some fields like the country name, country code, city, geographic coordinates and a few others.

In the case of the geoip filter used by logstash, the internal database with the geolocation information is the GeoLite2, provided by maxmind, and using this filter with an IP address it is possible to obtain, in addition to the geographical information, data related to the Autonomous System (AS), associated with the routing for the IP.

To use the geoip filter you need your event to have a field where the value is a public IP address and you also need to create a specific mapping for your index to store fields with geolocation data.

applying the filter

As an example of how to apply the geoip filter I will use a simple API that I made in Go that returns the status of the connections in a linux machine, this emulates part of what netstat does.

This API returns a JSON document with the following format.

{
    "srcip":"10.0.1.100",
    "srcport":56954,
    "dstip":"151.101.192.133",
    "dstport":443,
    "status":"ESTABLISHED"
}

The value of the srcip field is the local machine IP address and the value of the dstip field is the external IP address, this is the field that we will use with the geoip filter.

To query this API with logstash I will use the filter http_poller as the input in the pipeline, this basically make requests to an endpoint on a specificed schedule.

input {
    http_poller {
        urls => { "api" => "http://10.0.1.100:5000/netstat" } 
        schedule => { "every" => "30s"}
    }
}

The way that logstash receives the data makes no difference for the geoip filter, you only need a field with a public IP address.

The simplest configuration for the geoip filter has only one required option, the source, which is the name of the field that has the IP address that we want to find its geolocation data.

filter {
    geoip {
        source => "dstip"
    }
}

When an event pass through this filter with success, a new field named geoip is added to the event.

"geoip" => {
    "country_code2" => "IS",
    "continent_code" => "EU",
    "city_name" => "Reykjavik",
    "country_code3" => "IS",
    "region_code" => "1",
    "timezone" => "Atlantic/Reykjavik",
    "region_name" => "Capital Region",
    "location" => {
        "lon" => -21.9466,
        "lat" => 64.1432
    },
    "latitude" => 64.1432,
    "ip" => "31.209.137.10",
    "country_name" => "Iceland",
    "postal_code" => "101",
    "longitude" => -21.9466
}

If the geoip fails, the tag _geoip_lookup_failure will be added.

We can change the default behaviour using other options in the filter configuration.

target: name of the field where the geolocation data should be stored, the default is geoip.
default_database_type: it has only two options City and ASN, the first one is the default and gives you geographic information about the IP, the second one gives you information about the Autonomous System (AS) associated with the IP.
fields: the fields that should be returned, by default all available fields will be returned.
tag_on_failure: the name of the tag that should be added on failure, the default is _geoip_lookup_failure.

Since the idea is to enrich the events, we are going to use two geoip filters in sequence, one with the option default_database_type set as City and the other one with the same option set as ASN.

filter {
    geoip {
        default_database_type => "City"
        source => "dstip"
        target => "geo"
        tag_on_failure => ["geoip-city-failed"]
    }
    geoip {
        default_database_type => "ASN"
        source => "dstip"
        target => "geo"
        tag_on_failure => ["geoip-asn-failed"]
    }
}

filtering private addresses

When the field that we use as the source for the geoip filter can also have private IP addresses, we need to filter out those IPs.

A simple way to avoid that private IPs pass through the geoip filter is to use a conditional to add a tag to the events that have a private IP and another conditional to limit the geoip filter to only be applied on events without this tag.

In our example we need to filter out the network 10.0.1.0/24, the localhost IP 127.0.0.1 and the local routing IP 0.0.0.0.

if [dstip] =~ "^10.0.*" or [dstip] =~ "^127.0.*" or [dstip] == "0.0.0.0" { 
    mutate {
        add_tag => ["internal"]
    }
}

Now all the events where the destination is a private IP will have the tag internal and we can use this tag to avoid applying the geoip filter on those events.

if "internal" not in [tags] {
    geoip {
        default_database_type => "City"
        source => "dstip"
        target => "geo"
        tag_on_failure => ["geoip-city-failed"]
    }
    geoip {
        default_database_type => "ASN"
        source => "dstip"
        target => "geo"
        tag_on_failure => ["geoip-asn-failed"]
    }
}

mapping

Before we can start analyzing the data in elasticserch or create maps in kibana, we need to create a mapping to tell elasticsearch about the data type of every field.

Although elasticsearch can infer the data type and create a mapping at ingestion time, this does not work for the fields that need to store geolocation data, those fields need to be defined manually before the index creation.

The required part of the mapping is the one that defines the field geo.location as having the type geo_point, so if we use an index named endpoints, we would need to create the index and apply the mapping for this field.

PUT /endpoints
PUT /endpoints/_mapping
{
    "properties": {
        "geo": {
            "properties": {
                "location": {
                    "type": "geo_point"
                }
            }
        }
    }
}

This way we can guarantee that the field geo.location has the type geo_point and we can let elasticsearch create the mapping for the other fields when it index the first document.

While we can work like that without any problem, it is better to create a template for our index with the data type for every field.

In our example we can use the following template.

PUT _template/endpoints
{
    "order" : 0,
    "version" : 1,
    "index_patterns" : [ "endpoints" ],
    "settings" : {
      "index" : {
        "mapping" : {
          "ignore_malformed" : "true"
        },
        "refresh_interval" : "5s",
        "number_of_shards" : "1",
        "number_of_replicas" : "0"
      }
    },
    "mappings" : {
        "properties" : {
            "@timestamp" : {
                "format" : "strict_date_optional_time||epoch_millis",
                "type" : "date"
            },
            "@version" : { "type" : "keyword" },
            "status" : { "type" : "keyword" },
            "srcip" : { "type" : "ip" },
            "srcport" : { "type" : "keyword" },
            "dstip" : { "type" : "ip" },
            "dstport" : { "type" : "keyword" },
            "geo" : {
                "properties" : {
                    "as_org" : { "type" : "keyword" },
                    "asn" : { "type" : "keyword" },
                    "country_code2" : { "type" : "keyword" },
                    "country_code3" : { "type" : "keyword" },
                    "country_name" : { "type" : "keyword" },
                    "continent_code" : { "type" : "keyword" },
                    "city_name" : { "type" : "keyword" },
                    "region_code" : { "type" : "keyword" },
                    "region_name" : { "type" : "keyword" },
                    "postal_code" : { "type" : "keyword" },
                    "ip" : { "type" : "ip" },
                    "location" : { "type" : "geo_point" },
                    "latitude" : { "type" : "float" },
                    "longitude" : { "type" : "float" },
                    "timezone" : { "type" : "keyword" }
                }
            },
            "message" : { "type" : "text" },
            "tags" : { "type" : "keyword"}
        }
    }
}

Since the template is only applied on the index creation, if the index already exists it is necessary to delete it with the request DELETE endpoints

After we create the template we can add the output to elasticsearch in the pipeline.

output {
    elasticsearch {
        hosts => ["http://elk:9200"]
        index => "endpoints"
    }
}

visualizing the data and creating maps

As soon as we start logstash with the configured pipeline, the data from the API will start to be collected and sent to elasticsearch, after we create an index pattern we can visualize the data with the geolocation in kibana.

While our original event only had the information about the destination IP address, using the geoip filter allowed us to enrich the documents with more information about the destination IP address.

The field geo.country_name is an example of data enrichment by the geoip filter with the option default_database_type set as City and the field geo.as_org is an example of data enrichment by the same filter when we set the default_database_type to ASN, this is the reason we used two geoip filters in the pipeline.

We can also plot the data using the Maps tool in kibana.

If before our monitoring only had the information about the destination IP, now we can have an idea of where this destination is and we can visualize it in a map.

An important thing that we should consider is that the accuracy of the coordinates obtained with the geoip filter are not exact and can vary according with the geographic location, being more accurate in some countries, the type of the connection, if an IP is from a mobile connection or not, and the zoom level used in the map.

This link from maxmind page, allows us to check the accuracy in each case.

The result of the geoip filter should never be considered exact and may not correspond to reality.

For more information about the geoip filter you can check the official documentation from elastic.

logstash: improving performance by replacing grok with dissect.

Leandro Pereira — Sat, 26 Sep 2020 22:14:20 +0000

Introduction

It has been a while since I've been using the elastic stack to collect and store logs from applications, services and devices that I need to monitor. Since Logstash is responsible for receiving, parsing and publishing those logs, it is extremely important that we guarantee that it has a good performance.

A lot of the performance problems that I've had were caused by configuration errors, implementation or incorrect usage of some filters, mainly the grok filter.

While the grok filter is extremely useful, it is a filter based on regular expressions and it needs to validate every expression, depending on how many events per second logstash is receiving, those validations could lead to a increase load on the CPU which could affect the entire monitoring process.

Grok or Dissect?

One simple way to improve the performance of a pipeline is to check if it is possible to use another filter instead of grok. A filter that I've been using as a replacement for grok is the dissect filter.

The main difference between grok and dissect is that dissect does not use regular expressions to parse the message, the fields are defined by its position, everything between a %{ and a } is seen as a field and everything else is seen as a delimiter, this makes dissect faster and lighter than grok.

Considering the following example message:

2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message

We can parse that message into the following fields:

timestamp, ipaddr, thread, logger, loglevel, msg

To parse that message with grok and dissect, we use the following configuration.

grok

grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{IP:ipaddr} \[%{INT:thread}\] %{DATA:logger} %{WORD:loglevel} - %{GREEDYDATA:msg}"}
}

dissect

dissect {
    mapping => {
        "message" => "%{timestamp} %{ipaddr} [%{thread}] %{logger} %{loglevel} - %{msg}"
    }
}

We can see that the structure is very similar, but with grok we need to specify the regex that we will use to validate the field, for example TIMESTAMP_ISO8601 for the field timestamp or IP for the field ipaddr, this is something that we do not need to do when using dissect, because it will use what is on that position as a value for the field, this is one of the reasons that will make us choose between the two of them.

If the collected messsages in a pipeline always have the same structure, or even if they have small variations, using dissect instead of grok is possible and it is an advantage since the parsing will be faster and will need less processing power.

But how faster and how less processing power it will need?

Grok vs Dissect

To compare the performance of the two filters, I've used a simple pipeline with the generator filter as an input and the stdout filter as an output, I've also used the grok and dissect filters show before on different runs.

input {
    generator {
        lines => ["2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message"]
        count => 10000000
    }
}

filter {
    #
    # grok or dissect filter
    #
    if [sequence] != 0 and [sequence] != 9999999 {
         drop { }
     }
}
output {
    stdout { }
}

This pipeline basically generates 10 million messages, applies the specified filter, which could be grok or dissect, and use drop to drop every message except the first and the last one, which are show just to verify the beginning and ending of the processing.

This pipeline output is the following.

{
          "host" => "elk",
            "ipaddr" => "127.0.0.1",
       "message" => "2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message",
      "sequence" => 0,
      "loglevel" => "INFO",
        "logger" => "program.Logger",
      "@version" => "1",
     "timestamp" => "2020-08-20T21:45:50",
           "msg" => "sample message",
        "thread" => "1",
    "@timestamp" => 2020-08-26T02:00:38.127Z
}
{
          "host" => "elk",
            "ipaddr" => "127.0.0.1",
       "message" => "2020-08-20T21:45:50 127.0.0.1 [1] program.Logger INFO - sample message",
      "sequence" => 9999999,
      "loglevel" => "INFO",
        "logger" => "program.Logger",
      "@version" => "1",
     "timestamp" => "2020-08-20T21:45:50",
           "msg" => "sample message",
        "thread" => "1",
    "@timestamp" => 2020-08-26T02:02:34.018Z
}

Since I was interested in comparing the CPU usage of the machine during the filters execution, I've used nmon running on background and harvesting metrics every 1 second during 5 minutes resulting in 300 measures, which is more than enough for this case.

For the execution of the pipeline I've used a VM with 4 vCPUs and 4 GB of RAM running CentOS 7.8 and Logstash on the version 7.9.

Each measure from nmon has the following format.

ZZZZ,T0010,00:16:49,21-AUG-2020
CPU001,T0010,50.5,2.0,0.0,46.5,1.0
CPU002,T0010,99.0,0.0,0.0,1.0,0.0
CPU003,T0010,97.0,1.0,0.0,1.0,1.0
CPU004,T0010,96.0,1.0,0.0,3.0,0.0
CPU_ALL,T0010,86.4,0.8,0.0,12.8,0.0,,4
MEM,T0010,3789.0,-0.0,-0.0,1024.0,2903.2,-0.0,-0.0,1024.0,-0.0,281.9,577.3,-1.0,2.0,0.0,181.5
VM,T0010,30,0,0,2384,12760,-1,0,0,0,0,27,0,0,2541,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0
PROC,T0010,5,0,2524.7,-1.0,-1.0,-1.0,1.0,-1.0,-1.0,-1.0
NET,T0010,2.6,0.1,2.8,0.1
NETPACKET,T0010,10.9,2.0,12.9,2.0
JFSFILE,T0010,59.0,0.0,0.5,59.0,24.3,4.5,4.5,4.5
DISKBUSY,T0010,0.0,0.0,0.0,0.0
DISKREAD,T0010,0.0,0.0,0.0,0.0
DISKWRITE,T0010,0.0,0.0,0.0,0.0
DISKXFER,T0010,0.0,0.0,0.0,0.0
DISKBSIZE,T0010,0.0,0.0,0.0,0.0

From all those lines, just the line with CPU_ALL matters, more specifically the third column, which has the average usage percent of all CPUs at the time of the harvesting.

After some data manipulation of the harvested data during the execution of the pipeline with grok and after that, with dissect, we can visualize and compare the CPU usage and the execution time.

The initial usage spike that we see in the graphic is caused by initialization of Logstash, the folowing plateau corresponds to the usage during the processing of the messages.

Looking at the graphic we can see that the time to process the 10 million messages is basically the same for both filters tested in this specific use case, but we have a big difference in the cpu usage. We have an Average CPU Usage of 40% when using the dissect filter in comparison with an Average CPU Usage of more than 60% when using grok.

Conclusion and Links

Depending on the number of pipelines and events per second of each one of those pipelines, dedicating time to study if it is possible to replace grok by dissect is something that could lead to a huge improvement in the ingestion and analysis of data, besides that, it could also leads to costs cuts when you have your infrastructutre on the cloud since it will make possible to use smaller instances, as also less consumption of cpu credits on some cases.

On the cases where the logs have always the same structure, as web server logs, applicaton logs, firewalls and routers, or when you have control on the format that the logs will use, the dissect filter is the best choice when using Logstash to collect and parse those logs.