Družimo ELK in Exchange. Del 2

Nadaljujem svojo zgodbo o tem, kako povezati Exchange in ELK (začetek tukaj). Spomnim, da je ta kombinacija sposobna brez težav obdelovati izjemno veliko količino logov. Tokrat bomo govorili o tem, kako vzpostaviti delovanje Exchange z komponentami Logstash in Kibana.

Logstash v skladu ELK se uporablja za inteligentno obdelavo logov in njihovo pripravo za shranjevanje v Elastic v obliki dokumentov, na podlagi katerih je enostavno graditi različne vizualizacije v Kibani.

Namestitev

Obsega dva koraka:

• Namestitev in konfiguracija paketa OpenJDK.
• Namestitev in konfiguracija paketa Logstash.

Namestitev in konfiguracija paketa OpenJDK

Paket OpenJDK je potrebno prenesti in razpakirati v določen imenik. Nato je potrebno pot do tega imenika dodati v spremenljivke $env:Path in $env:JAVA_HOME operacijskega sistema Windows:

Preverimo različico Jave:

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)

Namestitev in konfiguracija paketa Logstash

Prenesite arhiv z distribucijo Logstash od tod. Arhiv je treba razpakirati v koren diska. Ni priporočljivo razpakirati v mapo C:Program Files , saj se Logstash ne bo normalno zagnal. Nato je potrebno v datoteko jvm.options narediti spremembe, povezane z dodeljevanjem pomnilnika za proces Jave. Priporočam, da navedete polovico sistemskega pomnilnika. Če ima strežnik 16 GB RAM-a, so privzeti ključi:

-Xms1g
-Xmx1g

je treba zamenjati z:

-Xms8g
-Xmx8g

Poleg tega je smiselno zakomentirati vrstico -XX:+UseConcMarkSweepGC. Več informacij o tem tukaj. Naslednji korak je ustvarjanje privzete konfiguracije v datoteki logstash.conf:

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

S to konfiguracijo Logstash prebere podatke iz konzole, jih posreduje skozi prazen filter in jih izpiše nazaj v konzolo. Ta konfiguracija omogoča preizkus funkcionalnosti Logstash. Za to ga zaženimo v interaktivnem načinu:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"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 se je uspešno zagnal na portu 9600.

Zadnji korak namestitve: zagon Logstash kot Windows storitve. To lahko storite, na primer, s pomočjo paketa NSSM:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

Odporna zasnova

Zagotavljanje varnosti dnevnikov med prenosom z izvornega strežnika zagotavlja mehanizem Persistent Queues.

Kako deluje

Shema razporeditve vrstic v procesu obdelave dnevnikov: input → queue → filter + output.

Plugin input prejme podatke iz vira dnevnikov, jih zapiše v vrsto in pošlje viru potrdilo o prejemu podatkov.

Sporočila iz vrste obdeluje Logstash, preidejo skozi filter in plugin output. Ko prejme potrdilo o pošiljanju dnevnika iz output, Logstash odstrani obdelani dnevnik iz vrste. Če Logstash ustavi, ostanejo v vrsti vsa neobdelana sporočila in sporočila, za katera ni bilo prejeto potrdilo o pošiljanju, Logstash pa jih bo obdelal pri naslednjem zagonu.

Setup

Upravlja se s ključev v datoteki C:Logstashconfiglogstash.yml:

• queue.type: (možni vrednosti — persisted in memory (privzeto)).
• path.queue: (pot do mape s datotekami vrstic, ki se privzeto shranjujejo v C:Logstashqueue).
• queue.page_capacity: (maksimalna velikost strani vrste, privzeta vrednost — 64mb).
• queue.drain: (true/false — vključuje/izključuje zaustavitev obdelave vrste pred izklopom Logstash. Ne priporočam vključitve, ker bo to neposredno vplivalo na hitrost izklopa strežnika).
• queue.max_events: (maksimalno število dogodkov v vrsti, privzeto — 0 (ni omejeno)).
• queue.max_bytes: (maksimalna velikost vrste v bajtih, privzeto — 1024mb (1gb)).

Če so nastavljene queue.max_events in queue.max_bytes, se sporočila prenehajo sprejemati v vrsto, ko dosežejo vrednost katere koli od teh nastavitev. Več o Persistent Queues je opisano tukaj.

Primer dela logstash.yml, ki odgovarja za nastavitev vrste:

queue.type: persisted
queue.max_bytes: 10gb

Setup

Konfiguracija Logstash običajno sestavlja tri dele, ki odgovarjajo različnim fazam obdelave vhodnih dnevnikov: sprejem (sekcija input), parsanje (sekcija filter) in pošiljanje v Elastic (sekcija output). Spodaj podrobneje obravnavamo vsak od njih.

Vhod

Vhodni tok surovih dnevnikov sprejemamo od agentov filebeat. Ta plugin natančno navedemo v sekciji input:

input {
  beats {
    port => 5044
  }
}

Po taki nastavitvi Logstash začne poslušati vrata 5044, in ko prejme dnevnike, jih obdeluje v skladu z nastavitvami sekcije filter. Po potrebi lahko kanal prejemanja dnevnikov od filebit zavijemo v SSL. Več o nastavitvah plugin-a beats je zapisano tukaj.

Filter

Vsi zanimivi tekstovni dnevniki, ki jih generira Exchange, so v formatu csv s polji, ki so opisana v samem dnevniku. Za obdelavo csv zapisov Logstash ponuja tri vtičnike: dissect, csv in grok. Prvi je naj hitrejši, a obdeluje le najpreprostejše dnevnike.
Na primer, naslednji zapis bo razdelil na dve (zaradi prisotnosti vejice znotraj polja), zaradi česar bo dnevnik narobe razbran:

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

Lahko ga uporabimo pri obdelavi dnevnikov, na primer, IIS. V tem primeru lahko sekcija filter izgleda takole:

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

Konfiguracija Logstash omogoča uporabo pogojnih operatorjev, zato lahko usmerimo vtičnik dissect le na dnevnike, ki so označeni z oznako filebeat IIS. Znotraj vtičnika usklajujemo vrednosti polj z njihovimi imeni, odstranimo izvorno polje message, ki je vsebovalo zapis iz dnevnika, in lahko dodamo polje po meri, ki bo, na primer, vsebovalo ime aplikacije, iz katere zbiramo dnevnike.

Pri dnevnikih sledenja je bolje uporabiti vtičnik csv, saj zna pravilno obdelati kompleksna polja:

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

Znotraj vtičnika usklajujemo vrednosti polj z njihovimi imeni, odstranimo izvorno polje message (in tudi polja tenant-id in schema-version), ki je vsebovalo zapis iz dnevnika, in lahko dodamo polje po meri, ki bo, na primer, vsebovalo ime aplikacije, iz katere zbiramo dnevnike.

Na izhodu iz faze filtriranja bomo dobili dokumente v prvi približni obliki, primerni za vizualizacijo v Kibani. Manjkalo nam bo naslednje:

• Številska polja bodo prepoznana kot besedilo, kar onemogoča izvajanje operacij z njimi. Konkretno, polja time-taken loga IIS, prav tako pa polja recipient-count in total-bites loga Tracking.
• Standardna časovna oznaka dokumenta bo vsebovala čas obdelave loga, ne pa časa njegove zapisane strani na strežniku.
• Field recipient-address bo videti kot ena stavka, kar ne omogoča analize s preštevanjem prejemnikov e-pošt.

Prišel je čas, da dodamo malo magie v proces obdelave logov.

Pretvorba številskih polj

Vtičnik dissect ima možnost convert_datatype, ki jo je mogoče uporabiti za pretvorbo besedilnega polja v digitalni format. Na primer, tako:

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

Treba je upoštevati, da je ta metoda primerna le, če polje zagotovo vsebuje niz. Null-vrednosti iz polj ta možnost ne obdeluje in povzroči izjemo.

Za loge sledenja je podobna metoda convert bolje ne uporabljati, saj polja recipient-count in total-bites lahko ostanejo prazna. Za pretvorbo teh polj je bolje uporabiti vtičnik mutate:

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

Razdelitev recipient_address na posamezne prejemnike

To nalogo je mogoče prav tako rešiti s pomočjo vtičnika mutate:

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

Sprememba timestamp

V primeru logov sledenja se naloga zelo preprosto reši z vtičnikom date, ki bo pomagal zapisati v polje timestamp datum in čas v želenem formatu iz polja date-time:

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

V primeru logov IIS bo treba združiti podatke polj date in time s pomočjo vtičnika mutate, navesti potrebno časovno območje in umestiti to časovno oznako v timestamp s pomočjo vtičnika 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" ]
}

Izhodni

Sekcija output se uporablja za pošiljanje obdelanih logov v prejemnik logov. V primeru neposredne pošiljke v Elastic se uporablja vtičnik elasticsearch, kjer je naveden naslov strežnika in vzorec imena indeksa za pošiljanje oblikovanega dokumenta:

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

Končna konfiguracija

Končna konfiguracija bo videti takole:

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

izvorna koda

• Kako instalirati OpenJDK 11 na Windowsu?
• Preuzmite Logstash
• Elastic koristi zastarjelu opciju UseConcMarkSweepGC #36828
• NSSM
• Stalni redovi
• Beats ulazni plugin
• Logstash Dude, gdje je moja pila? Moram analizirati svoje logove
• Dissect filter plugin
• Uslovi
• Mutate filter plugin
• Date filter plugin
• Elasticsearch izlazni plugin

Izvor: habr.com