🥇 Conectando ELK y Exchange. Parte 2

Continuo mi relato sobre cómo integrar Exchange y ELK (parte 1) aquí). Recuerdo que esta combinación puede manejar sin problemas una gran cantidad de registros. Esta vez hablaremos sobre cómo hacer funcionar Exchange con los componentes Logstash y Kibana.

Logstash en el stack ELK se utiliza para el procesamiento inteligente de registros y su preparación para ser almacenados en Elastic en forma de documentos, a partir de los cuales es conveniente construir diversas visualizaciones en Kibana.

Instalación

Consiste en dos etapas:

Instalación y configuración del paquete OpenJDK.
Instalación y configuración del paquete Logstash.

Instalación y configuración del paquete OpenJDK

El paquete OpenJDK debe ser descargado y descomprimido en un directorio específico. Luego, la ruta a este directorio debe ser añadida a las variables $env:Path y $env:JAVA_HOME del sistema operativo Windows:

Verifiquemos la versión de Java:

PS C:java -version
openjdk version "13.0.1" 2019-10-15
OpenJDK Runtime Environment (build 13.0.1+9)
OpenJDK 64-Bit Server VM (build 13.0.1+9, modo mezclado, compartición)

Instalación y configuración del paquete Logstash

Descargue el archivo comprimido con la distribución de Logstash desde aquí. El archivo debe ser descomprimido en la raíz del disco. No es recomendable descomprimir en una carpeta C:Program Files , ya que Logstash se negará a iniciarse correctamente. Luego, es necesario hacer las modificaciones en el archivo jvm.options que corresponden a la asignación de memoria RAM para el proceso de Java. Recomiendo especificar la mitad de la memoria del servidor. Si cuenta con 16 GB de RAM, las claves por defecto serían:

-Xms1g
-Xmx1g

deben ser reemplazadas por:

-Xms8g
-Xmx8g

Además, es conveniente comentar la línea -XX:+UseConcMarkSweepGC. Más detalles sobre esto aquí. El siguiente paso es crear la configuración por defecto en el archivo logstash.conf:

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

Con esta configuración, Logstash lee datos desde la consola, los pasa a través de un filtro vacío y los devuelve a la consola. Usar esta configuración permitirá verificar el funcionamiento de Logstash. Para ello, lo ejecutaremos en modo interactivo:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
El plugin stdin ahora está esperando entrada:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines en ejecución {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Punto final de la API de Logstash iniciado correctamente {:port=>9600}

Logstash se ha iniciado correctamente en el puerto 9600.

El paso final de la instalación: iniciar Logstash como un servicio de Windows. Esto se puede hacer, por ejemplo, utilizando el paquete NSSM:

PS C:...bin> .nssm.exe install logstash
¡Servicio "logstash" instalado con éxito!

Tolerancia a fallos

La integridad de los registros durante la transmisión desde el servidor de origen se garantiza mediante el mecanismo de Colas Persistentes.

Cómo funciona

Esquema de disposición de las colas en el proceso de procesamiento de registros: input → queue → filter + output.

El plugin input recibe datos de la fuente de logs, los escribe en la cola y envía una confirmación de recepción a la fuente.

Los mensajes de la cola son procesados por Logstash, pasan por un filtro y el plugin output. Al recibir la confirmación de envío del output, Logstash elimina el registro procesado de la cola. Si Logstash se detiene, todos los mensajes no procesados y aquellos para los que no se ha recibido confirmación de envío permanecen en la cola, y Logstash continuará procesándolos en el próximo inicio.

Configuración

Se regula mediante las claves en el archivo C:Logstashconfiglogstash.yml:

queue.type: (valores posibles — persisted y memory (predeterminado)).
path.queue: (ruta a la carpeta con los archivos de colas, que por defecto se almacenan en C:Logstashqueue).
queue.page_capacity: (tamaño máximo de la página de la cola, el valor predeterminado es — 64mb).
queue.drain: (true/false — activa/desactiva la detención del procesamiento de la cola antes de detener Logstash. No recomiendo habilitarlo, ya que afectará directamente la velocidad de apagado del servidor).
queue.max_events: (número máximo de eventos en la cola, por defecto — 0 (sin límite)).
queue.max_bytes: (tamaño máximo de la cola en bytes, por defecto — 1024mb (1gb)).

Si están configuradas queue.max_events y queue.max_bytes, los mensajes dejarán de ser aceptados en la cola al alcanzar el valor de cualquiera de estas configuraciones. Más sobre las Colas Persistentes se explica aquí.

Ejemplo de parte de logstash.yml, que responde a la configuración de la cola:

queue.type: persisted
queue.max_bytes: 10gb

Configuración

La configuración de Logstash generalmente consta de tres partes, responsables de diferentes fases del procesamiento de logs entrantes: recepción (sección input), análisis (sección filter) y envío a Elastic (sección output). A continuación, examinaremos cada una de ellas con más detalle.

Entrada

El flujo entrante con logs en crudo es recibido de los agentes filebeat. Precisamente este plugin es el que especificamos en la sección input:

input {
  beats {
    port => 5044
  }
}

Después de tal configuración, Logstash comienza a escuchar el puerto 5044 y, al recibir los logs, los procesa según la configuración de la sección filter. Si es necesario, se puede envolver el canal de recepción de logs de filebeat en SSL. Más sobre las configuraciones del plugin beats está escrito. aquí.

Filtrar

Todos los registros de texto interesantes generados por Exchange tienen un formato csv con los campos descritos en el archivo de registro. Para el análisis de registros csv, Logstash nos ofrece tres complementos: dissect, csv y grok. El primero es el más rápido, pero solo maneja el análisis de los registros más simples.
Por ejemplo, la siguiente entrada se dividirá en dos (debido a la presencia de una coma dentro del campo), lo que resultará en un análisis incorrecto del registro:

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

Se puede utilizar al analizar registros, por ejemplo, de IIS. En este caso, la sección de filtro podría lucir de la siguiente manera:

filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
    }
  }
}

La configuración de Logstash permite utilizar operadores condicionales, por lo que solo podemos dirigir los registros al complemento dissect que han sido etiquetados por filebeat con la etiqueta IIS. Dentro del complemento, asociamos los valores de los campos con sus nombres, eliminamos el campo original message, que contenía el registro del log, y podemos agregar un campo arbitrario, que podría, por ejemplo, contener el nombre de la aplicación de la que recopilamos los registros.

En el caso de los registros de seguimiento, es mejor usar el complemento csv, ya que puede manejar correctamente campos complejos:

filter {
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
}

Dentro del complemento, asociamos los valores de los campos con sus nombres, eliminamos el campo original message (así como los campos tenant-id y schema-version), que contenía el registro del log, y podemos agregar un campo arbitrario que podría, por ejemplo, contener el nombre de la aplicación de la que recopilamos los registros.

Al final de la etapa de filtrado, obtendremos documentos que en primera instancia estarán listos para la visualización en Kibana. Nos faltará lo siguiente:

Los campos numéricos se reconocerán como texto, lo que impide realizar operaciones con ellos. En particular, los campos time-taken del registro de IIS, así como los campos recipient-count y total-bites del registro de Tracking.
La marca de tiempo estándar del documento contendrá el tiempo de procesamiento del registro, no el tiempo de escritura en el lado del servidor.
Campo recipient-address aparecerá como una única cadena, lo que impide realizar un análisis que cuente los destinatarios de los correos electrónicos.

Es hora de añadir un poco de magia al proceso de procesamiento de registros.

La conversión de campos numéricos

El plugin dissect tiene una opción convert_datatype, que se puede usar para convertir un campo de texto a un formato numérico. Por ejemplo, así:

dissect {
  …
  convert_datatype => { "time-taken" => "int" }
  …
}

Cabe recordar que este método solo es adecuado si el campo definitivamente contendrá una cadena. Los valores nulos de los campos no son procesados por esta opción y se lanzará una excepción.

Para los registros de tracking, es preferible no usar un método convert similar, ya que los campos recipient-count y total-bites pueden estar vacíos. Es mejor usar el plugin mutate:

mutate {
  convert => [ "total-bytes", "integer" ]
  convert => [ "recipient-count", "integer" ]
}

Dividir recipient_address en destinatarios individuales

Esta tarea también se puede resolver con el plugin mutate:

mutate {
  split => ["recipient_address", ";"]
}

Modificar timestamp

En el caso de los registros de tracking, la tarea se resuelve fácilmente con el plugin Muestra la fecha y hora actuales del sistema., que ayudará a escribir en el campo timestamp la fecha y hora en el formato deseado desde el campo date-time:

date {
  match => [ "date-time", "ISO8601" ]
  timezone => "Europe/Moscow"
  remove_field => [ "date-time" ]
}

En el caso de los registros de IIS, será necesario combinar los datos de los campos Muestra la fecha y hora actuales del sistema. y time usando el plugin mutate, especificar la zona horaria que necesitamos y colocar esta marca de tiempo en timestamp con el uso del plugin date:

mutate { 
  add_field => { "data-time" => "%{date} %{time}" }
  remove_field => [ "date", "time" ]
}
date { 
  match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
  timezone => "UTC"
  remove_field => [ "data-time" ]
}

Salida

La sección output se utiliza para enviar los logs procesados al receptor de logs. En caso de enviar directamente a Elastic, se usa el plugin elasticsearch, en el que se especifica la dirección del servidor y la plantilla del nombre del índice para enviar el documento formado:

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Configuración final

La configuración final se verá de la siguiente manera:

entrada {
  beats {
    puerto => 5044
  }
}
 
filtro {
  si "IIS" en [tags] {
    analizar {
      mapeo => {
        "mensaje" => "%{fecha} %{hora} %{s-ip} %{cs-método} %{cs-uri-stem} %{cs-uri-query} %{s-puerto} %{cs-usuario} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-estado} %{sc-subestado} %{sc-win32-estado} %{tiempo-tomado}"
      }
      eliminar_campo => ["mensaje"]
      agregar_campo => { "aplicación" => "intercambio" }
      convertir_tipo_dato => { "tiempo-tomado" => "int" }
    }
    mutar { 
      agregar_campo => { "datos-tiempo" => "%{fecha} %{hora}" }
      eliminar_campo => [ "fecha", "hora" ]
    }
    fecha { 
      emparejar => [ "datos-tiempo", "YYYY-MM-dd HH:mm:ss" ]
      zona_horaria => "UTC"
      eliminar_campo => [ "datos-tiempo" ]
    }
  }
  si "Rastreo" en [tags] {
    csv {
      columnas => ["fecha-hora","ip-cliente","nombre-host-cliente","ip-servidor","nombre-host-servidor","contexto-origen","id-conector","origen","id-evento","id-mensaje-interno","id-mensaje","id-mensaje-red","direccion-destinatario","estado-destinatario","total-bytes","contador-destinatarios","direccion-destinatario-relacionada","referencia","asunto-del-mensaje","direccion-remitente","ruta-de-regreso","informacion-del-mensaje","direccionalidad","id-inquilino","ip-cliente-original","ip-servidor-original","datos-personalizados","tipo-trafico-transporte","id-log","version-esquema"]
      eliminar_campo => ["mensaje", "id-inquilino", "version-esquema"]
      agregar_campo => { "aplicación" => "intercambio" }
    }
    mutar {
      convertir => [ "total-bytes", "entero" ]
      convertir => [ "contador-destinatarios", "entero" ]
      dividir => ["direccion_destinatario", ";"]
    }
    fecha {
      emparejar => [ "fecha-hora", "ISO8601" ]
      zona_horaria => "Europa/Moscú"
      eliminar_campo => [ "fecha-hora" ]
    }
  }
}
 
salida {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    gestionar_plantilla => false
    índice => "Intercambio-%{+YYYY.MM.dd}"
  }
}

Enlaces útiles: