Einführung

Bei der Bereitstellung eines weiteren Systems standen wir vor der Notwendigkeit, eine große Menge an unterschiedlichen Logs zu verarbeiten. Als Werkzeug wählten wir ELK. In diesem Artikel berichten wir über unsere Erfahrungen bei der Konfiguration dieses Stacks.

Wir setzen nicht das Ziel, alle Möglichkeiten zu beschreiben, sondern möchten uns auf die Lösung praktischer Aufgaben konzentrieren. Dies liegt daran, dass es bei einer ausreichenden Menge an Dokumentation und bereits fertiggestellten Images viele Fallstricke gibt, die zumindest bei uns aufgetreten sind.

Wir haben den Stack über docker-compose bereitgestellt. Darüber hinaus verfügten wir über eine gut geschriebene docker-compose.yml, die es uns ermöglichte, den Stack nahezu problemlos hochzufahren. Es schien uns, dass der Sieg schon nah war, wir mussten nur noch ein paar Anpassungen für unsere Bedürfnisse vornehmen und alles wäre bereit.

Leider war der Versuch, das System für den Empfang und die Verarbeitung von Logs unserer Anwendung anzupassen, nicht sofort von Erfolg gekrönt. Daher entschieden wir, jeden Komponenten einzeln zu studieren, bevor wir zu deren Verbindungen zurückkehren.

Also begannen wir mit Logstash.

Umgebung, Bereitstellung, Start von Logstash im Container

Für die Bereitstellung verwenden wir docker-compose, die hier beschriebenen Experimente wurden auf MacOS und Ubuntu 18.04 durchgeführt.

Das in unserer ursprünglichen docker-compose.yml angegebene Logstash-Image ist docker.elastic.co/logstash/logstash:6.3.2

Das werden wir für die Experimente verwenden.

Um Logstash zu starten, haben wir eine separate docker-compose.yml geschrieben. Man hätte das Image auch über die Kommandozeile starten können, aber wir hatten ja eine konkrete Aufgabe, bei der alles über docker-compose gestartet wird.

Kurze Infos zu den Konfigurationsdateien

Wie aus der Beschreibung hervorgeht, kann Logstash entweder für einen Kanal gestartet werden, in diesem Fall muss eine *.conf-Datei übergeben werden, oder für mehrere Kanäle, wobei ihm in diesem Fall die Datei pipelines.yml übergeben werden muss, die wiederum auf die .conf-Dateien für jeden Kanal verweist.
Wir haben den zweiten Weg gewählt. Dieser erschien uns universeller und skalierbarer. Daher haben wir die pipelines.yml erstellt und ein Verzeichnis namens pipelines angelegt, in das wir die .conf-Dateien für jeden Kanal legen werden.

Im Container gibt es eine weitere Konfigurationsdatei — logstash.yml. Wir fassen diese nicht an und verwenden sie, wie sie ist.

Also, die Struktur unserer Verzeichnisse:

Für den Empfang der Eingabedaten nehmen wir zunächst an, dass dies TCP über den Port 5046 ist, und für die Ausgabe verwenden wir stdout.

Hier ist eine so einfache Konfiguration für den ersten Start. Schließlich besteht die anfängliche Aufgabe darin, zu starten.

Also, wir haben diese 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

Was sehen wir hier?

Netzwerke und Volumes wurden aus der ursprünglichen docker-compose.yml (der, wo der gesamte Stapel gestartet wird) entnommen und ich denke, dass sie nicht viel zur Gesamtübersicht beitragen.
Wir erstellen einen Dienst (services) logstash aus dem Image docker.elastic.co/logstash/logstash:6.3.2 und geben ihm den Namen logstash_one_channel.
Wir leiten den Port 5046 in den Container auf denselben internen Port weiter.
Wir binden unsere Konfigurationsdatei ./config/pipelines.yml an die Datei /usr/share/logstash/config/pipelines.yml im Container, wo logstash sie übernehmen wird, und machen sie nur lesbar, nur für den Fall.
Wir binden das Verzeichnis ./config/pipelines, in dem die Konfigurationsdateien der Kanäle liegen, an das Verzeichnis /usr/share/logstash/config/pipelines und machen auch dieses nur lesbar.

Die Datei pipelines.yml

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

Hier ist ein Kanal mit der ID HABR beschrieben und der Pfad zu seiner Konfigurationsdatei.

Und schließlich die Datei „./config/pipelines/habr_pipeline.conf"

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

Lassen Sie uns noch nicht auf die Beschreibung eingehen, versuchen wir es zu starten:

docker-compose up

Was sehen wir?

Der Container ist gestartet. Wir können seine Funktion überprüfen:

echo '13123123123123123123123213123213' | nc localhost 5046

Und wir sehen in der Konsole des Containers die Antwort:

Aber gleichzeitig sehen wir auch:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Lizenzinformationen konnten nicht vom Lizenzserver abgerufen werden {:message=>„Elasticsearch nicht erreichbar: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch“, …

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

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Pipelines laufen {: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 ist auf Logstash installiert, aber nicht auf Elasticsearch. Bitte installieren Sie X-Pack auf Elasticsearch, um die Überwachungsfunktion nutzen zu können. Andere Funktionen sind möglicherweise verfügbar.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Logstash API-Endpunkt erfolgreich gestartet {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Gesundheitsscheck wird ausgeführt, um zu überprüfen, ob eine Elasticsearch-Verbindung funktioniert {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Versuch, die Verbindung zu einem abgestürzten ES-Instanz wiederherzustellen, aber es trat ein Fehler auf. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch nicht erreichbar: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Gesundheitsscheck wird ausgeführt, um zu überprüfen, ob eine Elasticsearch-Verbindung funktioniert {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Versuch, die Verbindung zu einem abgestürzten ES-Instanz wiederherzustellen, aber es trat ein Fehler auf. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch nicht erreichbar: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

Und unser Log kriecht ständig nach oben.

Hier habe ich die Nachricht, dass die Pipeline erfolgreich gestartet wurde, grün markiert, rot eine Fehlermeldung und gelb eine Nachricht über den Versuch, Kontakt aufzunehmen mit elasticsearch:9200.
Dies geschieht aufgrund der Verfügbarkeit von Elasticsearch in der logstash.conf, die Bestandteil des Images ist. Logstash geht davon aus, dass es im ELK-Stack läuft, aber wir haben es getrennt.

Es ist möglich zu arbeiten, aber es ist nicht bequem.

Die Lösung besteht darin, diese Überprüfung über die Umgebungsvariable XPACK_MONITORING_ENABLED zu deaktivieren.

Wir werden die Änderung in der docker-compose.yml vornehmen und erneut starten:

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

Jetzt ist alles in Ordnung. Der Container ist bereit für Experimente.

Wir können wieder in die benachbarte Konsole eingeben:

echo '13123123123123123123123213123213' | nc localhost 5046

Und sehen:

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" => "Hallo Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

Arbeit in einem Kanal

So, wir haben gestartet. Jetzt können wir uns der Konfiguration von Logstash widmen. Wir werden die Datei pipelines.yml vorerst nicht anfassen und sehen, was wir mit einem einzigen Kanal erreichen können.

Es sollte erwähnt werden, dass das allgemeine Prinzip der Arbeit mit der Konfigurationsdatei des Kanals gut in der offiziellen Anleitung beschrieben ist, hier hier
Wenn du auf Russisch lesen möchtest, haben wir diesen Artikel(aber die Syntax der Anfragen dort ist veraltet, das sollte berücksichtigt werden.)

Lass uns der Reihe nach vom Abschnitt Input ausgehen. Die Arbeit mit tcp haben wir bereits gesehen. Was könnte hier noch interessant sein?

Testnachrichten, die heartbeat verwenden

Es gibt diese interessante Möglichkeit, automatische Testnachrichten zu generieren.
Dazu muss im Input-Bereich das Plugin heartbeat aktiviert werden.

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

Wir aktivieren es und erhalten einmal pro Minute Daten.

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

Wir möchten häufiger diese Daten erhalten, daher müssen wir den Parameter interval hinzufügen.
So werden wir alle 10 Sekunden eine Nachricht erhalten.

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

Daten aus einer Datei erhalten

Wir haben uns auch entschieden, den Dateimodus zu prüfen. Wenn es gut mit der Datei funktioniert, ist möglicherweise gar kein Agent nötig, zumindest für die lokale Nutzung.

Laut Beschreibung sollte der Betriebsmodus ähnlich wie tail -f sein, d.h. es liest neue Zeilen oder optional die gesamte Datei.

Also, was wollen wir erreichen:

Wir wollen die Zeilen erhalten, die in eine Log-Datei geschrieben werden.
Wir wollen Daten erhalten, die in mehrere Log-Dateien geschrieben werden, und dabei die Möglichkeit haben, zu unterscheiden, was woher stammt.
Wir wollen überprüfen, dass beim Neustart von logstash diese Daten nicht erneut empfangen werden.
Wir wollen überprüfen, dass wenn logstash abgeschaltet wird und weiterhin Daten in die Dateien geschrieben werden, wir beim Start von logstash diese Daten erhalten.

Für das Experiment fügen wir eine weitere Zeile in die docker-compose.yml ein, indem wir das Verzeichnis öffnen, in das wir die Dateien legen.

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

Und wir ändern den Input-Bereich in habr_pipeline.conf

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

Wir starten:

docker-compose up

Um Log-Dateien zu erstellen und zu beschreiben, werden wir den Befehl verwenden:

echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hallo 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 | }

Ja, es funktioniert!

Dabei sehen wir, dass das Feld path automatisch hinzugefügt wurde. Das bedeutet, dass wir später nach diesem Feld filtern können.

Versuchen wir es noch einmal:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hallo 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 | }

Und jetzt in eine andere Datei:

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hallo 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 | }

Ausgezeichnet! Die Datei wurde erfasst, der path wurde korrekt angegeben, alles gut.

Lassen Sie logstash stoppen und neu starten. Warten wir. Stille. Das heißt, wir erhalten diese Einträge nicht erneut.

Und jetzt das gewagteste Experiment.

Wir stoppen logstash und führen aus:

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

Wir starten logstash erneut und sehen:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hallo 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" => "Hallo 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! Alles wurde erfasst.

Aber ich muss vor folgendem warnen. Wenn der logstash-Container gelöscht wird (docker stop logstash_one_channel && docker rm logstash_one_channel), dann wird nichts erfasst. Innerhalb des Containers wurde die Position der Datei gespeichert, bis zu der sie gelesen wurde. Wenn man „von null“ startet, nimmt er nur neue Zeilen an.

Das Lesen bereits existierender Dateien

Angenommen, wir starten logstash zum ersten Mal, haben aber bereits Protokolle und möchten diese verarbeiten.
Wenn wir logstash mit dem oben verwendeten input-Abschnitt starten, erhalten wir nichts. Nur neue Zeilen werden von logstash bearbeitet.

Um Zeilen aus bestehenden Dateien zu ziehen, sollte in den input-Bereich eine zusätzliche Zeile hinzugefügt werden:

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

Es gibt jedoch einen Trick: Dies gilt nur für neue Dateien, die logstash noch nicht gesehen hat. Bei den Dateien, die logstash bereits gesehen hat, hat es bereits deren Größe gespeichert und wird nun nur neue Einträge daraus übernehmen.

Lassen Sie uns an dieser Stelle die input-Sektion studieren. Es gibt noch viele weitere Varianten, aber für unsere weiteren Experimente reicht dies vorerst.

Routinen und Datenverarbeitung

Lassen Sie uns die folgende Aufgabe lösen: Angenommen, wir erhalten Nachrichten aus einem Kanal, von denen einige informativ sind und andere Fehlermeldungen. Sie unterscheiden sich durch das Tag. Einige sind INFO, andere ERROR.

Wir müssen sie am Ausgang trennen. D.h. Informative Nachrichten schreiben wir in einen Kanal, die Fehlermeldungen in einen anderen.

Dafür wechseln wir vom input-Bereich zu filter und output.

Mit der filter-Sektion werden wir die eingehende Nachricht analysieren und aus ihr ein Hash (Schlüssel-Wert-Paare) erzeugen, mit dem wir dann arbeiten können, d.h. nach Bedingungen filtern. Und im output-Bereich wählen wir die Nachrichten aus und senden jede in ihren eigenen Kanal.

Nachrichtenanalyse mit grok

Um Textzeilen zu analysieren und aus ihnen eine Reihe von Feldern zu erhalten, gibt es in der filter-Sektion ein spezielles Plugin – grok.

Ohne den Anspruch, hier eine detaillierte Beschreibung zu geben (dafür verweisen ich auf offiziellen Dokumentation), gebe ich mein einfaches Beispiel an.

Dafür müssen wir uns über das Format der Eingabezeilen klar werden. Bei mir sehen sie so aus:

1 INFO message1
2 ERROR message2

Das heißt, die Kennung steht an erster Stelle, dann folgt INFO/ERROR, dann ein Wort ohne Leerzeichen.
Nicht schwer, aber um das Prinzip zu verstehen, reicht es.

Also müssen wir in der filter-Sektion im grok-Plugin ein Muster zum Analysieren unserer Zeilen definieren.

Es wird so aussehen:

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

Im Grunde handelt es sich um einen regulären Ausdruck. Es werden bereits vorhandene Muster verwendet, wie INT, LOGLEVEL, WORD. Ihre Beschreibung sowie andere Muster können hier angesehen werden. hier

Nun wird unsere Zeile beim Durchlaufen dieses Filters in ein Hash aus drei Feldern umgewandelt: message_id, message_type, message_text.

Diese werden in der output-Sektion ausgegeben.

Nachrichtenrouting im Abschnitt output mit dem Befehl if

Im Abschnitt output wollten wir, wie wir uns erinnern, die Nachrichten in zwei Streams unterteilen. Die einen – die iNFO, werden wir auf die Konsole ausgeben, die Fehler werden wir in eine Datei schreiben.

Wie können wir diese Nachrichten aufteilen? Die Aufgabenstellung weist bereits auf die Lösung hin – wir haben das hervorgehobene Feld message_type, das nur zwei Werte annehmen kann: INFO und ERROR. Genau danach werden wir mit dem if-Operator auswählen.

if [message_type] == "ERROR" {
        # Hier geben wir in die Datei aus
       } else
     {
      # Hier geben wir in stdout aus
    }

Die Beschreibung der Arbeit mit Feldern und Operatoren kann in diesem Abschnitt angesehen werden des offiziellen Handbuchs.

Jetzt zum eigentlichen Output.

Die Ausgabe auf der Konsole ist klar – stdout {}

Die Ausgabe in eine Datei – erinnern wir uns, dass wir das alles aus einem Container heraus starten und damit die Datei, in die wir das Ergebnis schreiben, von außen zugänglich ist, müssen wir dieses Verzeichnis in der docker-compose.yml öffnen.

Insgesamt:

Der Abschnitt output unserer Datei sieht so aus:

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

In der docker-compose.yml fügen wir ein weiteres Volume für die Ausgabe hinzu:

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

Wir starten, probieren es aus und sehen die Unterteilung in zwei Streams.

Quelle: habr.com