🥇Kita kanca karo ELK lan Exchange. Bagian 2

Aku nerusake crita babagan carane nggawe kanca Exchange lan ELK (wiwit kene). Ayo kula ngelingake yen kombinasi iki bisa ngolah jumlah log sing akeh banget tanpa ragu-ragu. Wektu iki kita bakal ngomong babagan carane Exchange bisa nggarap komponen Logstash lan Kibana.

Logstash ing tumpukan ELK digunakake kanggo ngolah log kanthi cerdas lan nyiapake kanggo penempatan ing Elastis ing wangun dokumen, kanthi basis sing trep kanggo mbangun macem-macem visualisasi ing Kibana.

Instalasi

Kasedhiya saka rong tahapan:

Nginstal lan konfigurasi paket OpenJDK.
Nginstal lan ngatur paket Logstash.

Nginstal lan konfigurasi paket OpenJDK

Paket OpenJDK kudu diundhuh lan dibongkar menyang direktori tartamtu. Banjur path menyang direktori iki kudu dilebokake ing variabel $env:Path lan $env:JAVA_HOME saka sistem operasi Windows:

Ayo mriksa versi Jawa:

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)

Nginstal lan ngatur paket Logstash

Ngundhuh file arsip kanthi distribusi Logstash saka kene. Arsip kudu dibongkar menyang oyod disk. Unpack menyang folder C:Program Files Iku ora worth iku, Logstash bakal nolak kanggo miwiti biasane. Banjur sampeyan kudu ngetik menyang file jvm.options mbenakake tanggung jawab kanggo alokasi RAM kanggo proses Jawa. Aku nyaranake nemtokake setengah saka RAM server. Yen duwe 16 GB RAM ing papan, banjur tombol standar yaiku:

-Xms1g
-Xmx1g

kudu diganti karo:

-Xms8g
-Xmx8g

Kajaba iku, disaranake menehi komentar ing baris kasebut -XX:+UseConcMarkSweepGC. Liyane babagan iki kene. Langkah sabanjure yaiku nggawe konfigurasi standar ing file logstash.conf:

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

Kanthi konfigurasi iki, Logstash maca data saka konsol, ngliwati saringan kosong, lan ngasilake maneh menyang konsol. Nggunakake konfigurasi iki bakal nyoba fungsi Logstash. Kanggo nindakake iki, ayo mbukak ing mode interaktif:

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 diluncurake kanthi sukses ing port 9600.

Langkah instalasi pungkasan: miwiti Logstash minangka layanan Windows. Iki bisa rampung, contone, nggunakake paket NSSM:

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

toleransi kesalahan

Keamanan log nalika ditransfer saka server sumber dijamin dening mekanisme Antrian Persistent.

Cara kerjane

Tata letak antrian sajrone proses log yaiku: input → queue → filter + output.

Plugin input nampa data saka sumber log, nulis menyang antrian, lan ngirim konfirmasi yen data wis ditampa menyang sumber.

Pesen saka antrian diproses dening Logstash, ngliwati filter lan plugin output. Nalika nampa konfirmasi saka output sing log wis dikirim, Logstash mbusak log diproses saka antrian. Yen Logstash mandheg, kabeh pesen lan pesen sing durung diproses sing ora ana konfirmasi sing ditampa tetep ana ing antrian, lan Logstash bakal terus ngolah nalika sabanjure diwiwiti.

imbuhan

Bisa diatur kanthi tombol ing file C:Logstashconfiglogstash.yml:

queue.type: (nilai sing bisa ditindakake - persisted и memory (default)).
path.queue: (path menyang folder karo file antrian, kang disimpen ing C: Logstashqueue minangka standar).
queue.page_capacity: (ukuran kaca antrian maksimal, nilai standar yaiku 64mb).
queue.drain: (bener / salah - mbisakake / disables mungkasi Processing antrian sadurunge mati Logstash. Aku ora nyaranake mbisakake, amarga iki bakal langsung mengaruhi kacepetan mati server).
queue.max_events: (nomer maksimum acara ing antrian, gawan punika 0 (unlimited)).
queue.max_bytes: (ukuran antrian maksimum ing bita, standar - 1024mb (1gb)).

Yen diatur queue.max_events и queue.max_bytes, banjur pesen mandheg ditampa ing antrian nalika nilai setelan kasebut wis tekan. Sinau luwih lengkap babagan Antrian Persistent kene.

Conto bagean logstash.yml sing tanggung jawab kanggo nyetel antrian:

queue.type: persisted
queue.max_bytes: 10gb

imbuhan

Konfigurasi Logstash biasane kasusun saka telung bagean, tanggung jawab kanggo macem-macem fase ngolah log mlebu: nampa (bagean input), parsing (bagean filter) lan ngirim menyang Elastis (bagean output). Ing ngisor iki kita bakal nliti saben wong.

input

Kita nampa aliran mlebu kanthi log mentah saka agen filebeat. Plugin iki sing dituduhake ing bagean input:

input {
  beats {
    port => 5044
  }
}

Sawise konfigurasi iki, Logstash wiwit ngrungokake port 5044, lan nalika nampa log, proses kasebut miturut setelan bagean filter. Yen perlu, sampeyan bisa mbungkus saluran kanggo nampa log saka filebit ing SSL. Waca liyane babagan setelan plugin beats kene.

Filter

Kabeh log teks sing menarik kanggo diproses sing digawe Exchange ana ing format csv kanthi kolom sing diterangake ing file log kasebut. Kanggo parsing cathetan csv, Logstash nawakake telung plugin: mbedakake, csv lan grok. Sing pisanan paling bya, nanging copes karo parsing mung log paling gampang.
Contone, bakal dipérang rekaman ing ngisor iki dadi loro (amarga ana koma ing njero lapangan), mula log kasebut bakal diurai kanthi ora bener:

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

Bisa digunakake nalika parsing log, contone, IIS. Ing kasus iki, bagean filter bisa katon kaya iki:

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

Konfigurasi Logstash ngidini sampeyan nggunakake statements kondisional, dadi kita mung bisa ngirim log sing diwenehi tag filebeat menyang plugin dissect IIS. Ing plugin kita cocog karo nilai lapangan kanthi jeneng, mbusak kolom asli message, sing ngemot entri saka log, lan kita bisa nambah lapangan adat sing bakal, contone, ngemot jeneng aplikasi saka kang kita ngumpulake log.

Ing kasus log nelusuri, luwih becik nggunakake plugin csv; bisa ngolah kolom kompleks kanthi bener:

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

Ing plugin kita cocog karo nilai lapangan kanthi jeneng, mbusak kolom asli message (lan uga lapangan tenant-id и schema-version), sing ngemot entri saka log, lan kita bisa nambah lapangan adat, sing bakal, contone, ngemot jeneng aplikasi saka kang kita ngumpulake log.

Ing metu saka tataran nyaring, kita bakal nampa dokumen ing kira-kira pisanan, siap kanggo visualisasi ing Kibana. Kita bakal ilang ing ngisor iki:

Kolom numerik bakal diakoni minangka teks, sing ngalangi operasi kasebut. Yaiku, sawah time-taken Log IIS, uga kolom recipient-count и total-bites Tracking Log.
Stempel wektu dokumen standar bakal ngemot wektu log diproses, dudu wektu ditulis ing sisih server.
lapangan recipient-address bakal katon kaya siji situs konstruksi, sing ora ngidini analisis kanggo ngetung panampa layang.

Iku wektu kanggo nambah sihir sethitik kanggo proses Processing log.

Ngonversi kolom angka

Plugin dissect duwe pilihan convert_datatype, sing bisa digunakake kanggo ngowahi kolom teks menyang format digital. Contone, kaya iki:

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

Sampeyan kudu eling yen cara iki mung cocok yen lapangan mesthi ngemot senar. Opsi ora ngolah nilai Null saka lapangan lan mbuwang pengecualian.

Kanggo nglacak log, luwih becik ora nggunakake metode konversi sing padha, amarga kolom kasebut recipient-count и total-bites bisa kosong. Kanggo ngowahi kolom kasebut luwih becik nggunakake plugin mutasi:

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

Pisah alamat_penerima dadi panampa individu

Masalah iki uga bisa ditanggulangi kanthi nggunakake plugin mutate:

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

Ngganti cap wektu

Ing kasus log nelusuri, masalah kasebut gampang ditanggulangi dening plugin tanggal, sing bakal mbantu sampeyan nulis ing lapangan timestamp tanggal lan wektu ing format sing dibutuhake saka lapangan date-time:

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

Ing kasus log IIS, kita kudu nggabungake data lapangan date и time nggunakake plugin mutate, ndhaftar zona wektu sing dibutuhake lan pasang stempel wektu iki timestamp nggunakake plugin tanggal:

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

output

Bagean output digunakake kanggo ngirim log sing diproses menyang panrima log. Yen ngirim langsung menyang Elastic, plugin digunakake elastisark, sing nemtokake alamat server lan cithakan jeneng indeks kanggo ngirim dokumen sing digawe:

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

Konfigurasi pungkasan

Konfigurasi pungkasan bakal katon kaya iki:

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

Link migunani:

Source: www.habr.com

Kita kanca karo ELK lan Exchange. Bagean 2

Instalasi

Nginstal lan ngatur paket Logstash

toleransi kesalahan

Cara kerjane

imbuhan

imbuhan

input

Filter

Ngonversi kolom angka

Pisah alamat_penerima dadi panampa individu

Ngganti cap wektu

output

Konfigurasi pungkasan

Add a comment Отменить ответ