🥇Împăcăm ELK și Exchange. Partea 2

Continuu povestea despre cum să integrez Exchange și ELK (începutul aici). Reamintesc că această combinație poate gestiona fără ezitare un număr foarte mare de jurnale. De data aceasta, vom discuta despre modul în care să configurăm Exchange cu componentele Logstash și Kibana.

Logstash din stiva ELK este utilizat pentru prelucrarea inteligentă a jurnalele și pregătirea acestora pentru a fi plasate în Elastic sub formă de documente, pe baza cărora este convenabil să construiești diverse vizualizări în Kibana.

Instalare

Constă în două etape:

Instalarea și configurarea pachetului OpenJDK.
Instalarea și configurarea pachetului Logstash.

Instalarea și configurarea pachetului OpenJDK

Pachetul OpenJDK trebuie descărcat și desfăcut într-un director specific. Apoi, calea până la acest director trebuie adăugată în variabilele $env:Path și $env:JAVA_HOME ale sistemului de operare Windows:

Verificăm versiunea 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)

Instalarea și configurarea pachetului Logstash

Descarcă arhiva fișierului cu distribuitorul Logstash de aici. Arhiva trebuie desfăcută în rădăcina discului. A desfășura în folderul C:Program Files nu este recomandat, Logstash va refuza să pornească corect. Apoi, este necesar să faci modificări în fișierul jvm.options care se referă la alocarea memoriei pentru procesul Java. Recomand să specifici jumătate din memoria RAM a serverului. Dacă are 16 GB de RAM, atunci cheile din mod implicit sunt:

-Xms1g
-Xmx1g

trebuie înlocuite cu:

-Xms8g
-Xmx8g

În plus, este recomandabil să comentezi linia -XX:+UseConcMarkSweepGC. Mai multe detalii despre aceasta aici. Pasul următor — crearea configurației implicite în fișierul logstash.conf:

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

Atunci când folosești această configurație, Logstash citește date din consolă, le trece printr-un filtru gol și le redă înapoi în consolă. Aplicarea acestei configurații va permite testarea funcționalității Logstash. Pentru aceasta, îl vom porni în modul interactiv:

PS C:...bin"main"}
The stdin plugin is now waiting for input:
[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           ] Successfully started Logstash API endpoint {:port=>9600}

Logstash s-a pornit cu succes pe portul 9600.

Ultimul pas al instalării: pornirea Logstash ca serviciu Windows. Acest lucru se poate face, de exemplu, cu ajutorul pachetului NSSM:

PS C:...bin> .nssm.exe install logstash
Serviciul "logstash" a fost instalat cu succes!

Redundanță

Integritatea jurnalelor în timpul transmiterii de pe serverul sursă este asigurată de mecanismul Persistent Queues.

Cum funcționează

Schema de plasare a cozii în procesul de prelucrare a jurnalelor: input → queue → filter + output.

Pluginul input primește date de la sursa jurnalelor, le scrie în coadă și trimite o confirmare sursei despre primirea datelor.

Mesajele din coadă sunt procesate de Logstash, trec prin filtrul și pluginul output. La primirea confirmării de la output că jurnalul a fost trimis, Logstash șterge jurnalul procesat din coadă. Dacă Logstash se oprește, toate mesajele neprocesate și cele pentru care nu s-a primit confirmare de trimitere rămân în coadă, iar Logstash va continua procesarea lor la următoarea pornire.

Configurare

Se reglează prin chei în fișierul C:Logstashconfiglogstash.yml:

queue.type: (valori posibile — persisted și memory (implicit)).
path.queue: (calea către dosarul cu fișierele cozii, care implicit se află în C:Logstashqueue).
queue.page_capacity: (dimensiunea maximă a paginii cozii, valoarea implicită — 64mb).
queue.drain: (true/false — activează/dezactivează oprirea procesării cozii înainte de a opri Logstash. Nu recomand activarea, deoarece va influența direct viteza de oprire a serverului).
queue.max_events: (numărul maxim de evenimente din coadă, implicit — 0 (fără limită)).
queue.max_bytes: (dimensiunea maximă a cozii în bytes, implicit — 1024mb (1gb)).

Dacă sunt configurate queue.max_events și queue.max_bytes, mesajele nu mai sunt acceptate în coadă la atingerea valorii oricărei dintre aceste setări. Mai multe informații despre Persistent Queues sunt descrise aici.

Exemplu de parte din logstash.yml, responsabilă pentru configurarea cozii:

queue.type: persisted
queue.max_bytes: 10gb

Configurare

Configurarea Logstash constă de obicei din trei părți, responsabile pentru diferite etape de prelucrare a jurnalelor de intrare: recepție (secțiunea input), parsare (secțiunea filter) și trimitere în Elastic (secțiunea output). Mai jos, vom examina fiecare dintre ele în detaliu.

Input

Fluxul de intrare cu jurnale brute este primit de la agenții filebeat. Acest plugin este cel pe care îl specificăm în secțiunea input:

input {
  beats {
    port => 5044
  }
}

După această configurare, Logstash începe să asculte pe portul 5044, iar la primirea jurnalelor le procesează conform setărilor din secțiunea filter. Dacă este necesar, canalul de primire a jurnalelor de la filebeat poate fi criptat cu SSL. Mai multe informații despre setările pluginului beats sunt prezentate aici.

Filter

Toate jurnalele de text interesante generate de Exchange au format csv, cu câmpurile descrise în fișierul de jurnal. Pentru parsarea înregistrărilor csv, Logstash ne oferă trei pluginuri: dissect, csv și grok. Primul este cel mai rapid, dar reușește să parseze doar cele mai simple jurnale.
De exemplu, următoarea înregistrare va fi împărțită în două (datorită prezenței unei virgule în interiorul câmpului), ceea ce va duce la o parseare greșită a jurnalului:

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

Poate fi folosit la parsarea jurnalelor, de exemplu, IIS. În acest caz, secțiunea filter poate arăta astfel:

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

Configurarea Logstash permite utilizarea operațiunilor condiționale, astfel încât putem direcționa doar jurnalele etichetate cu tagul filebeat IIS. În cadrul pluginului, mapăm valorile câmpurilor cu denumirile lor, eliminăm câmpul inițial message, care conținea înregistrarea din jurnal, și putem adăuga un câmp arbitrar, care va conține, de exemplu, numele aplicației din care colectăm jurnalele.

În cazul jurnalelor de tracking, este mai bine să utilizăm pluginul csv, care poate gestiona corect câmpurile complexe:

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

În cadrul pluginului, mapăm valorile câmpurilor cu denumirile lor, eliminăm câmpul inițial message (și câmpurile tenant-id și schema-version), care conținea înregistrarea din jurnal, și putem adăuga un câmp arbitrar, care va conține, de exemplu, numele aplicației din care colectăm jurnalele.

La ieșirea din etapa de filtrare, vom obține documente care sunt, în prima aproximare, gata de vizualizare în Kibana. Ne vor lipsi următoarele:

Câmpurile numerice vor fi recunoscute ca text, ceea ce nu permite realizarea operațiunilor asupra lor. Adică, câmpurile time-taken din jurnalul IIS, precum și câmpurile recipient-count și total-bites din jurnalul Tracking.
Stampila de timp standard a documentului va conține timpul de procesare a jurnalului și nu timpul înregistrării acestuia pe partea serverului.
Câmp recipient-address va arăta ca o singură construcție, ceea ce nu permite analiza prin numărarea recepționerilor de e-mailuri.

A venit timpul să adăugăm puțină magie procesului de procesare a jurnalelor.

Conversia câmpurilor numerice

Pluginul dissect are o opțiune convert_datatype, pe care o poți folosi pentru a converti un câmp text într-un format numeric. De exemplu, astfel:

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

Trebuie să reții că această metodă este potrivită doar dacă câmpul va conține cu certitudine un șir. Valorile Null din câmpuri nu sunt procesate de opțiune și generează o excepție.

Pentru jurnalele de tracking, metoda similară de conversie ar trebui evitată, deoarece câmpurile recipient-count și total-bites pot fi goale. Pentru conversia acestor câmpuri, este mai bine să folosești pluginul mutate:

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

Împărțirea recipient_address în destinatari individuali

Această problemă poate fi, de asemenea, rezolvată cu pluginul mutate:

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

Modificarea stampilei de timp

În cazul jurnalele de tracking, problema este foarte simplu de rezolvat cu pluginul date, care va ajuta să scrii în câmpul timestamp data și ora în formatul dorit din câmpul date-time:

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

În cazul jurnalele IIS, va fi necesar să combinăm datele câmpurilor date și time cu ajutorul pluginului mutate, să definim fusul orar dorit și să plasăm această stampilă de timp în timestamp cu ajutorul pluginului 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" ]
}

Registrul direcției, registrul de ieșire. În acest proiect, nu ne vor fi necesare.

Secțiunea output este folosită pentru a trimite jurnalele procesate către receptorul de jurnale. În cazul trimiterii direct în Elastic, se folosește pluginul elasticsearch, în care se specifică adresa serverului și modelul numelui indexului pentru a trimite documentul formatat:

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

Configurația finală

Configurarea finală va arăta astfel:

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

Linkuri utile: