🥇ELK und Exchange zusammenbringen. Teil 2

Ich setze meine Erzählung fort, wie man Exchange und ELK kombiniert (Teil 1) hier). Ich erinnere daran, dass diese Kombination ohne Probleme eine sehr große Menge an Logs verarbeiten kann. Diesmal werden wir darüber sprechen, wie man die Zusammenarbeit von Exchange mit den Komponenten Logstash und Kibana einrichtet.

Logstash im ELK-Stack wird für die intelligente Verarbeitung von Logs und deren Vorbereitung zur Ablage in Elastic in Form von Dokumenten verwendet, die sich bequem für verschiedene Visualisierungen in Kibana nutzen lassen.

Installation von

Es besteht aus zwei Schritten:

Installation und Konfiguration des OpenJDK-Pakets.
Installation und Konfiguration des Logstash-Pakets.

Installation und Konfiguration des OpenJDK-Pakets

Das OpenJDK-Paket muss heruntergeladen und in ein bestimmtes Verzeichnis entpackt werden. Anschließend muss der Pfad zu diesem Verzeichnis in die Umgebungsvariablen $env:Path und $env:JAVA_HOME des Windows-Betriebssystems eingetragen werden:

Überprüfen wir die Java-Version:

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, mixed mode, sharing)

Installation und Konfiguration des Logstash-Pakets

Laden Sie die Archivdatei mit der Logstash-Distribution herunter hier herunter. Das Archiv sollte im Wurzelverzeichnis des Laufwerks entpackt werden. Entpacken Sie es nicht in den Ordner C:Program Files , da Logstash sonst nicht ordnungsgemäß gestartet werden kann. Anschließend müssen Sie die Datei bearbeiten jvm.options Einstellungen, die für die Zuweisung von Arbeitsspeicher an den Java-Prozess verantwortlich sind. Ich empfehle, die Hälfte des Arbeitsspeichers des Servers anzugeben. Bei 16 GB RAM wären die Standardwerte:

-Xms1g
-Xmx1g

sollten ersetzt werden durch:

-Xms8g
-Xmx8g

Außerdem ist es sinnvoll, die Zeile zu kommentieren -XX:+UseConcMarkSweepGC. Weitere Informationen dazu hier. Der nächste Schritt ist die Erstellung einer Standardkonfiguration in der Datei logstash.conf:

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

Mit dieser Konfiguration liest Logstash Daten von der Konsole, leitet sie durch einen leeren Filter und gibt sie zurück an die Konsole. Die Anwendung dieser Konfiguration ermöglicht es, die Funktionsfähigkeit von Logstash zu überprüfen. Dafür starten wir es im interaktiven Modus:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline gestartet {"pipeline.id"=>"main"}
Das stdin-Plugin wartet jetzt auf Eingaben:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines laufen {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Logstash-API-Endpunkt erfolgreich gestartet {:port=>9600}

Logstash wurde erfolgreich auf Port 9600 gestartet.

Der letzte Schritt der Installation: Start von Logstash als Windows-Dienst. Dies kann beispielsweise mit dem Paket NSSM:

PS C:...bin> .nssm.exe install logstash
Dienst "logstash" wurde erfolgreich installiert!

Ausfallsicherheit

Die Sicherheit der Logs bei der Übertragung vom Quellserver wird durch den Mechanismus der persistenten Warteschlangen gewährleistet.

Wie es funktioniert

Schema der Anordnung der Warteschlangen im Logverarbeitungsprozess: input → queue → filter + output.

Das Input-Plugin empfängt Daten von der Logquelle, speichert sie in der Warteschlange und sendet eine Bestätigung der Datenerfassung an die Quelle.

Nachrichten aus der Warteschlange werden von Logstash verarbeitet, gehen durch den Filter und das Output-Plugin. Nach Erhalt der Bestätigung vom Output über den Versand des Logs entfernt Logstash das bearbeitete Log aus der Warteschlange. Wenn Logstash stoppt, bleiben alle nicht verarbeiteten Nachrichten und Nachrichten, für die keine Bestätigung über den Versand erhalten wurde, in der Warteschlange und Logstash setzt deren Verarbeitung beim nächsten Start fort.

Konfiguration

Steuerung erfolgt über Schlüssel in der Datei C:Logstashconfiglogstash.yml:

queue.type: (mögliche Werte — persisted und memory (Standard)).
path.queue: (Pfad zum Verzeichnis mit Warteschlangendateien, die standardmäßig in C:Logstashqueue gespeichert sind).
queue.page_capacity: (maximale Größe der Warteschlange, Standardwert — 64 MB).
queue.drain: (true/false — aktiviert/deaktiviert das Anhalten der Verarbeitung der Warteschlange vor dem Herunterfahren von Logstash. Es wird nicht empfohlen, dies zu aktivieren, da es die Herunterfahrgeschwindigkeit des Servers direkt beeinflusst).
queue.max_events: (maximale Anzahl von Ereignissen in der Warteschlange, standardmäßig — 0 (ohne Begrenzung)).
queue.max_bytes: (maximale Größe der Warteschlange in Bytes, standardmäßig — 1024mb (1gb)).

Wenn konfiguriert queue.max_events und queue.max_bytes, dann werden Nachrichten in die Warteschlange nicht mehr aufgenommen, wenn einer dieser Werte erreicht wird. Weitere Informationen zu Persistent Queues finden Sie hier.

Beispiel für einen Teil von logstash.yml, der für die Warteschlangen-Konfiguration zuständig ist:

queue.type: persisted
queue.max_bytes: 10gb

Konfiguration

Die Konfiguration von Logstash besteht normalerweise aus drei Teilen, die für verschiedene Phasen der Verarbeitung eingehender Protokolle zuständig sind: Empfang (Input-Bereich), Parsing (Filter-Bereich) und Versand an Elastic (Output-Bereich). Unten betrachten wir jede dieser Phasen genauer.

Input

Der eingehende Strom mit Rohprotokollen wird von den Filebeat-Agenten empfangen. Genau dieses Plugin geben wir im Input-Bereich an:

input {
  beats {
    port => 5044
  }
}

Nach dieser Konfiguration beginnt Logstash, Port 5044 zu überwachen, und verarbeitet die empfangenen Protokolle gemäß den Einstellungen im Abschnitt filter. Bei Bedarf kann der Empfangskanal von filebit in SSL verschlüsselt werden. Weitere Informationen zu den Einstellungen des Beats-Plugins finden Sie hier. hier.

Filter

Alle interessanten Textprotokolle, die von Exchange generiert werden, haben das CSV-Format mit den in der Protokolldatei beschriebenen Feldern. Zum Parsen der CSV-Einträge bietet Logstash drei Plugins an: dissect, csv und grok. Das erste ist das schnellste, kann jedoch nur die einfachsten Protokolle parsen.
Zum Beispiel wird der folgende Eintrag aufgrund des Vorhandenseins eines Kommas im Feld in zwei Teile aufgeteilt, was zu einem falschen Parsing des Protokolls führt:

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

Es kann beispielsweise beim Parsen von IIS-Protokollen verwendet werden. In diesem Fall könnte der filter-Bereich wie folgt aussehen:

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

Die Konfiguration von Logstash ermöglicht die Verwendung von Bedingungsoperatoren, sodass wir im Dissect-Plugin nur die Logs überwachen können, die mit dem Filebeat-Tag gekennzeichnet sind IIS. Innerhalb des Plugins ordnen wir die Feldwerte ihren Namen zu, entfernen das ursprüngliche Feld message, das den Eintrag aus dem Log enthielt, und können ein beliebiges Feld hinzufügen, das beispielsweise den Namen der Anwendung enthält, aus der wir die Logs sammeln.

Bei Tracking-Logs ist es besser, das CSV-Plugin zu verwenden, da es komplexe Felder korrekt verarbeiten kann:

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

Innerhalb des Plugins ordnen wir die Feldwerte ihren Namen zu, entfernen das ursprüngliche Feld message (und auch die Felder tenant-id und schema-version), das einen Logeintrag enthielt, und wir können ein beliebiges Feld hinzufügen, das beispielsweise den Namen der Anwendung enthält, aus der wir die Logs sammeln.

Aus der Filterphase erhalten wir Dokumente, die im ersten Ansatz für die Visualisierung in Kibana bereit sind. Uns wird Folgendes fehlen:

Numerische Felder werden als Text erkannt, was bedeutet, dass wir keine Operationen damit durchführen können. Insbesondere beziehen sich die Felder time-taken der IIS-Logs sowie die Felder recipient-count und total-bites der Tracking-Logs.
Der Standard-Zeitstempel des Dokuments wird die Zeit der Logverarbeitung enthalten, nicht die Zeit, zu der er auf der Serverseite aufgezeichnet wurde.
Feld recipient-address wird in einer einzigen Zeile erscheinen, was eine Analyse mit Zählungen der Empfänger von E-Mails nicht ermöglicht.

Es ist Zeit, ein wenig Magie in den Logverarbeitungsprozess einzufügen.

Konvertierung von numerischen Feldern

Das Dissect-Plugin hat eine Option convert_datatype, die verwendet werden kann, um ein Textfeld in ein numerisches Format zu konvertieren. Zum Beispiel so:

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

Es ist wichtig zu beachten, dass diese Methode nur geeignet ist, wenn das Feld definitiv eine Zeichenfolge enthält. Null-Werte aus den Feldern werden von dieser Option nicht verarbeitet und führen zu einer Ausnahme.

Für die Tracking-Logs sollte diese convert-Methode besser nicht verwendet werden, da die Felder recipient-count und total-bites leer sein können. Zur Konvertierung dieser Felder ist es besser, das Plugin mutate:

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

Die recipient_address in einzelne Empfänger aufteilen

Dieses Problem kann ebenfalls mit dem Plugin mutate gelöst werden:

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

Den Zeitstempel ändern

Bei den Tracking-Logs kann dieses Problem sehr einfach mit dem Plugin dategelöst werden, das dabei hilft, in das Feld timestamp das Datum und die Uhrzeit im gewünschten Format aus dem Feld date-time:

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

Bei den IIS-Logs müssen wir die Daten der Felder date und time mit dem Plugin mutate zusammenführen, die benötigte Zeitzone festlegen und diesen Zeitstempel in timestamp mit Hilfe des Plugins date einfügen:

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

Ausgabe

Der Abschnitt output wird verwendet, um bearbeitete Protokolle an den Protokollempfänger zu senden. Bei der direkten Sendung an Elasticsearch wird ein Plugin verwendet, elasticsearchin dem die Serveradresse und das Indexnamensschema für das gesendete Dokument angegeben werden:

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

Endkonfiguration

Die Endkonfiguration sieht wie folgt aus:

input {
  beats {
    port => 5044
  }
}
 
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" }
      convert_datatype => { "time-taken" => "int" }
    }
    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" ]
    }
  }
  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" }
    }
    mutate {
      convert => [ "total-bytes", "integer" ]
      convert => [ "recipient-count", "integer" ]
      split => ["recipient_address", ";"]
    }
    date {
      match => [ "date-time", "ISO8601" ]
      timezone => "Europe/Moscow"
      remove_field => [ "date-time" ]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Nützliche Links: