🥇Spojujeme ELK a Exchange. Část 2

Pokračuji ve svém vyprávění o tom, jak spojit Exchange a ELK (začátek tady). Připomínám, že tato kombinace je schopna bez váhání zpracovávat velmi velké množství logů. Tentokrát si povíme, jak nastavit spolupráci Exchange s komponenty Logstash a Kibana.

Logstash ve stacku ELK se používá pro inteligentní zpracování logů a jejich přípravu k umístění do Elastic jako dokumenty, na základě kterých je pohodlné vytvářet různé vizualizace v Kibana.

Instalace

Skládá se ze dvou fází:

Instalace a nastavení balíčku OpenJDK.
Instalace a nastavení balíčku Logstash.

Instalace a nastavení balíčku OpenJDK

Balíček OpenJDK je třeba stáhnout a rozbalit do určitého adresáře. Poté je nutné přidat cestu k tomuto adresáři do proměnných $env:Path a $env:JAVA_HOME operačního systému Windows:

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

Instalace a nastavení balíčku Logstash

Soubor-archiv s distribucí Logstash stáhněte odsud. Archiv je třeba rozbalit do kořene disku. Nerozbalujte do složky C:Program Files , Logstash se odmítne správně spustit. Poté je nutné provést úpravy v souboru jvm.options , které se týkají alokace paměti pro proces Java. Doporučuji nastavit polovinu fyzické paměti serveru. Pokud má k dispozici 16 GB RAM, pak výchozí klíče:

-Xms1g
-Xmx1g

je nutné nahradit za:

-Xms8g
-Xmx8g

Kromě toho je rozumné zakomentovat řádek -XX:+UseConcMarkSweepGC. Podrobnosti o tom tady. Dalším krokem je vytvoření výchozí konfigurace v souboru logstash.conf:

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

Při použití této konfigurace Logstash čte data z konzole, prochází je prázdným filtrem a vrací je zpět do konzole. Použití této konfigurace umožní zkontrolovat funkčnost Logstash. K tomu ho spustíme v interaktivním režimu:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
Plugin stdin nyní čeká na vstup:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Úspěšně spuštěn API koncový bod Logstash {:port=>9600}

Logstash byl úspěšně spuštěn na portu 9600.

Posledním krokem instalace je spuštění Logstash jako služby Windows. To lze provést například pomocí balíčku NSSM:

PS C:...bin> .nssm.exe install logstash
Služba "logstash" byla úspěšně nainstalována!

Odolnost proti výpadkům

Zachování logů při přenosu z původního serveru zajišťuje mechanismus Persistent Queues.

Jak to funguje

Schéma umístění front v procesu zpracování logů: input → queue → filter + output.

Plugin input získává data ze zdroje logů, zapisuje je do fronty a posílá zdroji potvrzení o přijetí dat.

Zprávy z fronty zpracovává Logstash, procházejí filtrem a pluginem output. Po obdržení potvrzení od output o odeslání logu Logstash odstraní zpracovaný log z fronty. Pokud se Logstash zastaví, všechny nezpracované zprávy a zprávy, pro které nebylo obdrženo potvrzení o odeslání, zůstávají ve frontě, a Logstash je zpracovává při příštím spuštění.

Nastavení

Reguluje se klíči v souboru C:Logstashconfiglogstash.yml:

queue.type: (možné hodnoty — persisted a memory (výchozí)).
path.queue: (cesta k adresáři s soubory front, které se výchozí ukládají do C:Logstashqueue).
queue.page_capacity: (maximální velikost stránky fronty, výchozí hodnota — 64mb).
queue.drain: (true/false — zapíná/vypíná zastavení zpracování fronty před vypnutím Logstash. Nedoporučuji zapínat, protože to přímo ovlivní rychlost vypnutí serveru).
queue.max_events: (maximální počet událostí ve frontě, výchozí hodnota — 0 (neomezeno)).
queue.max_bytes: (maximální velikost fronty v bajtech, výchozí hodnota — 1024mb (1gb)).

Pokud jsou nastaveny queue.max_events a queue.max_bytes, pak zprávy přestávají být přijímány do fronty při dosažení hodnoty jakéhokoli z těchto nastavení. Podrobněji o Persistent Queues se zmiňuje tady.

Příklad části logstash.yml, která se věnuje konfiguraci fronty:

queue.type: persisted
queue.max_bytes: 10gb

Nastavení

Konfigurace Logstash obvykle sestává ze tří částí, které se věnují různým fázím zpracování příchozích logů: příjem (sekce input), parsing (sekce filter) a odeslání do Elastic (sekce output). Následující část podrobněji zkoumá každou z nich.

Vstup

Příchozí tok se syrovými logy přijímáme od agentů filebeat. Přesně tento plugin uvádíme v sekci input:

input {
  beats {
    port => 5044
  }
}

Po takovém nastavení Logstash začíná poslouchat port 5044 a při obdržení logů je zpracovává podle nastavení v sekci filter. Pokud je potřeba, můžeme kanál pro příjem logů od filebit zabalit do SSL. Podrobnosti o nastavení pluginu beats jsou uvedeny. tady.

Filtr

Všechny zajímavé textové protokoly generované Exchange mají formát csv s popsanými poli v samotných protokolech. Pro analýzu csv záznamů nám Logstash nabízí tři pluginy: dissect, csv a grok. První z nich je nejrychlejší, ale dokáže analyzovat pouze nejjednodušší protokoly.Například následující záznam rozdělí na dvě části (kvůli přítomnosti čárky uvnitř pole), což způsobí nesprávnou analýzu protokolu:
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

Můžete ho použít při analýze protokolů, například, IIS. V tomto případě může sekce filtru vypadat takto:

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

Konfigurace Logstash umožňuje používat

podmínkové operátory, protože do pluginu dissect můžeme poslat pouze protokoly, které byly označeny tagem filebeatIIS. Uvnitř pluginu přiřazujeme hodnoty polí k jejich názvům, odstraňujeme původní pole,které obsahovalo záznam z protokolu, a můžeme přidat libovolné pole, které bude například obsahovat název aplikace, ze které shromažďujeme protokoly. zprávaV případě logů sledování je lepší použít plugin csv, který dokáže správně zpracovat složitá pole:

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

Uvnitř pluginu přiřazujeme hodnoty polí k jejich názvům, odstraňujeme původní pole

(a také pole zpráva tenant-id schema-version a ), které obsahovalo záznam z protokolu, a můžeme přidat libovolné pole, které bude například obsahovat název aplikace, ze které shromažďujeme protokoly.Na konci fáze filtrování získáme dokumenty, které jsou na první pohled připravené k vizualizaci v Kibana. Co nám chybí, následující:

Na konci fáze filtrování získáme dokumenty, které budou v prvním přiblížení připravené k vizualizaci v Kibana. Budeme potřebovat následující:

Číselná pole budou rozpoznána jako text, což neumožňuje provádět operace s nimi. A konkrétně pole time-taken logu IIS, jakož i pole recipient-count a total-bites logu Tracking.
Standardní časová značka dokumentu bude obsahovat čas zpracování logu, nikoli čas jeho zápisu na straně serveru.
Pole recipient-address bude vypadat jako jeden řetězec, což neumožňuje provádět analýzu s počtem příjemců e-mailů.

Nastal čas přidat do procesu zpracování logů trochu magie.

Konverze číselných polí

Plugin dissect má možnost convert_datatype, kterou je možné použít k převodu textového pole na číselný formát. Například takto:

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

Je dobré si pamatovat, že tato metoda funguje pouze v případě, že pole skutečně obsahuje řetězec. Null-hodnoty z polí tato možnost nezpracovává a vyvolává výjimku.

Pro logy trackingu je lepší stejnou metodu convert nepoužívat, protože pole recipient-count a total-bites mohou být prázdná. Pro konverzi těchto polí je lepší použít plugin mutate:

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

Rozdělení recipient_address na jednotlivé příjemce

Tento úkol lze také vyřešit pomocí pluginu mutate:

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

Úprava timestamp

V případě logů trackingu se úkol velmi jednoduše řeší pluginem date, který pomůže napsat do pole timestamp datum a čas v požadovaném formátu z pole date-time:

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

V případě logů IIS bude nutné sloučit data pole date a čas pomocí pluginu mutate, nastavit požadovanou časovou zónu a umístit tuto časovou značku do timestamp pomocí pluginu 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" ]
}

Výstup

Sekce output se používá pro odeslání zpracovaných logů do příjemce logů. V případě odeslání přímo do Elastic se používá plugin elasticsearch, ve kterém je uvedena adresa serveru a šablona názvu indexu pro odeslání vytvořeného dokumentu:

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

Závěrečná konfigurace

Konečná konfigurace bude vypadat následujícím způsobem:

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

Užitečné odkazy: