ELK-Stack-Integration: Synchronisation von Logstash, Elasticsearch und Kibana

Wenn Logstash, Elasticsearch und Kibana nicht synchron sind, verschwinden Logs, Dashboards bleiben leer oder Felder erscheinen mit dem falschen Typ. Die ELK-Stack-Integration dreht sich weniger um das Starten von drei Diensten, sondern vielmehr darum, dass Indexnamen, Mappings, Zeitstempel, Anmeldeinformationen und Datenansichten übereinstimmen.

Diese Anleitung führt durch eine praktische Logging-Pipeline: Logstash empfängt Ereignisse, parst sie, sendet sie an Elasticsearch, und Kibana liest die resultierenden Indizes oder Datenströme. Die Beispiele verwenden die klassische Logstash-Pipeline-Syntax und Elasticsearch-APIs, die Sie in den Kibana Dev Tools ausführen können.

Den Datenfluss verstehen

Verfolgen Sie ein Ereignis durch den Stack:

1. Logstash empfängt Daten von Beats, TCP, Syslog, Dateien, Warteschlangen oder einer anderen Eingabe.
2. Logstash-Filter parsen, reichern an, benennen um und normalisieren Felder.
3. Elasticsearch indiziert das Ereignis mithilfe von Vorlagen, Mappings und Lebenszyklusrichtlinien.
4. Kibana fragt Elasticsearch über eine Datenansicht ab und zeigt das Ereignis in Discover, Dashboards, Lens oder Benachrichtigungen an.

Die meisten Integrationsfehler treten an den Grenzen auf. Logstash kann keine Verbindung herstellen, Elasticsearch lehnt das Dokument ab, oder Kibana zeigt auf die falsche Datenansicht oder den falschen Zeitbereich.

Logstash-Konfiguration für sauberen Datenfluss

Logstash-Pipelines haben drei Hauptblöcke: input, filter und output. Halten Sie jeden Block einfach und testbar.

Eingabe-Plugins

Häufige Eingabe-Plugins sind:

• beats: Empfängt Ereignisse von Filebeat, Metricbeat und anderen Beats.
• tcp / udp: Empfängt Ereignisse über Netzwerk-Sockets.
• file: Liest lokale Dateien. Dies ist nützlich für kleine Bereitstellungen und Tests, aber Agents sind normalerweise besser für verteilte Produktionshosts.
• syslog: Empfängt Syslog-Nachrichten.

Beispiel für Beats-Eingabe mit TLS:

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

Stellen Sie sicher, dass der Port geöffnet ist, das Zertifikat zur Verbindungsart der Clients passt und die Optionsnamen Ihrer installierten Plugin-Version entsprechen. Neuere Versionen des Beats-Eingabe-Plugins verwenden ssl_enabled.

Filter-Plugins

Filter wandeln rohe Ereignisse in nützliche Felder um. Die Reihenfolge ist wichtig, da Logstash Filter sequenziell ausführt.

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

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

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

Verwenden Sie grok für unstrukturierten Text, date zum Setzen von @timestamp, mutate zur Feldbereinigung und geoip, wenn Sie eine IP-basierte Standortanreicherung benötigen. Testen Sie grok-Muster an echten Logzeilen, bevor Sie sie in Produktion nehmen. Ein kleiner Parsing-Fehler kann Tausende von Ereignissen mit fehlenden Feldern an Elasticsearch senden.

Ausgabe-Plugin

Für den ELK-Stack ist die Elasticsearch-Ausgabe das übliche Ziel.

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"
  }
}

Der index-Wert ist der Vertrag mit Elasticsearch-Vorlagen und Kibana-Datenansichten. Wenn Logstash my-logs-2026.05.23 schreibt, sollten Ihre Vorlage und Datenansicht mit my-logs-* übereinstimmen.

Erwägen Sie für größere Umgebungen Datenströme und Index-Lebenszyklus-Management anstelle von manuell verwalteten täglichen Indizes. Wenn Sie Datenströme verwenden, folgen Sie den aktuellen Elastic-Leitlinien für die Logstash-Ausgabe zu data_stream-Einstellungen, anstatt Datenstrom- und klassische Indexoptionen zu mischen.

Elasticsearch-Vorlagen und Mappings

Elasticsearch benötigt konsistente Mappings, bevor Dokumente eintreffen. Andernfalls kann das erste Dokument einen Feldtyp setzen, der spätere Ereignisse stört. Ein Statuscode, der zuerst als "200" eintrifft, könnte zu Text oder Keyword werden, anstatt zu einer Zahl.

Beispiel für eine zusammensetzbare Indexvorlage:

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"}
      }
    }
  }
}

Verwenden Sie keyword für exakte Übereinstimmungen und Aggregationen, text für Volltextsuche, numerische Typen für Metriken und Statuscodes und date für Zeitfelder. Halten Sie die Shard-Anzahl bescheiden, es sei denn, Sie haben einen gemessenen Grund, mehr hinzuzufügen. Zu viele kleine Shards können die Cluster-Leistung beeinträchtigen.

Kibana-Datenansichten

Die aktuelle Kibana-Oberfläche verwendet Datenansichten. Ältere Versionen nannten sie Indexmuster. Erstellen Sie eine Datenansicht, die mit den Indexnamen oder Datenströmen übereinstimmt, die Elasticsearch tatsächlich hat.

Typische Einrichtung:

1. Gehen Sie zu Stack Management -> Kibana -> Data Views.
2. Erstellen Sie eine Datenansicht wie my-logs-*.
3. Wählen Sie @timestamp als Zeitfeld.
4. Öffnen Sie Discover und erweitern Sie die Zeitauswahl während des Tests.

Wenn Discover leer ist, gehen Sie nicht davon aus, dass Logstash fehlgeschlagen ist. Überprüfen Sie den Zeitbereich, das Datenansichtsmuster und ob @timestamp korrekt geparst wurde.

Fehlerbehebung bei häufigen Integrationsproblemen

Daten erscheinen nicht in Kibana

Überprüfen Sie jeden Schritt:

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

Überprüfen Sie dann:

• Logstash-Logs auf Verbindungs-, Authentifizierungs-, TLS- oder Mapping-Fehler.
• Elasticsearch-Logs auf abgelehnte Dokumente und Sicherheitsfehler.
• Das Kibana-Datenansichtsmuster und den ausgewählten Zeitbereich.
• Ob der Ereigniszeitstempel in der Zukunft, in der Vergangenheit oder fehlend ist.

Dokumente werden von Elasticsearch abgelehnt

Mapping-Konflikte sind häufig. Zum Beispiel sendet ein Ereignis http.response.status_code als 200, während ein anderes "OK" sendet. Elasticsearch kann nicht beide in einem integer-Feld speichern.

Korrigieren Sie den Logstash-Filter, sodass das Feld konsistent typisiert ist, oder leiten Sie fehlerhafte Ereignisse zur Überprüfung an einen separaten Index weiter. Löschen und erstellen Sie Indizes nicht ständig neu, ohne die Pipeline zu reparieren, die die fehlerhaften Dokumente erstellt.

Logstash verbraucht zu viel CPU

Teure grok-Muster, hohes Ereignisvolumen und große mehrzeilige Ereignisse können die Logstash-CPU schnell in die Höhe treiben. Messen Sie zunächst, welche Pipeline ausgelastet ist, vereinfachen Sie dann Muster, verankern Sie Regex und verlagern Sie einfaches Parsen auf Beats oder Elasticsearch-Ingest-Pipelines, wenn dies einfacher zu betreiben ist.

Kibana-Abfragen sind langsam

Langsame Dashboards sind oft auf weite Zeitbereiche, Aggregationen mit hoher Kardinalität, zu viele Shards oder Felder zurückzuführen, die als text gemappt sind, wenn Kibana keyword benötigt. Verwenden Sie engere Dashboard-Standardwerte, ILM-Rollover und Feld-Mappings, die zu Ihren Visualisierungen passen.

Fazit

Behandeln Sie die ELK-Stack-Integration als einen Vertrag zwischen drei Ebenen. Logstash muss vorhersagbare Felder ausgeben, Elasticsearch muss sie korrekt mappen und speichern, und Kibana muss die richtige Datenansicht über den richtigen Zeitbereich abfragen. Wenn etwas kaputt geht, folgen Sie einem Beispielereignis von der Eingabe über den Index bis zum Dashboard.