Integración de ELK Stack: Sincronizando Logstash, Elasticsearch y Kibana

Cuando Logstash, Elasticsearch y Kibana están desincronizados, los registros desaparecen, los paneles se ven vacíos o los campos aparecen con el tipo incorrecto. La integración de ELK Stack consiste menos en iniciar tres servicios y más en asegurarse de que los nombres de índices, mapeos, marcas de tiempo, credenciales y vistas de datos estén todos de acuerdo.

Esta guía recorre un pipeline de registro práctico: Logstash recibe eventos, los analiza, los envía a Elasticsearch y Kibana lee los índices o flujos de datos resultantes. Los ejemplos utilizan la sintaxis clásica de pipeline de Logstash y las API de Elasticsearch que puedes ejecutar desde las Herramientas de Desarrollo de Kibana.

Comprendiendo el Flujo de Datos

Rastrea un evento a través del stack:

1. Logstash recibe datos de Beats, TCP, syslog, archivos, colas u otra entrada.
2. Los filtros de Logstash analizan, enriquecen, renombran y normalizan campos.
3. Elasticsearch indexa el evento usando plantillas, mapeos y políticas de ciclo de vida.
4. Kibana consulta Elasticsearch a través de una vista de datos y muestra el evento en Discover, paneles, Lens o alertas.

La mayoría de los errores de integración ocurren en los límites. Logstash no puede conectarse, Elasticsearch rechaza el documento o Kibana está mirando la vista de datos o el rango de tiempo incorrectos.

Configuración de Logstash para un Flujo de Datos Limpio

Los pipelines de Logstash tienen tres bloques principales: input, filter y output. Mantén cada bloque simple y comprobable.

Plugins de Entrada

Los plugins de entrada comunes incluyen:

• beats: Recibe eventos de Filebeat, Metricbeat y otros Beats.
• tcp / udp: Recibe eventos a través de sockets de red.
• file: Lee archivos locales. Esto es útil para implementaciones pequeñas y pruebas, pero los agentes suelen ser mejores para hosts de producción distribuidos.
• syslog: Recibe mensajes de syslog.

Ejemplo de entrada Beats con TLS:

input {
  beats {
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/etc/pki/tls/certs/logstash.crt"
    ssl_key => "/etc/pki/tls/private/logstash.key"
  }
}

Asegúrate de que el puerto esté abierto, el certificado coincida con cómo se conectan los clientes y los nombres de las opciones coincidan con la versión del plugin instalado. Las versiones recientes del plugin de entrada Beats usan ssl_enabled.

Plugins de Filtro

Los filtros convierten eventos sin procesar en campos útiles. El orden importa porque Logstash ejecuta los filtros secuencialmente.

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

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

  mutate {
    remove_field => [ "message" ]
  }
}

Usa grok para texto no estructurado, date para establecer @timestamp, mutate para limpiar campos y geoip cuando necesites enriquecimiento de ubicación basado en IP. Prueba los patrones grok con líneas de registro reales antes de ponerlos en producción. Un pequeño error de análisis puede enviar miles de eventos a Elasticsearch con campos faltantes.

Plugin de Salida

Para el stack ELK, la salida de Elasticsearch es el destino habitual.

output {
  elasticsearch {
    hosts => ["https://elasticsearch-node1:9200", "https://elasticsearch-node2:9200"]
    index => "my-logs-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${LOGSTASH_ES_PASSWORD}"
    ssl_enabled => true
    cacert => "/etc/logstash/certs/http_ca.crt"
  }
}

El valor de index es el contrato con las plantillas de Elasticsearch y las vistas de datos de Kibana. Si Logstash escribe my-logs-2026.05.23, tu plantilla y vista de datos deben coincidir con my-logs-*.

Para entornos más grandes, considera flujos de datos y Gestión del Ciclo de Vida de Índices en lugar de índices diarios gestionados manualmente. Si usas flujos de datos, sigue la guía actual de Elastic para la salida de Logstash con configuraciones de data_stream en lugar de mezclar opciones de flujo de datos e índices clásicos.

Plantillas y Mapeos de Elasticsearch

Elasticsearch necesita mapeos consistentes antes de que lleguen los documentos. De lo contrario, el primer documento puede establecer un tipo de campo que rompa eventos posteriores. Un código de estado que llega primero como "200" puede convertirse en texto o keyword en lugar de un número.

Ejemplo de plantilla de índice componible:

PUT _index_template/my_log_template
{
  "index_patterns": ["my-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": {
      "properties": {
        "@timestamp": {"type": "date"},
        "message": {"type": "text"},
        "host.name": {"type": "keyword"},
        "log.level": {"type": "keyword"},
        "http.response.status_code": {"type": "integer"}
      }
    }
  }
}

Usa keyword para coincidencias exactas y agregaciones, text para búsqueda de texto completo, tipos numéricos para métricas y códigos de estado, y date para campos de tiempo. Mantén un número modesto de shards a menos que tengas una razón medida para agregar más. Demasiados shards pequeños pueden afectar el rendimiento del clúster.

Vistas de Datos de Kibana

La interfaz de usuario actual de Kibana usa vistas de datos. Las versiones anteriores las llamaban patrones de índice. Crea una vista de datos que coincida con los nombres de índices o flujos de datos que Elasticsearch realmente tiene.

Configuración típica:

1. Ve a Stack Management -> Kibana -> Data Views.
2. Crea una vista de datos como my-logs-*.
3. Elige @timestamp como campo de tiempo.
4. Abre Discover y amplía el selector de tiempo mientras pruebas.

Si Discover está vacío, no asumas que Logstash falló. Verifica el rango de tiempo, el patrón de la vista de datos y si @timestamp se analizó correctamente.

Solución de Problemas Comunes de Integración

Los Datos No Aparecen en Kibana

Verifica cada salto:

GET _cat/indices/my-logs-*?v
GET my-logs-*/_search?size=1&sort=@timestamp:desc

Luego verifica:

• Los registros de Logstash en busca de errores de conexión, autenticación, TLS o mapeo.
• Los registros de Elasticsearch en busca de documentos rechazados y fallos de seguridad.
• El patrón de la vista de datos de Kibana y el rango de tiempo seleccionado.
• Si la marca de tiempo del evento está en el futuro, en el pasado o falta.

Documentos Rechazados por Elasticsearch

Los conflictos de mapeo son comunes. Por ejemplo, un evento envía http.response.status_code como 200, mientras que otro envía "OK". Elasticsearch no puede almacenar ambos en un campo integer.

Arregla el filtro de Logstash para que el campo tenga un tipo consistente, o enruta los eventos incorrectos a un índice separado para revisión. No sigas eliminando y recreando índices sin arreglar el pipeline que crea los documentos incorrectos.

Logstash Usa Demasiada CPU

Los patrones grok costosos, el alto volumen de eventos y los eventos multilínea grandes pueden aumentar rápidamente la CPU de Logstash. Comienza midiendo qué pipeline está ocupado, luego simplifica los patrones, ancla las expresiones regulares y mueve el análisis simple a Beats o pipelines de ingesta de Elasticsearch cuando sea más fácil de operar.

Las Consultas de Kibana Son Lentas

Los paneles lentos a menudo provienen de rangos de tiempo amplios, agregaciones de alta cardinalidad, demasiados shards o campos mapeados como text cuando Kibana necesita keyword. Usa valores predeterminados de panel más estrechos, rollover de ILM y mapeos de campo que coincidan con tus visualizaciones.

Conclusión

Trata la integración de ELK Stack como un contrato entre tres capas. Logstash debe emitir campos predecibles, Elasticsearch debe mapearlos y almacenarlos correctamente, y Kibana debe consultar la vista de datos correcta en el rango de tiempo correcto. Cuando algo se rompe, sigue un evento de muestra desde la entrada hasta el índice y el panel.