ELK Caso práctico

Caso práctico de stack ELK + Metricbeat

El objetivo del presente trabajo es monitorizar los nodos del sistem 24Online para el Banco
Bolivariano. Procederemos a montar un stack ELK (Elasticsearch, Logstash y Kibana) en la
máquina que tengo como servidor y recoger métricas con Metricbeat, HeartBeat y FileBeat
de cada nodo.

Por tanto, para empezar el entorno que se tiene es:

 Servidor elk-server (RedHat 7.3, IP 172.17.1.134) con ELK instalado. En el propio

servidor también tengo instalado Metricbeat para tener métricas de rendimiento.
 Máquinas virtualizadasnode1, node2 ....-node<n> con RedHat que envian métricas a
elk-server. Como estas máquinas envían métricas al servidor elk-server las
consideraré clientes.

Preparación del repositorio de elastic.co

Para instalar los distintos componentes tanto en servidor como clientes utilizaremos el
repositorio oficial de elastic.co

En todos los nodos (servidor y clientes) hacemos:

$ curl -L -O
https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-oss-6.4.3-x86_64.rpm
$ sudo rpm -vi metricbeat-oss-6.4.3-x86_64.rpm

Instalación y configuración de ELK en el servidor

En el servidor elk-server instalo el stack ELK:

sudo apt-get update

sudo apt-get install elasticsearch logstash kibana

Configuramos cada componente:

 IP/hostname de Kibana y URL de Elasticsearch en el fichero

/etc/kibana/kibana.yml

①
ELK Caso práctico

server.host: "192.168.1.39"

elasticsearch.url: "http://192.168.1.39:9200"


 IP/hostname de Elasticsearch en /etc/elasticsearch/elasticsearch.yml

network.host: 192.168.1.39


 IP/hostname de Logstash en /etc/logstash/logstash.yml

http.host: "192.168.1.39"

Habilito (para que arranque con el inicio de la máquina) y levanto cada componente:

# Habilitar

systemctl enable kibana

systemctl enable logstash

systemctl enable elasticsearch

# Iniciar

systemctl start kibana

systemctl start logstash

systemctl start elasticsearch

②
 ELK Caso práctico

Instalación de Metricbeat

Servidor

En mi caso he instalado Metricbeat en el servidor ya que me interesa tener métricas del

mismo.

Metricbeat es un agente de envío de código abierto que se utiliza para recopilar y enviar el
sistema operativo y las métricas de servicio a uno o más destinos, incluido Logstash.

Paso 1 - Instalar

deb (Debian / Ubuntu)

$ curl -L -O
https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-oss-6.4.3-amd64.deb
$ sudo dpkg -i metricbeat-oss-6.4.3-amd64.deb

rpm (Redhat / Centos / Fedora)

$ curl -L -O
https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-oss-6.4.3-x86_64.rpm
$ sudo rpm -vi metricbeat-oss-6.4.3-x86_64.rpm

Paso 2 - Localiza el archivo de configuración

deb / rpm /etc/metricbeat/metricbeat.yml

Paso 3 - Configurar los módulos

Configure los datos que desea enviarnos, editando los módulos. Se encuentran ejemplos de
estas configuraciones metricbeat.reference.ymlen la misma carpeta que el archivo de
configuración. El módulo de estado del sistema está habilitado de forma predeterminada para
recopilar métricas sobre su servidor, como el uso de la CPU, el uso de la memoria, las
métricas de E / S de la red y las estadísticas de procesos:

metricbeat.modules:

- module: system

metricsets:

- cpu

- filesystem

- memory

- network

- process

enabled: true

③
 ELK Caso práctico

period: 10s

processes: ['.*']

cpu_ticks: false

Paso 4 - Configurar salida

Enviaremos a Logstash para que tengamos la opción de ejecutar filtros antes de que se indexen
los datos.
Comenta el bloque de salida elasticsearch.

## Comment out elasticsearch output

#output.elasticsearch:

# hosts: ["localhost:9200"]

Descomente y cambie la salida de logstash para que coincida a continuación.

output.logstash:

hosts: ["your-logstash-host:your-port"]

loadbalance: true

ssl.enabled: true

Paso 5 - Validar la configuración

Vamos a comprobar que el archivo de configuración es sintácticamente correcto.
deb / rpm

$ sudo chown -R root /etc/metricbeat

$ sudo chmod go-w /etc/metricbeat/metricbeat.yml

$ sudo metricbeat -e -c /etc/metricbeat/metricbeat.yml

Paso 6 - Iniciar metricbeat

Ok, es hora de empezar a ingerir datos!
deb / rpm

$ sudo service metricbeat start

El fichero de configuración es /etc/metricbeat/metricbeat.yml. Configuramos el endpoint de

Kibana:

#============================== Kibana
=====================================

④
ELK Caso práctico

# Starting with Beats version 6.0.0, the dashboards are loaded via the Kibana API.

# This requires a Kibana endpoint configuration.

setup.kibana:

# Kibana Host

# Scheme and port can be left out and will be set to the default (http and 5601)

# In case you specify and additional path, the scheme is required: http://localhost:5601/path

# IPv6 addresses should always be defined as: https://[2001:db8::1]:5601

host: "192.168.1.39:5601"

...

En mi caso he elegido como endpoint Logstash:

...

#================================ Outputs
=====================================

...

#----------------------------- Logstash output --------------------------------

#output.logstash:

# The Logstash hosts

#hosts: ["192.168.1.39:5044"]

...

Iniciamos Metricbeat:

systemctl start metricbeat.service

Cargamos template para Elasticsearch y dashboards para Kibana:

# Carga de template, veremos "Loaded index template" si todo ha cargado bien

metricbeat setup --template -E output.logstash.enabled=false -E

'output.elasticsearch.hosts=["192.168.1.39:9200"]'

# Carga de dashboards, obtendremos "Loaded dashboards"

metricbeat setup --dashboards

⑤
ELK Caso práctico

Ahora, volvemos al fichero de configuración /etc/metricbeat/metricbeat.yml, comentamos el

endpoint de Elasticsearch y descomentamos el de Logstash. Después, reiniciamos Metricbeat
con systemctl restart metricbeat.service

Clientes wcg-node

Esta parte es muy sencilla ya que que la configuración de ELK para recibir métricas de
Metricbeat ya la hemos realizado en el paso anterior. Ahora en cada uno de los clientes
(wcg-node1,2 y 3) simplemente instalo el paquete para Metricbeat:

sudo apt-get install metricbeat

Y configuro en /etc/metricbeat/metricbeat.yml el endpoint para Kibana y Logstash:

#============================== Kibana
=====================================

# Starting with Beats version 6.0.0, the dashboards are loaded via the Kibana API.

# This requires a Kibana endpoint configuration.

setup.kibana:

# Kibana Host

# Scheme and port can be left out and will be set to the default (http and 5601)

# In case you specify and additional path, the scheme is required: http://localhost:5601/path

# IPv6 addresses should always be defined as: https://[2001:db8::1]:5601

host: "192.168.1.39:5601"

...

#----------------------------- Logstash output --------------------------------

output.logstash:

# The Logstash hosts

hosts: ["192.168.1.39:5044"]

Iniciamos en cada máquina Metricbeat con

systemctl start metricbeat.service

Comprobando dashboards

⑥
ELK Caso práctico

En mi máquina servidor donde tengo el stack ELK, en un navegador voy a

http://192.168.1.39:5601/app/kibana y después en los dashboards busco metric,
seleccionando [Metricbeat System] Overview

Desde aquí podemos entrar al dashboard de los distintos nodos que están enviando métricas,
por ejemplo mi servidor:

⑦
ELK Caso práctico

sudo /opt/logstash/bin/logstash-plugin install logstash-input-beats

To test the configuration:

$ sudo service logstash configtest

Configuration OK

⑧
ELK Caso práctico

logstash.conf

input {

beats {

port => 5044

ssl => true

ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"

ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"

⑨
ELK Caso práctico

output {

elasticsearch {

hosts => ["localhost:9200"]

sniffing => true

manage_template => false

index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"

document_type => "%{[@metadata][type]}"

logstash-beats.conf

input {

beats {

port => 5044

output {

elasticsearch {

hosts => "localhost:9200"

manage_template => false

index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"

logstash-apache.conf

input {

file {

path => "/var/log/logstash/*_log"

⑩
ELK Caso práctico

filter {

if [path] =~ "access" {

mutate { replace => { type => "apache_access" } }

grok {

match => { "message" => "%{COMBINEDAPACHELOG}" }

date {

match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]

} else if [path] =~ "error" {

mutate { replace => { type => "apache_error" } }

} else {

mutate { replace => { type => "random_logs" } }

output {

elasticsearch { hosts => ["localhost:9200"] }

stdout { codec => rubydebug }

logstash-syslog.conf

Input this configuration into the file:

input {

tcp {

port => 5000

type => syslog

filter {

11
ELK Caso práctico

if [type] == "syslog" {

grok {

match => { "message" =>

"%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:
syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }

add_field => [ "received_at", "%{@timestamp}" ]

add_field => [ "received_from", "%{host}" ]

date {

match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]

output {

elasticsearch { hosts => ["localhost:9200"] }

stdout { codec => rubydebug }

demo-metrics-pipeline.conf

Input these configurations:

input {

beats {

port => 5044

output {

elasticsearch {

hosts => "localhost:9200"

manage_template => false

index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"

12
ELK Caso práctico

logstash-simple.conf

Input these configurations:

input { stdin { } }

filter {

grok {

match => { "message" => "%{COMBINEDAPACHELOG}" }

date {

match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]

output {

elasticsearch { hosts => ["localhost:9200"] }

stdout { codec => rubydebug }

13
ELK Caso práctico

Elastic Search, LogStash, Kibana y Beats

¿Qué es la búsqueda elástica?

Elastic Search es una plataforma informática distribuida para buscar texto rápidamente. El
inventor de Elastic Search construyó esto para ayudar a su esposa a buscar recetas de
cocina.

¿Qué es Kibana?

Kibana es una aplicación web para visualizar datos almacenados en Elastic Search.

¿Qué es LogStash?

LogStash es una herramienta de extracción, transformación y carga para búsqueda

elástica. LogStash extrae datos de varias fuentes de datos. Lo transforma en un formato que
Elastic Search entiende y lo almacena en Elastic Search.

¿Qué son los Beats?

Beats son servicios livianos instalados en servidores para la recolección de datos.

Ejemplo

En la reunión, el orador repasó un ejemplo que voy a parafrasear a continuación:

Digamos que tengo una máquina que ejecuta un servidor web apache y que quiero recopilar
todos los errores del registro de apache. Además, quiero recopilar el uso de la CPU, el
espacio en disco y las métricas de uso de la memoria.

Para recopilar datos, instalo dos Beats en mi máquina. The File Beat para recopilar datos de
registro. El tiempo de la métrica para recopilar métricas como el uso de la CPU, el espacio en
disco y el uso de la memoria.

El tiempo del archivo verificará periódicamente los cambios en el registro de apache. Cuando
si encuentra cambios, enviará el registro a LogStash. LogStash ejecutará un filtro en él para
encontrar todos los errores y luego almacenará los resultados en Elastic Search.

14
ELK Caso práctico

Cuando se ejecute el compás de las métricas, recopilará las métricas y las almacenará
directamente en Elastic Search. Dado que los datos métricos son simples, el tiempo métrico
no tiene que enviar los datos a LogStash para transformarlos.

Para visualizar los datos podemos usar Kibana para crear cuadros de mando. Ya existe un
panel de control fijo para los datos de métricas. Aquí hay un ejemplo

Genial, ¿cómo puedo usarlo?

Se siente como una de las formas en que podría usar la pila ELK más los latidos para el
monitoreo basado en eventos de su infraestructura. Aparte de eso, no estoy realmente seguro,
pero estoy interesado en explorar esto más.
¿Para qué no debería usarlo?

La búsqueda elástica no sigue los principios de ACID que siguen las bases de datos RDMS,
por lo que no me atrevería a usar esto como fuente de datos principal para una aplicación.

El Elastic Stack intenta que sea lo más fácil posible ingerir datos en Elasticsearch. Filebeat es
una gran herramienta para rastrear archivos y viene con un conjunto de módulos que permite
ingerir una amplia gama de formatos de registro comunes con una configuración mínima. Si la
información que está buscando ingerir no está cubierta por estos módulos, los nodos de
ingesta Logstash y Elasticsearch brindan una manera flexible y poderosa de analizar y
procesar la mayoría de los tipos de datos basados en texto.

En esta publicación del blog, proporcionaremos una breve introducción a Logstash y le

mostraremos cómo trabajar con ella cuando desarrolle una configuración para analizar
algunos registros de acceso de caché de Squid de muestra y cómo estos se ingieren en
Elasticsearch.
Un breve resumen de Logstash

Logstash es un motor de procesamiento y recopilación de datos basado en

complementos. Viene con una amplia gama de complementos que hace que sea fácil de
configurar para recopilar, procesar y reenviar datos en diferentes arquitecturas.

15
ELK Caso práctico

El procesamiento se organiza en una o más tuberías. En cada canalización, uno o

más complementos de entrada reciben o recopilan datos que luego se colocan en una cola
interna. Esto es por defecto pequeño y se mantiene en la memoria, pero se puede configurar
para que sea más grande y se mantenga en el disco para mejorar la confiabilidad y la
capacidad de recuperación.

Los subprocesos de procesamiento leen los datos de la cola en micro lotes y los procesan a
través de cualquier complemento de filtro configurado en secuencia. Logstash
out-of-the-box viene con una gran cantidad de complementos que se dirigen a tipos
específicos de procesamiento, y es así como los datos se analizan, procesan y enriquecen.

Una vez que se han procesado los datos, los subprocesos de procesamiento envían los datos
a los complementos de salida apropiados, que son responsables de formatear y enviar los
datos en adelante, por ejemplo, a Elasticsearch.

Los complementos de entrada y salida también pueden tener un complemento de

códecconfigurado. Esto permite analizar y / o formatear los datos antes de que se coloquen
en la cola interna o se envíen a un complemento de salida.

Instalando Logstash y Elasticsearch

Con el fin de ejecutar los ejemplos de esta entrada del blog, primero tendrá que instalar
Logstash y Elasticsearch . Siga los enlaces para obtener instrucciones sobre cómo hacer
esto para su sistema operativo. Usaremos la versión 6.2.4 de Elastic Stack.

Especificando tuberías

Las tuberías de logstash se crean en función de uno o más archivos de configuración. Antes
de comenzar, lo guiaremos rápidamente a través de las diferentes opciones disponibles. Los
directorios descritos en esta sección pueden variar según el modo de instalación y el sistema
operativo, y se definen en la documentación .

Pipeline único utilizando un solo archivo de configuración

16
ELK Caso práctico

La forma más sencilla de iniciar Logstash, y la forma en que lo usaremos en esta publicación
del blog, es hacer que Logstash cree una canalización única basada en un único archivo de
configuración que especificamos a través del parámetro de línea de comandos -f .

Canalización única utilizando múltiples archivos de configuración.

Logstash también se puede configurar para usar todos los archivos en un directorio específico
como archivos de configuración. Esto se puede establecer a través del archivo logstash.yml o
pasando una ruta de directorio a través de la línea de comandos usando el -fparámetro de
línea de comandos. Este es el valor predeterminado si instala Logstash como un servicio.

Cuando se proporciona un directorio, todos los archivos en ese directorio se concatenarán en

orden lexicográfico y luego se analizarán como un único archivo de configuración. Por lo tanto,
los datos de todas las entradas serán procesados por todos los filtros y enviados a todas las
salidas a menos que usted controle el flujo usando condicionales .

Usando múltiples tuberías

Para poder utilizar varias tuberías en Logstash, deberá editar los archivos pipelines.yml que
vienen con Logstash. Esto se puede encontrar en el directorio de configuración y contiene
archivos de configuración y parámetros de configuración para todas las tuberías admitidas por
esa instancia de Logstash.

El uso de múltiples conductos le permite separar diferentes flujos lógicos, lo que puede reducir
en gran medida la complejidad y la cantidad de condicionales utilizados. Esto hace que la
configuración sea más fácil de ajustar y mantener. A medida que los datos que fluyen a través
de una tubería al mismo tiempo se vuelven más homogéneos, también pueden generar
grandes ganancias de rendimiento, ya que los complementos de salida se pueden usar de
manera más eficiente.

Creando una primera configuración

Cualquier configuración de Logstash debe contener al menos un complemento de entrada y

un complemento de salida. Los filtros son opcionales. Como primer ejemplo de cómo puede
verse un archivo de configuración simple, comenzaremos con uno que lea un conjunto de
datos de prueba de un archivo y los envíe a la consola de forma estructurada. Esta es una
configuración que es muy útil cuando se desarrolla una configuración, ya que le permite iterar
y construir rápidamente la configuración. A lo largo de esta publicación del blog asumiremos
que se llama nuestro archivo de configuración test.confy que se almacena en
el “/home/logstash”directorio junto con el archivo que contiene nuestros datos de prueba:

input{

file {

path => [ "/home/logstash/testdata.log" ]

sincedb_path => "/ dev / null"

17
ELK Caso práctico

start_position => " begin "

filter {

output {

stdout {

codec => rubydebug

Aquí vemos los tres grupos de nivel superior que forman parte de todas las configuraciones
Logstash: input, filter, y output. En la sección de entrada, hemos especificado
un complemento de entrada de archivo y hemos proporcionado la ruta de acceso de
nuestro archivo de datos de prueba a través de la pathdirectiva. Establecemos
la start_positiondirectiva en “beginning”para indicar al complemento que lea el archivo desde
el principio cada vez que se descubra un nuevo archivo.

Para mantener un registro de qué datos dentro de cada archivo de entrada se han procesado,
el complemento de entrada del archivo Logstash utiliza un archivo llamado sincedb para
registrar la posición actual. Como nuestra configuración se utiliza para el desarrollo,
queremos que el archivo se vuelva a leer repetidamente y, por lo tanto, deshabilitamos el uso
del archivo sincedb. Hacemos esto configurando la sincedb_pathdirectiva “/dev/null”en
sistemas basados en Linux. En Windows, en su lugar, se establecería en “nul”.

Aunque el complemento de entrada de archivo Logstash es una excelente manera de

comenzar a desarrollar configuraciones, Filebeat es el producto recomendado para la
recopilación de registros y el envío desde servidores host . Filebeat puede enviar los
registros a Logstash , y Logstash puede recibir y procesar estos registros con la entrada
Beats . La lógica de análisis descrita en esta publicación de blog aún es aplicable en
cualquier escenario, pero Filebeat está más optimizado para el rendimiento y requiere un
menor uso de recursos, lo que es ideal para ejecutar como agente.

El stdoutcomplemento de salida escribirá los datos en la consola y el rubydebugcódec ayuda

a mostrar la estructura, lo que simplifica la depuración durante el desarrollo de la
configuración.

Iniciando Logstash

Para verificar que Logstash y nuestro archivo de configuración funcionan, creamos un archivo
nombrado “testdata.log”en el “/home/logstash”directorio. Esto contiene la cadena "Hello
Logstash!" Seguido de una nueva línea.

18
ELK Caso práctico

Suponiendo que tengamos el binario logstash en nuestra ruta, podemos iniciar Logstash
usando el siguiente comando:

logstash - r - f “/ home/logstash/test.conf ”

Además del -fparámetro de línea de comando que discutimos anteriormente, también usamos
la -rbandera. Esto le dice a Logstash que recargue automáticamente la configuración siempre
que identifique que la configuración ha cambiado. Esto es muy útil, especialmente durante el
desarrollo. A medida que deshabilitamos el archivo sincedb, el archivo de entrada se volverá
a leer cada vez que se vuelva a cargar la configuración, lo que nos permite probar
rápidamente la configuración a medida que continuamos desarrollándola.

La salida a la consola desde Logstash muestra algunos registros relacionados con su

inicio. Después de que el archivo será procesado y verá algo como esto:

"message" => "¡Hola Logstash!" ,

"@version" => "1" ,

"ruta" => "/home/logstash/testdata.log" ,

"@timestamp" => 2018 - 04 - 24T12 : 40 : 09.105Z ,

"host" => " localhost "

Este es el evento que Logstash ha procesado. Puede ver que los datos se almacenan en el
campo del mensaje y que Logstash ha agregado algunos metadatos para el evento en la
forma de la marca de tiempo que se procesó y donde se originó.

Todo esto es muy bueno y demuestra que la mecánica está funcionando. Ahora agregaremos
algunos datos de prueba más realistas y mostraremos cómo se puede analizar esto.

¿Cómo puedo analizar mis registros?

A veces hay un filtro perfecto que puede usarse para analizar sus datos, por ejemplo,
el archivo json filte r en caso de que sus registros estén en formato JSON. Muchas veces,
sin embargo, necesitamos analizar registros en diferentes tipos de formatos de texto. El
ejemplo que usaremos en esta publicación de blog son algunas líneas de registros de acceso
de caché de Squid que se muestran a continuación:

1524206424.034 19395 207.96 . 0.0 TCP_MISS / 304 15363 GET http :

//elastic.co/android-chrome-192x192.gif - DIRECT / 10.0.5.120 - 1524206424.145 106 207.96 .
0.0 TCP_HIT / 200 68247 GET http : //elastic.co/guide/en/logstash/current/images/logstash.gif
- NINGUNO / - image / gif

19
ELK Caso práctico

Cada línea contiene información sobre una solicitud al caché de Squid y se puede dividir en
varios campos distintos que debemos analizar.

Cuando se analizan los registros de texto, hay dos filtros que se utilizan comúnmente: analizar
los registros de acuerdo con los delimitadores, mientras que el trabajo de grok se basa en la
coincidencia de expresiones regulares.

El filtro de disección funciona muy bien cuando la estructura de los datos está bien definida y
puede ser muy rápida y eficiente. También suele ser más fácil comenzar, especialmente para
los usuarios que no están familiarizados con las expresiones regulares.

Grok es generalmente más poderoso y puede manejar una mayor variedad de datos. Sin
embargo, la coincidencia de expresiones regulares puede usar más recursos y ser más lenta,
especialmente si no se optimiza correctamente.

Antes de continuar con el análisis, reemplazamos el contenido del testdata.logarchivo con

estas dos líneas de registro y nos aseguramos de que cada línea vaya seguida de una nueva
línea.

Análisis de registros con disección

Cuando trabaje con el filtro de disección, especifique una secuencia de campos para extraer,
así como los delimitadores entre estos campos. El filtro realiza una sola pasada sobre los
datos y compara los delimitadores en el patrón. Al mismo tiempo, los datos entre los
delimitadores se asignan a los campos especificados. El filtro no valida el formato de los datos
que se extraen.

Los separadores utilizados al analizar estos datos utilizando el filtro de disección se resaltan
en color rosa a continuación.

El primer campo contiene la marca de tiempo y es seguido por uno o más espacios,
dependiendo de la longitud del siguiente campo de duración. Podemos especificar el campo
de marca de tiempo como% {marca de tiempo}, pero para hacer que acepte un número
variable de espacios como separador, debemos agregar un sufijo -> al campo. Todos los
demás separadores en la entrada del registro consisten en un solo carácter. Por lo tanto,
podemos comenzar a construir el patrón, dando como resultado la siguiente sección de filtro:

"@version" => "1",

20
ELK Caso práctico

"mensaje" => "1524206424.034 19395 207.96.0.0 TCP_MISS / 304 15363 GET

http://elastic.co/android-chrome-192x192.gif - DIRECT / 10.0.5.120 -" ,

"@timestamp" => 2018-04-24T12: 42: 23.003Z,

"ruta" => "/home/logstash/testdata.log",

"host" => "localhost",

"duration" => "19395 ",

"timestamp" => "1524206424.034",

"rest" => "TCP_MISS/304 15363Read More GET

http://elastic.co/android-chrome-192x192.gif - DIRECT/10.0.5.120 -",

"client_ip" => "207.96.0.0"

Ahora podemos continuar construyendo el patrón paso a paso. Una vez que hayamos
analizado con éxito todos los campos, podemos eliminar el campo del mensaje para que no
tengamos los mismos datos dos veces. Podemos hacer esto usando la directiva remove_field,
que solo se ejecuta si el análisis fue exitoso, resultando en el siguiente bloque de filtro.

filter {

dissect {

mapping => { "message" => "% {timestamp->}% {duration}% {client_address}%

{cache_result} /% {status_code}% {bytes}% {request_method}% {url}% {user}%
{hierarchy_code} /% {server}% {content_type} " }

remove_field => [“ mensaje ”]

Cuando se ejecuta contra los datos de muestra, el primer registro tiene el siguiente aspecto:

"user" => "-",

"content_type" => "-",

"host" => "localhost",

"cache_result" => "TCP_MISS",

"@timestamp" => 2018-04-24T12: 43 : 07.406Z,

"duration" => "19395",

"request_method" => "GET",

21
ELK Caso práctico

"url" => "http://elastic.co/android-chrome-192x192.gif",

"timestamp" => "1524206424.034",

"status_code" => "304",

"server" => "10.0.5.120Read More",

"@version" => "1",

"client_address" => "207.96.0.0",

"bytes" => "15363",

"path" => "/home/logstash/testdata.log",

"hierarchy_code" => "DIRECT"

La documentación contiene algunos buenos ejemplos, y esta publicación de

blogproporciona una buena discusión sobre el diseño y el propósito del filtro.

Eso fue bastante fácil, ¿verdad? Vamos a procesar esto un poco más adelante, pero primero
veamos cómo podemos hacer lo mismo con grok.

¿Cómo trabajo mejor con grok?

Grok usa patrones de expresión regular para hacer coincidir ambos campos y
delimitadores. La siguiente figura muestra los campos que se capturarán en azul y los
delimitadores en rojo.

Grok comenzará a hacer coincidir los patrones configurados desde el inicio y continuará hasta
que se haya asignado el evento completo o se haya determinado que no se puede encontrar
una coincidencia. Dependiendo de los tipos de patrones utilizados, esto puede requerir que el
grok procese partes de los datos varias veces.

Grok viene con una amplia gama de patrones confeccionados. Algunos de los más genéricos
se pueden encontrar aquí , pero este repositorio también contiene un gran número de
patrones bastante especializados para tipos de datos comunes. En realidad, hay uno
para analizar los registros de acceso de Squid , pero en lugar de utilizarlo directamente, en
este tutorial mostraremos cómo construirlo desde cero. Sin embargo, muestra que puede
valer la pena revisar en este repositorio si ya existe algún patrón adecuado, antes de
comenzar a crear un patrón personalizado.

Al crear una configuración de grok, hay una serie de patrones estándar que se utilizan
comúnmente:

 WORD- patrón que coincide con una sola palabra

22
ELK Caso práctico

 NUMBER: patrón que coincide con un número entero o flotante positivo o negativo
 POSINT - patrón que coincide con un entero positivo
 IP: patrón que coincide con una dirección IPv4 o IPv6
 NOTSPACE - patrón que coincide con cualquier cosa que no sea un espacio
 ESPACE - patrón que coincide con cualquier número de espacios consecutivos
 DATA: patrón que coincide con una cantidad limitada de cualquier tipo de datos
 GREEDYDATA - patrón que coincide con todos los datos restantes
Estos son los patrones que usaremos cuando construyamos nuestra configuración de filtro de
grok. La forma de crear configuraciones de grok es generalmente comenzar desde la
izquierda y construir gradualmente el patrón, que captura el resto de los datos utilizando un
patrón GREEDYDATA. Podemos comenzar utilizando el siguiente patrón y bloque de filtro:

23
ELK Caso práctico

filter {

grok {

match => { "message" => "% {NUMBER: timestamp}% {SPACE}% {GREEDYDATA:

rest}"

Este patrón le indica a grok que busque un número al comienzo de la cadena y lo almacene
en un campo llamado timestamp. Después de eso, coincide con una cantidad de espacios
antes de almacenar el resto de los datos en un campo llamado rest. Cuando cambiamos el
bloque de filtro de disección para este, el primer registro sale como sigue:

"timestamp" => "1524206424.034",

"rest" => "19395 207.96.0.0 TCP_MISS / 304 15363 GET

http://elastic.co/android-chrome-192x192.gif - DIRECT / 10.0.5.120 -",

" ruta "=>" /home/logstash/testdata.log ",

" message "=>" 1524206424.034 19395 207.96.0.0 TCP_MISS / 304 15363 GET
http://elastic.co/android-chrome-192x192.gif - DIRECT/10.0.5.120Read More -",
"@timestamp" => 2018-04-24T12:45:11.026Z,

"@version" => "1",

"host" => "localhost"

Usando el depurador de grok

Aunque podemos desarrollar el patrón completo de esta manera, hay una herramienta en
Kibana que puede ayudar a simplificar la creación del patrón grok: el Depurador Grok. En el
video a continuación, mostramos cómo podemos usar esto para construir los patrones para
los registros de ejemplo utilizados en esta publicación de blog.

Una vez que se haya construido la configuración, podemos eliminar el campo del mensaje
una vez que el análisis haya tenido éxito, lo que hará que el bloque de filtro tenga este
aspecto:

filter {

grok {

24
ELK Caso práctico

match => { "message" => "% {NUMBER: timestamp}% {SPACE}% {NUMBER: duration} \
s% {IP: client_address} \ s% {WORD: cache_result} /% {POSINT : status_code} \ s%
{NUMBER: bytes} \ s% {WORD: request_method} \ s% {NOTSPACE: url} \ s% {NOTSPACE:
user} \ s% {WORD: hierarchy_code} /% {NOTSPACE: server} \ s% {NOTSPACE:
content_type} " }

remove_field => [ " message " ]

Esto es similar al patrón ya hecho, pero no idéntico. Cuando se ejecuta con los datos de
muestra, el primer registro se analiza de la misma manera que cuando usamos el filtro de
disección:

25
ELK Caso práctico

"request_method" => "GET",

"cache_result" => "TCP_MISS",

"@timestamp" => 2018-04-24T12: 48: 15.123Z,

"timestamp" => "1524206424.034",

"usuario" => "-",

"bytes" => "15363",

"ruta" => "/home/logstash/testdata.log",

"hierarchy_code" => "DIRECT",

"duration" => "19395",

"client_address" => "207.96.0.0",

"@version" => "1",

"status_code" => "304",

"url" => "httpRead More://elastic.co/android-chrome-192x192.gif",

"content_type" => "-",

"host" => "localhost",

"server" => "10.0.5.120"

Tuning grok para el rendimiento

Grok es una herramienta muy potente y flexible para analizar datos, pero el uso ineficiente de
los patrones puede dar como resultado un rendimiento inferior al esperado. Por lo tanto, le
recomendamos que lea esta publicación del blog sobre el ajuste de rendimiento de
Grok antes de comenzar a utilizar Grok en serio.

Asegurar que los campos son del tipo correcto

Como ha visto en los ejemplos anteriores, todos los campos se han analizado como campos
de cadena. Antes de enviar esta opción para Elasticsearch en forma de documentos JSON,
nos gustaría cambiar el bytes, durationy status_codelos campos de números enteros
y timestampflotar.

Una forma de hacerlo sería utilizar un archivo mutate r y su convertopción.

26
ELK Caso práctico

mutate {

convert => {

"bytes" => "integer"

"duration" => "integer"

"status_code" => "integer"

"timestamp" => "float"

También podemos lograr esto directamente en los filtros de disección y grok. En el filtro de
disección, lo haríamos a través de la directiva convert_datatype.

27
ELK Caso práctico

filter {

dissect {

mapping => {

"message" => "% {timestamp->}% {duration}% {client_address}% {cache_result} /%

{status_code}% {bytes}% {request_method}% {url}% {user}% {hierarchy_code} /% {server}%
{content_type} "

remove_field => [" message "]

convert_datatype => {

" bytes "=>" int "

" duration "=>" int "

" status_code "= > "int”

“timestampRead More”=> “float”

Al usar grok, puede especificar el tipo directamente después del nombre del campo en el
patrón.

28
ELK Caso práctico

filter {

grok {

match => {

"message" =>
"%{NUMBER:timestamp:float}%{SPACE}%{NUMBER:duration:int}\s%{IP:client_address}\s%{
WORD:cache_result}/%{POSINT:status_code:int}\s%{NUMBER:bytes:int}\s%{WORD:request
_method}\s%{NOTSPACE:url}\s%{NOTSPACE:user}\s%{WORD:hierarchy_code}/%{NOTSP
ACE:server}\s%{NOTSPACE:content_typeRead More}"

remove_field => ["message"]

Usando el filtro de fecha

La marca de tiempo extraída del registro es en segundos y milisegundos desde

la época . Nos gustaría tomar esto y convertirlo a un formato de marca de tiempo estándar
que podamos almacenar en el @timestampcampo. Para esto utilizaremos el complemento
de filtro de fecha junto con el UNIXpatrón, que coincide con los datos que tenemos.

date {

match => [ "timestamp" , "UNIX" ] }

Todas las marcas de tiempo estándar almacenadas en Elasticsearch están en zona horaria
UTC. Como esto también se aplica a nuestra marca de tiempo extraída, no necesitamos
especificar ninguna zona horaria. Si tuviera una marca de tiempo en otro formato, podría
especificar este formato en lugar del UNIXpatrón predefinido .

Una vez que hayamos agregado esto y las conversiones de tipo a nuestra configuración, el
primer evento tendrá este aspecto:

"duration" => 19395,

"host" => "localhost",

"@timestamp" => 2018-04-20T06:40:24.034Z,

"bytes" => 15363,

"user" => "-",

"path" => "/home/logstash/testdata.log",

29
ELK Caso práctico

"content_type" => "-",

"@version" => "1",

"url" => "http://elastic.co/android-chrome-192x192.gif",

"server" => "10.0.5.120",

"client_address" => "207.96.0.0",

"timestamp" => 1524206424Read More.034,

"status_code" => 304,

"cache_result" => "TCP_MISS",

"request_method" => "GET",

"hierarchy_code" => "DIRECT"

Parece que ahora tenemos el formato que queremos y estamos listos para comenzar a enviar
los datos a Elasticsearch.

¿Cómo envío datos a Elasticsearch?

Antes de comenzar a enviar datos a Elasticsearch utilizando el complemento de salida

elasticsearch , debemos analizar la función de las asignaciones y cómo éstas difieren de los
tipos a los que puede convertir en Logstash.

Elasticsearch tiene la capacidad de detectar automáticamente cadenas y campos numéricos,

y la asignación seleccionada se basará en el primer documento con el que se encuentre un
nuevo campo. Dependiendo de la apariencia de los datos, esto puede o no dar la asignación
correcta, por ejemplo, si tenemos un campo que generalmente es un flotante pero puede ser
'0' para algunos registros. Es posible que esto se pueda asignar como un entero en lugar de
flotante, dependiendo de qué documento se procesó primero.

Elasticsearch también tiene la capacidad de detectar automáticamente los campos de fecha,

siempre y cuando estén en el formato estándar que produce el filtro de fecha.

Otros tipos de campos, por ejemplo, geo_point e ip , no pueden detectarse automáticamente

y deben definirse explícitamente a través de una plantilla de índice . Las plantillas de índice
se pueden administrar directamente en Elasticsearch a través de una API, pero también es
posible tener Logstash para asegurar que la plantilla adecuada se cargue a través del
complemento de salida de elasticsearch.

Para nuestros datos, generalmente estamos contentos con las asignaciones

predeterminadas. El servercampo puede contener un guión o una dirección IP válida, por lo
que no lo mapearemos como un campo IP. Un campo que requiere una asignación manual es
el client_addresscampo, que queremos que sea del tipo ip. También tenemos algunos
campos de cadena que deseamos poder agregar, pero no es necesario realizar una
búsqueda de texto libre. Los mapearemos explícitamente como campos de palabras

30
ELK Caso práctico

clave . Estos campos son: user, path, content_type, cache_result, request_method, server,
y hierarchy_code.

Tenemos la intención de almacenar nuestros datos en índices basados en el tiempo a

partir del prefijo squid-. Supondremos que Elasticsearch se está ejecutando con la
configuración predeterminada en el mismo host que Logstash para este ejemplo.

Luego podemos crear la siguiente plantilla almacenada en un archivo

llamado squid_mapping.json:

"index_patterns": ["squid-*"],

"mappings": {

"doc": {

"properties": {

"client_address": { "type": "ip" },

"user": { "type": "keyword" },

"path": { "type": "keyword" },

"content_type": { "type": "keyword" },

"cache_result": { "type": "keyword" },

"request_method": { "type": "keyword" },

"server": { "type": "keyword" },

"hierarchy_code": { "typeRead More": "keyword" }

Esta plantilla está configurada para aplicarse a todos los índices que coincidan con el patrón
de índice squid-*. Para el tipo de documento doc(predeterminado en Elasticsearch 6.x),
especifica el mapeo para que el client_addresscampo sea ip y los otros campos especificados
para ser mapeados a keyword.

Podríamos subir esto directamente a Elasticsearch , pero en su lugar mostraremos cómo

configurar el complemento de salida de elasticsearch para manejarlo. En la sección de salida
de nuestra configuración de Logstash, agregaremos un bloque como este:

elasticsearch {

31
ELK Caso práctico

hosts => [ "localhost: 9200" ]

index => "squid -% {+ YYYY.MM.dd}"

manage_template => true template => "/home/logstash/squid_mapping.json"

template_name => "squid_template " }

Si ahora ejecutamos esta configuración e indexamos los documentos de muestra en

Elasticsearch, terminamos con lo siguiente cuando recuperamos las asignaciones resultantes
para el índice a través de la API de obtención de asignaciones :rough the get mapping
API:

32
ELK Caso práctico

"squid-2018.04.20" : {

"mappings" : {

"doc" : {

"properties" : {

"@timestamp" : {

"type" : "date"

},

"@version" : {

"type" : "text",

"fields" : {

"keyword" : {

"type" : "keyword",

"ignore_above" : 256

},

"bytes" : {

"type" : "long"

},

"cache_result" : {

"type" : "keyword"

},

"type" : "ip"

},

"content_type" : {

"typeRead More" : "keyword"

},

33
ELK Caso práctico

"duration" : {

"type" : "long"

},

"hierarchy_code" : {

"type" : "keyword"

},

"host" : {

"type" : "text",

"fields" : {

"keyword" : {

"type" : "keyword",

"ignore_above" : 256

},

"path" : {

"type" : "keyword"

},

"request_method" : {

"type" : "keyword"

},

"server" : {

"type" : "keyword"

},

"status_code" : {

"type" : "long"

},

"timestamp" : {

"type" : "float"

},

34
ELK Caso práctico

"url" : {

"type" : "text",

"fields" : {

"keyword" : {

"type" : "keyword",

"ignore_above" : 256

},

"user" : {

"type" : "keyword"

Podemos ver que nuestra plantilla se ha aplicado y que los campos que especificamos están
asignados correctamente. Como el enfoque principal de esta publicación del blog es Logstash,
solo hemos arañado la superficie de cómo funcionan las asignaciones. Puede leer más sobre
este tema esencial en la documentación .

35
ELK Caso práctico

Conclusiones

En esta publicación del blog, hemos mostrado cómo trabajar mejor con Logstash mientras
desarrollamos una configuración personalizada y de muestra, y nos aseguramos de que se
haya escrito correctamente en Elasticsearch. Sin embargo, esto solo ha arañado la superficie
de lo que puede lograr con Logstash. Eche un vistazo a los documentos y publicaciones de
blog a los que hemos vinculado a través de esta publicación de blog, pero también revise
la guía oficial de introducción y vea todos los complementos de entrada , salida y filtro
disponibles. Una vez que comprenda lo que está disponible, Logstash se convierte
rápidamente en la navaja suiza del procesamiento de datos.

Si tiene problemas o tiene preguntas adicionales, siempre puede comunicarse con nosotros
en la categoría Logstash en nuestro foro de discusión . Si desea ver otros datos de muestra y
confusiones de logstash, hay más ejemplos en https://github.com/elastic/examples/ .

36
ELK Caso práctico

Hertbeat
Paso 1: Instalar Heartbeat
A diferencia de la mayoría de los Beats, que instala en los nodos perimetrales, normalmente
instala Heartbeat como parte del servicio de monitoreo que se ejecuta en una máquina
separada y posiblemente incluso fuera de la red donde se ejecutan los servicios que desea
monitorear.
Antes de comenzar : si no ha instalado Elastic Stack, hágalo ahora. Consulte Cómo empezar
con la pila elástica .

Para descargar e instalar Heartbeat, use los comandos que funcionan con su sistema.

Si usa Apt o Yum, puede instalar Heartbeat desde nuestros repositorios para
actualizar a la versión más reciente con mayor facilidad.

Vea nuestra página de descarga para otras opciones de instalación, como

imágenes de 32 bits.

curl -L -O https://artifacts.elastic.co/downloads/beats/heartbeat/heartbeat-7.0.0-x86_64.rpm
sudo rpm -vi heartbeat-7.0.0-x86_64.rpm

Configurando y ejecutando Heartbeat

Si comenzó con la configuración predeterminada, Heartbeat está configurado para hacer ping
a una instancia local de Elasticsearch y exportar métricas a la misma instancia local. Para
cambiar esta configuración y ajustar Heartbeat según sus necesidades, deberá abrir el
archivo de configuración de Heartbeat (ubicado en las instalaciones de Deb
en: /etc/heartbeat/heartbeat.yml).

Si ha usado alguno de los otros Beats, se encontrará familiarizado con la estructura del
archivo.

La sección más importante del archivo es la sección Monitores. Un "monitor" le dice a

Heartbeat qué URL debe monitorear individualmente y qué regla de programación debe usar.

Como ejemplo simple, agreguemos a la configuración existente nuestra instancia local

de Kibana como un servicio que queremos monitorear. La configuración ahora se vería como
sigue:

heartbeat.monitors:

- type: http

urls: ["http://localhost:9200"]

37
ELK Caso práctico

schedule: '@every 10s'

timeout: 16s

- type: http

urls: ["http://localhost:5601"]

schedule: '*/5 * * * * * *'

La programación se basa en una sintaxis similar a cron, y el tipo de monitor puede ser HTTP,
TCP o ICMP. Cada uno de los monitores tiene un caso de uso diferente: lo más probable es
que use el tipo ICMP, por ejemplo, para una simple comprobación de ping de si un servicio
está disponible o no y el tipo HTTP para conectarse a través de HTTP.

Las otras secciones en el archivo de configuración son similares a las otras Beats e incluyen
el registro general y las opciones de salida. Convenientemente, un ejemplo de configuración
completa que incluye todas las opciones disponibles en: etc / heartbeat / heartbeat.full.yml .

Heartbeat y systemd
Los paquetes DEB y RPM incluyen una unidad de servicio para sistemas Linux con
systemd. En estos sistemas, puede administrar Heartbeat usando los comandos habituales
de systemd.

Iniciar y detener Heartbeat

Utilice systemctlpara iniciar o detener Heartbeat:

systemctl start heartbeat

systemctl stop heartbeat

De forma predeterminada, el servicio Heartbeat se inicia automáticamente cuando se inicia el

sistema. Para habilitar o deshabilitar el uso de inicio automático:

systemctl enable heartbeat

systemctl disable heartbeat

Estado del heartbeat y registros

Para obtener el estado del servicio, use systemctl:

38
ELK Caso práctico

Systemctl status heartbeat

Los registros se almacenan por defecto en journald. Para ver los registros, use journalctl:

journalctl - u heartbeat.service

El archivo de unidad incluido en los paquetes establece la -e bandera de forma

predeterminada. Este indicador hace que Heartbeat log to stderr e inhabilite
otras salidas de log. Systemd almacena todas las salidas enviadas a stderr en
journald.

39