ProHoster > Blog > Administración > Aplicación práctica de ELK. Configuramos logstash

Aplicación práctica de ELK. Configuramos logstash

Introducción

Al implementar otro sistema, nos encontramos con la necesidad de procesar una gran cantidad de registros diversos. Elegimos ELK como herramienta. En este artículo hablaremos de nuestra experiencia configurando este stack.

No tenemos como objetivo describir todas sus capacidades, sino centrarnos en la solución de problemas prácticos. Esto se debe a que, a pesar de la abundancia de documentación y de imágenes ya preparadas, hay muchos escollos, al menos los hemos encontrado nosotros.

Desplegamos el stack a través de docker-compose. Además, teníamos un docker-compose.yml bien elaborado que nos permitió levantar el stack casi sin problemas. Y nos parecía que la victoria estaba cerca, solo teníamos que ajustarlo un poco a nuestras necesidades y todo estaría listo.

Desafortunadamente, el intento de ajustar el sistema para recibir y procesar registros de nuestra aplicación no tuvo éxito de inmediato. Por lo tanto, decidimos que era mejor estudiar cada componente por separado, y luego regresar a sus interrelaciones.

Así que comenzamos con logstash.

Entorno, despliegue, ejecución de Logstash en un contenedor

Utilizamos docker-compose para el despliegue, los experimentos descritos aquí se realizaron en MacOS y Ubuntu 18.0.4.

La imagen de logstash que estaba especificada en nuestro docker-compose.yml es docker.elastic.co/logstash/logstash:6.3.2

Esta la utilizaremos para nuestros experimentos.

Para ejecutar logstash, escribimos un docker-compose.yml separado. Claro que podríamos haber lanzado la imagen desde la línea de comandos, pero estábamos resolviendo una tarea específica donde todo se ejecuta desde docker-compose.

Breve sobre los archivos de configuración

Como se desprende de la descripción, logstash se puede ejecutar ya sea para un solo canal, en cuyo caso necesita un archivo *.conf, o para varios canales, en cuyo caso debe recibir un archivo pipelines.yml, que a su vez hará referencia a archivos .conf para cada canal.
Optamos por el segundo camino. Nos pareció más versátil y escalable. Por lo tanto, creamos pipelines.yml y configuramos un directorio pipelines, donde colocaremos los archivos .conf para cada canal.

Dentro del contenedor hay otro archivo de configuración: logstash.yml. No lo tocamos, lo usamos tal como está.

Así que, la estructura de nuestros directorios es la siguiente:

Aplicación práctica de ELK. Configuramos logstash

Para obtener los datos de entrada, por ahora asumimos que es tcp a través del puerto 5046, y para la salida utilizaremos stdout.

Esta es una configuración simple para el primer inicio. Dado que la tarea inicial es lanzar.

Entonces, tenemos este docker-compose.yml.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

¿Qué vemos aquí?

  1. Las redes y volúmenes fueron tomados del docker-compose.yml original (donde se lanza todo el stack) y creo que no afectan mucho a la imagen general.
  2. Creamos un servicio (services) logstash, de la imagen docker.elastic.co/logstash/logstash:6.3.2 y le asignamos el nombre logstash_one_channel.
  3. Redirigimos el puerto 5046 al mismo puerto interno.
  4. Mapeamos nuestro archivo de configuración de canales ./config/pipelines.yml al archivo /usr/share/logstash/config/pipelines.yml dentro del contenedor, desde donde lo recogerá logstash y lo hacemos de solo lectura, por si acaso.
  5. Mapeamos el directorio ./config/pipelines, donde tenemos los archivos de configuración de canales, al directorio /usr/share/logstash/config/pipelines y también lo hacemos de solo lectura.

Aplicación práctica de ELK. Configuramos logstash

El archivo pipelines.yml.

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

Aquí se describe un canal con el identificador HABR y la ruta a su archivo de configuración.

Y finalmente, el archivo ./config/pipelines/habr_pipeline.conf.

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

No vamos a profundizar en su descripción por ahora, intentemos ejecutarlo:

docker-compose up

¿Qué estamos viendo?

El contenedor se ha iniciado. Podemos comprobar su funcionamiento:

echo '13123123123123123123123213123213' | nc localhost 5046

Y vemos en la consola del contenedor la respuesta:

Aplicación práctica de ELK. Configuramos logstash

Pero al mismo tiempo, también vemos:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Unable to retrieve license information from license server {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch", …

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Pipeline started successfully {:pipeline_id=>".monitoring-logstash", :thread=>"#"}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Pipelines running {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[]}
logstash_one_channel | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics ] X-Pack is installed on Logstash but not on Elasticsearch. Please install X-Pack on Elasticsearch to use the monitoring feature. Other features may be available.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Ejecutando comprobación de salud para ver si la conexión a Elasticsearch está funcionando {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Intentó resucitar la conexión a una instancia de ES muerta, pero recibió un error. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch Inalcanzable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Ejecutando comprobación de salud para ver si la conexión a Elasticsearch está funcionando {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Intentó resucitar la conexión a una instancia de ES muerta, pero recibió un error. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch Inalcanzable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

Y nuestro registro sigue subiendo todo el tiempo.

Aquí he destacado en verde el mensaje de que el pipeline se inició correctamente, en rojo el mensaje de error y en amarillo el mensaje sobre el intento de conectar con elasticsearch:9200.
Esto se debe a que en logstash.conf, incluido en la imagen, hay una verificación de la disponibilidad de Elasticsearch. Logstash supone que opera dentro del stack ELK, y nosotros lo hemos separado.

Se puede trabajar, pero no es conveniente.

La solución es desactivar esta verificación a través de la variable de entorno XPACK_MONITORING_ENABLED.

Haremos un cambio en docker-compose.yml y lo reiniciaremos:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

Ahora está todo bien. El contenedor está listo para experimentos.

Podemos volver a escribir en la consola vecina:

echo '13123123123123123123123213123213' | nc localhost 5046

Y ver:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

Trabajo en un solo canal

Así que hemos arrancado. Ahora podemos dedicar tiempo a la configuración de logstash. Por ahora no tocaremos el archivo pipelines.yml, veamos qué podemos obtener trabajando con un solo canal.

Cabe mencionar que el principio general de trabajo con el archivo de configuración del canal está bien descrito en la guía oficial, aquí aquí
Si deseas leer en español, hemos utilizado este artículo(pero el sintaxis de las consultas es antigua, hay que tenerlo en cuenta).

Pasemos secuencialmente a la sección de entrada. Ya hemos visto el trabajo por tcp. ¿Qué más puede ser interesante aquí?

Mensajes de prueba, utilizando heartbeat

Hay una interesante posibilidad de generar mensajes de prueba automáticos.
Para esto, en la sección de entrada hay que activar el plugin heartbeat.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  }

Activamos y comenzamos a recibir cada minuto

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hola Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

Queremos recibir con más frecuencia, hay que añadir el parámetro interval.
Así recibiremos un mensaje cada 10 segundos.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

Obtención de datos desde un archivo

También decidimos mirar el modo de archivo. Si funciona bien con el archivo, entonces tal vez no se necesite ningún agente, al menos para uso local.

Por la descripción, el modo de operación debería ser similar a tail -f, es decir, lee nuevas líneas o, como opción, lee todo el archivo.

Entonces, ¿qué queremos obtener?

  1. Queremos recibir las líneas que se añaden a un archivo de registro.
  2. Queremos recibir datos que se escriben en varios archivos de registro, con la posibilidad de distinguir de dónde proviene cada dato.
  3. Queremos comprobar que al reiniciar logstash no volverá a recibir esos datos.
  4. Queremos verificar que si logstash se apaga, pero los datos siguen escribiéndose en los archivos, entonces, cuando lo iniciemos de nuevo, recibiremos esos datos.

Para realizar el experimento, añadiremos una línea más en docker-compose.yml, abriendo el directorio en el que colocamos los archivos.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

Y cambiamos la sección de entrada en habr_pipeline.conf

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

Iniciamos:

docker-compose up

Para crear y escribir archivos de registro vamos a usar el comando:

echo '1' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

¡Sí, funciona!

Además, vemos que se ha agregado automáticamente el campo path. Esto significa que podremos filtrar registros más adelante.

Intentemos de nuevo:

echo '2' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Ahora en otro archivo:

 echo '1' >> logs/number2.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

¡Genial! El archivo se procesó correctamente, el path se indicó correctamente, todo va bien.

Detenemos logstash y lo reiniciamos. Esperemos. Silencio. Es decir, no obtendremos estos registros nuevamente.

Y ahora el experimento más audaz.

Detenemos logstash y ejecutamos:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

Reiniciamos logstash y vemos:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

¡Hurra! Todo se ha procesado.

Sin embargo, hay que advertir que si el contenedor de logstash se elimina (docker stop logstash_one_channel && docker rm logstash_one_channel), no se procesará nada. Dentro del contenedor se guardó la posición del archivo hasta donde se había leído. Si se ejecuta 'desde cero', solo aceptará nuevas líneas.

Lectura de archivos ya existentes

Supongamos que estamos iniciando logstash por primera vez, pero ya tenemos registros y quisiéramos procesarlos.
Si ejecutamos logstash con la sección input que utilizamos anteriormente, no obtendremos nada. Solo se procesarán las nuevas líneas que logstash reciba.

Para que se extraigan las líneas de los archivos existentes, debemos agregar una línea adicional en la sección input:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

Cabe mencionar que esto solo funciona para nuevos archivos que logstash aún no ha visto. Para aquellos archivos que ya han sido procesados por logstash, ya ha registrado su tamaño y ahora solo tomará las nuevas entradas en ellos.

Nos detendremos aquí en el estudio de la sección input. Hay muchas más variantes, pero para nuestros experimentos, por ahora, esto será suficiente.

Enrutamiento y transformación de datos

Intentaremos resolver la siguiente tarea: supongamos que estamos recibiendo mensajes de un canal, algunos de ellos son informativos y otros son mensajes de error. Se diferencian por las etiquetas. Algunos son INFO, otros son ERROR.

Necesitamos dividirlos en la salida. Es decir, los mensajes informativos deben ir a un canal y los mensajes de error a otro.

Para esto, pasamos de la sección input a filter y output.

Con la sección filter desglosaremos el mensaje entrante, obteniendo de él un hash (pares clave-valor), que se puede manipular según las condiciones. Y en la sección output, seleccionaremos los mensajes y enviaremos cada uno a su canal correspondiente.

Desglose de un mensaje con grok

Para analizar cadenas de texto y obtener un conjunto de campos, en la sección filter hay un plugin especial: grok.

Sin el objetivo de ofrecer aquí una descripción detallada (para eso remitimos a documentación oficial), daré un simple ejemplo.

Para esto, es necesario definir el formato de las cadenas de entrada. Las mías son las siguientes:

1 INFO message1
2 ERROR message2

Es decir, el identificador en primer lugar, luego INFO/ERROR y después alguna palabra sin espacios.
No es complicado, pero para entender el principio de funcionamiento será suficiente.

Así que, en la sección filter, en el plugin grok debemos definir un patrón para analizar nuestras cadenas.

Se verá así:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  }

En esencia, esto es una expresión regular. Se utilizan patrones predeterminados, como INT, LOGLEVEL, WORD. Su descripción, así como otros patrones, se puede consultar aquí aquí

Ahora, al pasar por este filtro, nuestra cadena se convertirá en un hash de tres campos: message_id, message_type, message_text.

Estos serán los que se mostrarán en la sección output.

Rutina de mensajes en la sección de salida mediante el comando if

En la sección de salida, como recordamos, íbamos a dividir los mensajes en dos flujos. Los de tipo INFO los mostraremos en la consola, mientras que los de error los escribiremos en un archivo.

¿Cómo podemos dividir estos mensajes? La condición del problema ya sugiere la solución: tenemos un campo destacado message_type, que solo puede tener dos valores: INFO y ERROR. Lo utilizaremos para hacer la selección con el operador if.

if [message_type] == "ERROR" {
        # Aquí lo escribimos en un archivo
       } else
     {
      # Aquí lo mostramos en stdout
    }

Puedes ver la descripción del trabajo con campos y operadores en esta sección del manual oficial.

Ahora, hablemos de la salida en sí.

La salida en la consola es bastante clara: stdout {}

Pero la salida en un archivo: recordemos que estamos ejecutando todo desde un contenedor y, para que el archivo en el que escribimos el resultado sea accesible desde el exterior, necesitamos abrir este directorio en docker-compose.yml.

Total:

La sección de salida de nuestro archivo se ve así:

output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "formato personalizado: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

En docker-compose.yml añadimos otro volumen para la salida:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

Iniciamos, probamos y vemos la división en dos flujos.

Fuente: habr.com

Compra un hosting fiable para sitios web con protección contra DDoS, servidores VPS VDS 🔥 Compra un hosting fiable para sitios web con protección contra DDoS, servidores VPS VDS | ProHoster