Continuu a mo storia nantu à cumu fà amici Exchange è ELK (iniziu ). Lasciami ricurdà chì sta cumminazione hè capace di processà un gran numaru di logs senza esitazione. Sta volta avemu Mulateri Di L'parlerà di quantu à ottene u scambiu u travagliu cù Logstash è cumpunenti Kibana.
Logstash in a pila ELK hè utilizatu per processà in modu intelligente i logs è preparalli per u piazzamentu in Elastic in forma di documenti, nantu à a basa di quale hè cunvenutu per custruisce diverse visualizazioni in Kibana.
rimarchevuli
Hè custituitu di dui fasi:
- Stallà è cunfigurà u pacchettu OpenJDK.
- Stallà è cunfigurà u pacchettu Logstash.
Stallà è cunfigurà u pacchettu OpenJDK
U pacchettu OpenJDK deve esse scaricatu è unpacked in un cartulare specificu. Allora u percorsu à questu repertoriu deve esse inseritu in e $env:Path è $env:JAVA_HOME variabili di u sistema operatore Windows:
Cuntrollamu a versione di 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)
Stallà è cunfigurà u pacchettu Logstash
Scaricate l'archiviu cù a distribuzione Logstash . L'archiviu deve esse unpacked à a radica di u discu. Unpack à u cartulare C:Program Files Ùn vale a pena, Logstash ricusarà di principià normalmente. Allora vi tocca à entre in u schedariu jvm.options correzioni rispunsevuli di assignà RAM per u prucessu Java. Vi cunsigliu di specificà a mità di a RAM di u servitore. S'ellu hà 16 GB di RAM à bordu, allora i chjavi predeterminati sò:
-Xms1g
-Xmx1g
deve esse rimpiazzatu cù:
-Xms8g
-Xmx8g
Inoltre, hè cunsigliu di cummentà a linea -XX:+UseConcMarkSweepGC. Più nantu à questu . U prossimu passu hè di creà una cunfigurazione predeterminata in u schedariu logstash.conf:
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
Cù sta cunfigurazione, Logstash leghje e dati da a cunsola, u passa per un filtru viotu, è i rimette à a cunsola. Utilizà sta cunfigurazione pruverà a funziunalità di Logstash. Per fà questu, eseguimu in modu interattivu:
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 hà lanciatu cù successu in u portu 9600.
U passu finali di a stallazione: lanciate Logstash cum'è un serviziu Windows. Questu pò esse fattu, per esempiu, cù u pacchettu :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
tolleranza à i difetti
A sicurità di i logs quandu sò trasferiti da u servitore fonte hè assicurata da u mecanismu di fila persistenti.
Cumu travagliu
U layout di fila durante u processu di log hè: input → queue → filter + output.
U plugin di input riceve dati da una fonte di log, scrive in una fila, è manda cunferma chì i dati sò stati ricevuti à a fonte.
I missaghji da a fila sò processati da Logstash, passati per u filtru è u plugin di output. Quandu riceve cunferma da a pruduzzioni chì u logu hè statu mandatu, Logstash elimina u logu processatu da a fila. Se Logstash si ferma, tutti i missaghji è i missaghji micca processati per i quali nisuna cunferma hè stata ricevuta restanu in a fila, è Logstash continuarà à processà a prossima volta chì principia.
cutter
Ajustable da e chjave in u schedariu C:Logstashconfiglogstash.yml:
queue.type: (valori pussibuli -persistedиmemory (default)).path.queue: (percorsu à u cartulare cù i schedarii di fila, chì sò almacenati in C: Logstashqueue per difettu).queue.page_capacity: (dimensione massima di a pagina di fila, u valore predeterminatu hè 64 MB).queue.drain: (true / false - attiva / disattiva a stopping processing fila prima di chjude Logstash. Ùn ricumandemu micca di attivà, perchè questu affettarà direttamente a velocità di chjusu di u servitore).queue.max_events: (numaru massimu di avvenimenti in a fila, predeterminatu hè 0 (illimitatu)).queue.max_bytes: (taglia massima di fila in byte, predefinitu - 1024mb (1gb)).
Se cunfiguratu queue.max_events и queue.max_bytes, allura i missaghji ùn sò più accettati in a fila quandu u valore di qualsiasi di sti paràmetri hè righjuntu. Sapete di più nantu à e file persistenti .
Un esempiu di a parte di logstash.yml rispunsevuli di stallà a fila:
queue.type: persisted
queue.max_bytes: 10gb
cutter
A cunfigurazione di Logstash hè generalmente custituita da trè parti, rispunsevuli di e diverse fasi di trasfurmazioni di logs in entrata: riceve (sezzione di input), parsing (sezzione di filtru) è mandà à Elastic (sezzione di output). Sottu avemu da piglià un ochju più vicinu à ognunu di elli.
entrée
Ricevemu u flussu in entrata cù logs crudi da l'agenti filebeat. Hè stu plugin chì indichemu in a sezione di input:
input {
beats {
port => 5044
}
}
Dopu à sta cunfigurazione, Logstash principia à sente u portu 5044, è quandu riceve logs, li processa secondu i paràmetri di a sezione di filtru. In casu di necessariu, pudete chjappà u canali per riceve logs da filebit in SSL. Leghjite più nantu à i paràmetri di u plugin beats .
Filter
Tutti i logs di testu chì sò interessanti per u processu chì Exchange genera sò in formatu csv cù i campi descritti in u schedariu di logu stessu. Per l'analisi di record csv, Logstash ci offre trè plugins: , csv è grok. U primu hè u più , ma copes with parsing only the simple logs.
Per esempiu, dividerà u seguente record in dui (per via di a prisenza di una virgola in u campu), per quessa chì u logu serà analizatu incorrectamente:
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…
Pò esse usatu quandu analizà i logs, per esempiu, IIS. In questu casu, a seccione di filtru pò esse cusì:
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" }
}
}
}
A cunfigurazione di Logstash vi permette di utilizà , cusì pudemu mandà solu logs chì sò stati marcati cù u tag filebeat à u plugin dissect IIS. Dentru u plugin currispondemu i valori di u campu cù i so nomi, sguassate u campu originale message, chì cuntene una entrata da u logu, è pudemu aghjunghje un campu persunalizatu chì, per esempiu, cuntene u nome di l'appiecazione da quale cogliemu logs.
In u casu di logs di seguimentu, hè megliu aduprà u plugin csv; pò processà currettamente i campi cumplessi:
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" }
}
}
Dentru u plugin currispondemu i valori di u campu cù i so nomi, sguassate u campu originale message (è ancu campi tenant-id и schema-version), chì cuntene una entrata da u logu, è pudemu aghjunghje un campu persunalizatu, chì, per esempiu, cuntene u nome di l'appiecazione da quale cogliemu logs.
À l'uscita da a tappa di filtrazione, riceveremu documenti in una prima approssimazione, pronti per a visualizazione in Kibana. Ci mancherà i seguenti:
- I campi numerichi seranu ricunnisciuti cum'è testu, chì impedisce l'operazione nantu à elli. Vale à dì i campi
time-takenLog IIS, è ancu campirecipient-countиtotal-bitesLog Tracking. - U timestamp standard di u documentu cuntene u tempu chì u logu hè statu processatu, micca u tempu chì hè statu scrittu da u servitore.
- chjosu
recipient-addressSerà cum'è un situ di custruzzione, chì ùn permette micca l'analisi per cuntà i destinatari di e lettere.
Hè ora di aghjunghje un pocu di magia à u prucessu di trasfurmazioni di log.
Cunvertisce campi numerichi
U plugin dissect hà una opzione convert_datatype, chì pò esse usatu per cunvertisce un campu di testu à un furmatu digitale. Per esempiu, cusì:
dissect {
…
convert_datatype => { "time-taken" => "int" }
…
}
Hè vale a pena ricurdà chì stu metudu hè adattatu solu se u campu certamenti cuntene una stringa. L'opzione ùn processa micca i valori nulli da i campi è lancia una eccezzioni.
Per tracking logs, hè megliu micca aduprà un metudu di cunversione simili, postu chì i campi recipient-count и total-bites pò esse viotu. Per cunvertisce questi campi hè megliu aduprà un plugin :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Spliting recipient_address in destinatari individuali
Stu prublema pò ancu esse risoltu cù u plugin mutate:
mutate {
split => ["recipient_address", ";"]
}
Cambia u timestamp
In u casu di logs di seguimentu, u prublema hè facilmente risoltu da u plugin , chì vi aiuterà à scrive in u campu timestamp data è ora in u formatu necessariu da u campu date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
In u casu di logs IIS, avemu bisognu di cumminà i dati di u campu date и time aduprendu u plugin mutante, registrate u fusu orariu chì avemu bisognu è mette stu stampu di tempu timestamp usendu u plugin di data:
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" ]
}
radicali avrìanu pututu
A sezione di output hè aduprata per mandà logs processati à u ricevitore di log. In casu di mandà direttamente à Elastic, un plugin hè utilizatu , chì specifica l'indirizzu di u servitore è u mudellu di nome d'indici per mandà u documentu generatu:
output {
elasticsearch {
hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
manage_template => false
index => "Exchange-%{+YYYY.MM.dd}"
}
}
Cunfigurazione finale
A cunfigurazione finale sarà cusì:
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}"
}
}
Ligami utili:
Source: www.habr.com