Jatkan tarinaani kuinka saada ystäviä Exchangesta ja ELK:sta (alku ). Haluan muistuttaa, että tämä yhdistelmä pystyy käsittelemään erittäin suuren määrän tukkeja epäröimättä. Tällä kertaa puhumme siitä, kuinka Exchange saadaan toimimaan Logstash- ja Kibana-komponenttien kanssa.
ELK-pinossa olevalla Logstashilla tukkeja käsitellään älykkäästi ja valmistetaan sijoitettaviksi Elasticiin dokumenttien muodossa, joiden perusteella Kibanaan on kätevä rakentaa erilaisia visualisointeja.
Asennus
Koostuu kahdesta vaiheesta:
- OpenJDK-paketin asennus ja konfigurointi.
- Logstash-paketin asennus ja konfigurointi.
OpenJDK-paketin asennus ja konfigurointi
OpenJDK-paketti on ladattava ja purettava tiettyyn hakemistoon. Sitten polku tähän hakemistoon on syötettävä Windows-käyttöjärjestelmän muuttujiin $env:Path ja $env:JAVA_HOME:
Katsotaanpa Java-versiota:
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)
Logstash-paketin asennus ja konfigurointi
Lataa arkistotiedosto Logstash-jakelulla . Arkisto on purettava levyn juureen asti. Pura kansioon C:Program Files Se ei ole sen arvoista, Logstash kieltäytyy käynnistymästä normaalisti. Sitten sinun on syötettävä tiedostoon jvm.options korjaukset, jotka vastaavat RAM-muistin varaamisesta Java-prosessille. Suosittelen määrittämään puolet palvelimen RAM-muistista. Jos siinä on 16 Gt RAM-muistia, oletusavaimet ovat:
-Xms1g
-Xmx1g
on korvattava seuraavalla:
-Xms8g
-Xmx8g
Lisäksi on suositeltavaa kommentoida riviä -XX:+UseConcMarkSweepGC. Tästä lisää . Seuraava vaihe on oletusmääritysten luominen logstash.conf-tiedostoon:
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
Tällä kokoonpanolla Logstash lukee tiedot konsolista, siirtää ne tyhjän suodattimen läpi ja tulostaa ne takaisin konsoliin. Tämän kokoonpanon käyttäminen testaa Logstashin toimivuuden. Suorita se tätä varten interaktiivisessa tilassa:
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 käynnistyi onnistuneesti portissa 9600.
Viimeinen asennusvaihe: käynnistä Logstash Windows-palveluna. Tämä voidaan tehdä esimerkiksi paketin avulla :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
vikasietoisuus
Lokien turvallisuus lähdepalvelimelta siirrettynä varmistetaan Persistent Queues -mekanismilla.
Miten se toimii
Jonojen asettelu lokin käsittelyn aikana on: tulo → jono → suodatin + lähtö.
Tuloliitännäinen vastaanottaa tiedot lokilähteestä, kirjoittaa sen jonoon ja lähettää vahvistuksen siitä, että tiedot on vastaanotettu lähteelle.
Logstash käsittelee jonosta tulevat viestit, ja ne kulkevat suodattimen ja tulosteen läpi. Saatuaan lähdöstä vahvistuksen lokin lähettämisestä Logstash poistaa käsitellyn lokin jonosta. Jos Logstash pysähtyy, kaikki käsittelemättömät viestit ja viestit, joille ei ole saatu vahvistusta, jäävät jonoon, ja Logstash jatkaa niiden käsittelyä seuraavan käynnistyksen yhteydessä.
säätö
Säädettävissä tiedostossa olevilla avaimilla C:Logstashconfiglogstash.yml:
queue.type: (mahdolliset arvot -persistedиmemory (default)).path.queue: (polku kansioon jonotiedostoilla, jotka on oletuksena tallennettu C:Logstashqueueen).queue.page_capacity: (jonon sivun enimmäiskoko, oletusarvo on 64 Mt).queue.drain: (true/false - ottaa käyttöön/poistaa jonon käsittelyn pysäyttämisen ennen Logstashin sammuttamista. En suosittele sen ottamista käyttöön, koska se vaikuttaa suoraan palvelimen sammutusnopeuteen).queue.max_events: (enimmäismäärä tapahtumia jonossa, oletusarvo on 0 (rajoittamaton)).queue.max_bytes: (jonon enimmäiskoko tavuina, oletusarvo - 1024 Mt (1 Gt)).
Jos määritetty queue.max_events и queue.max_bytes, viestejä ei enää hyväksytä jonoon, kun jokin näistä asetuksista saavutetaan. Lisätietoja pysyvistä jonoista .
Esimerkki logstash.yml:n osasta, joka vastaa jonon asettamisesta:
queue.type: persisted
queue.max_bytes: 10gb
säätö
Logstash-kokoonpano koostuu yleensä kolmesta osasta, jotka vastaavat saapuvien lokien käsittelyn eri vaiheista: vastaanotto (syöttöosa), jäsentäminen (suodatinosa) ja lähettäminen Elasticille (tulostusosa). Alla tarkastelemme jokaista niistä lähemmin.
panos
Vastaanotamme saapuvan streamin raakalokeineen filebeat-agenteista. Ilmoitamme syöttöosiossa tämän laajennuksen:
input {
beats {
port => 5044
}
}
Tämän määrityksen jälkeen Logstash alkaa kuunnella porttia 5044 ja vastaanottaessaan lokeja käsittelee niitä suodatinosion asetusten mukaisesti. Tarvittaessa voit kääriä kanavan lokien vastaanottamista varten filebitistä SSL:ään. Lue lisää beats-laajennuksen asetuksista .
Suodattaa
Kaikki Exchangen luomat käsittelyyn kiinnostavat tekstilokit ovat csv-muodossa ja niissä on itse lokitiedostossa kuvatut kentät. Logstash tarjoaa meille kolme lisäosaa csv-tietueiden jäsentämiseen: , csv ja grok. Ensimmäinen on eniten , mutta selviää vain yksinkertaisimpien lokien jäsentämisestä.
Se esimerkiksi jakaa seuraavan tietueen kahteen osaan (kentän sisällä olevan pilkun vuoksi), minkä vuoksi loki jäsennetään väärin:
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…
Sitä voidaan käyttää jäsennettäessä lokeja, esimerkiksi IIS. Tässä tapauksessa suodatinosio voi näyttää tältä:
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" }
}
}
}
Logstash-kokoonpanon avulla voit käyttää , joten voimme lähettää dissect-laajennukseen vain lokit, jotka on merkitty filebeat-tunnisteella IIS. Liitännäisen sisällä yhdistämme kenttien arvot niiden nimiin, poista alkuperäinen kenttä message, joka sisälsi merkinnän lokista, ja voimme lisätä mukautetun kentän, joka sisältää esimerkiksi sen sovelluksen nimen, josta keräämme lokeja.
Seurantalokien tapauksessa on parempi käyttää csv-laajennusta, joka voi käsitellä monimutkaisia kenttiä oikein:
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" }
}
}
Liitännäisen sisällä sovitamme kenttien arvot niiden nimiin, poista alkuperäinen kenttä message (ja myös kentät tenant-id и schema-version), joka sisälsi merkinnän lokista, ja voimme lisätä mukautetun kentän, joka sisältää esimerkiksi sen sovelluksen nimen, josta keräämme lokeja.
Suodatusvaiheesta poistuttaessa saamme asiakirjat ensiarviossa valmiina visualisoitavaksi Kibanaan. Meiltä puuttuu seuraavat:
- Numeeriset kentät tunnistetaan tekstiksi, mikä estää toiminnot niillä. Nimittäin kentät
time-takenIIS-loki sekä kentätrecipient-countиtotal-bitesLokin seuranta. - Vakiodokumentin aikaleima sisältää lokin käsittelyajan, ei sen kirjoittamisaikaa palvelinpuolella.
- Kenttä
recipient-addressnäyttää yhdeltä rakennustyömaalta, joka ei mahdollista analyysiä kirjeiden vastaanottajien laskemiseksi.
On aika lisätä hieman taikuutta lokinkäsittelyprosessiin.
Numeeristen kenttien muuntaminen
Dissect-pluginilla on vaihtoehto convert_datatype, jota voidaan käyttää tekstikentän muuntamiseen digitaaliseen muotoon. Esimerkiksi näin:
dissect {
…
convert_datatype => { "time-taken" => "int" }
…
}
On syytä muistaa, että tämä menetelmä sopii vain, jos kenttä sisältää ehdottomasti merkkijonon. Vaihtoehto ei käsittele nolla-arvoja kentistä ja antaa poikkeuksen.
Lokien seurantaan on parempi olla käyttämättä samanlaista muunnosmenetelmää, koska kentät recipient-count и total-bites voi olla tyhjä. Näiden kenttien muuntamiseen on parempi käyttää laajennusta :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Jaetaan vastaanottajan_osoite yksittäisiksi vastaanottajiksi
Tämä ongelma voidaan ratkaista myös käyttämällä mutate-laajennusta:
mutate {
split => ["recipient_address", ";"]
}
Aikaleiman vaihtaminen
Seurantalokien tapauksessa laajennus ratkaisee ongelman erittäin helposti , joka auttaa sinua kirjoittamaan kenttään timestamp päivämäärä ja aika vaaditussa muodossa kentästä date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
IIS-lokien tapauksessa meidän on yhdistettävä kenttätiedot date и time Mute-laajennuksella rekisteröidään tarvitsemamme aikavyöhyke ja aseta tämä aikaleima timestamp käyttämällä päivämäärälaajennusta:
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" ]
}
ulostulo
Tulostusosaa käytetään käsiteltyjen lokien lähettämiseen lokin vastaanottajalle. Jos lähetät suoraan Elasticille, käytetään laajennusta , joka määrittää palvelimen osoitteen ja indeksin nimimallin luodun asiakirjan lähettämistä varten:
output {
elasticsearch {
hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
manage_template => false
index => "Exchange-%{+YYYY.MM.dd}"
}
}
Lopullinen kokoonpano
Lopullinen kokoonpano näyttää tältä:
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}"
}
}
Hyödyllisiä linkkejä:
Lähde: will.com