Men Exchange va ELK bilan qanday do'stlashish haqida hikoyamni davom ettiraman (boshi ). Shuni eslatib o'tamanki, bu kombinatsiya ikkilanmasdan juda katta miqdordagi jurnallarni qayta ishlashga qodir. Bu safar biz Exchangening Logstash va Kibana komponentlari bilan qanday ishlashi haqida gaplashamiz.
ELK stekidagi logstash jurnallarni oqilona qayta ishlash va ularni hujjatlar ko'rinishida Elastic-ga joylashtirishga tayyorlash uchun ishlatiladi, buning asosida Kibana-da turli vizualizatsiyalarni yaratish qulay.
sozlama
Ikki bosqichdan iborat:
- OpenJDK paketini o'rnatish va sozlash.
- Logstash paketini o'rnatish va sozlash.
OpenJDK paketini o'rnatish va sozlash
OpenJDK paketini yuklab olish va ma'lum bir katalogga ochish kerak. Keyin ushbu katalogga yo'l Windows operatsion tizimining $env:Path va $env:JAVA_HOME o'zgaruvchilariga kiritilishi kerak:
Keling, Java versiyasini tekshiramiz:
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 paketini o'rnatish va sozlash
Logstash tarqatish bilan arxiv faylini yuklab oling . Arxiv diskning ildiziga olib tashlanishi kerak. Jildga oching C:Program Files Bunga loyiq emas, Logstash odatdagidek boshlashdan bosh tortadi. Keyin faylga kirishingiz kerak jvm.options Java jarayoni uchun RAMni ajratish uchun mas'ul bo'lgan tuzatishlar. Serverning operativ xotirasining yarmini ko'rsatishni tavsiya qilaman. Bortida 16 GB operativ xotira bo'lsa, standart kalitlar:
-Xms1g
-Xmx1g
bilan almashtirilishi kerak:
-Xms8g
-Xmx8g
Bundan tashqari, chiziqni sharhlash tavsiya etiladi -XX:+UseConcMarkSweepGC. Bu haqida batafsil . Keyingi qadam logstash.conf faylida standart konfiguratsiyani yaratishdir:
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
Ushbu konfiguratsiya yordamida Logstash konsoldan ma'lumotlarni o'qiydi, uni bo'sh filtrdan o'tkazadi va uni yana konsolga chiqaradi. Ushbu konfiguratsiyadan foydalanish Logstash funksiyasini sinab ko'radi. Buning uchun uni interaktiv rejimda ishga tushiramiz:
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 9600 portida muvaffaqiyatli ishga tushirildi.
O'rnatishning oxirgi bosqichi: Logstash-ni Windows xizmati sifatida ishga tushiring. Bu, masalan, paket yordamida amalga oshirilishi mumkin :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
xatolarga chidamlilik
Manba serveridan uzatilganda jurnallarning xavfsizligi Doimiy navbatlar mexanizmi bilan ta'minlanadi.
Qanday ishlaydi
Jurnalni qayta ishlash jarayonida navbatlar tartibi quyidagicha: kiritish β navbat β filtr + chiqish.
Kirish plagini jurnal manbasidan ma'lumotlarni oladi, ularni navbatga yozadi va ma'lumotlar manbaga olinganligini tasdiqlaydi.
Navbatdagi xabarlar Logstash tomonidan qayta ishlanadi, filtr va chiqish plaginidan o'tadi. Chiqishdan jurnal yuborilganligi haqidagi tasdiqni olganda, Logstash qayta ishlangan jurnalni navbatdan olib tashlaydi. Agar Logstash to'xtab qolsa, qayta ishlanmagan barcha xabarlar va tasdiqlanmagan xabarlar navbatda qoladi va Logstash keyingi safar ishga tushganda ularni qayta ishlashda davom etadi.
moslashish
Fayldagi tugmalar yordamida sozlanishi C:Logstashconfiglogstash.yml:
queue.type: (mumkin qiymatlar -persistedΠΈmemory (default)).path.queue: (sukut bo'yicha C:Logstashqueue da saqlanadigan navbat fayllari bo'lgan jildga yo'l).queue.page_capacity: (maksimal navbat sahifa hajmi, standart qiymat 64mb).queue.drain: (to'g'ri/noto'g'ri - Logstash-ni o'chirishdan oldin navbatni qayta ishlashni to'xtatishni yoqadi/o'chiradi. Men uni yoqishni tavsiya etmayman, chunki bu to'g'ridan-to'g'ri serverni o'chirish tezligiga ta'sir qiladi).queue.max_events: (navbatdagi hodisalarning maksimal soni, standart 0 (cheksiz)).queue.max_bytes: (baytdagi maksimal navbat hajmi, standart - 1024mb (1gb)).
Agar sozlangan bo'lsa queue.max_events ΠΈ queue.max_bytes, keyin ushbu sozlamalardan birortasining qiymatiga erishilganda xabarlar navbatga qabul qilinishini to'xtatadi. Doimiy navbatlar haqida ko'proq bilib oling .
Navbatni o'rnatish uchun mas'ul logstash.yml qismiga misol:
queue.type: persisted
queue.max_bytes: 10gb
moslashish
Logstash konfiguratsiyasi odatda uchta qismdan iborat bo'lib, kiruvchi jurnallarni qayta ishlashning turli bosqichlari uchun javobgardir: qabul qilish (kirish bo'limi), tahlil qilish (filtr bo'limi) va Elastic-ga yuborish (chiqish qismi). Quyida biz ularning har birini batafsil ko'rib chiqamiz.
kirish
Biz kiruvchi oqimni filebeat agentlaridan xom jurnallar bilan olamiz. Kirish bo'limida biz ko'rsatadigan ushbu plagin:
input {
beats {
port => 5044
}
}
Ushbu konfiguratsiyadan so'ng, Logstash 5044 portni tinglashni boshlaydi va jurnallarni qabul qilishda ularni filtr bo'limi sozlamalariga muvofiq qayta ishlaydi. Agar kerak bo'lsa, SSL-da filebit-dan jurnallarni qabul qilish uchun kanalni o'rashingiz mumkin. Beats plagin sozlamalari haqida ko'proq o'qing .
filter
Exchange tomonidan qayta ishlash uchun qiziqarli bo'lgan barcha matn jurnallari jurnal faylida tasvirlangan maydonlar bilan csv formatida bo'ladi. Csv yozuvlarini tahlil qilish uchun Logstash bizga uchta plaginni taklif qiladi: , csv va grok. Birinchisi eng ko'p , lekin faqat eng oddiy jurnallarni tahlil qilish bilan shug'ullanadi.
Masalan, u quyidagi yozuvni ikkiga bo'ladi (maydon ichida vergul borligi sababli), shuning uchun jurnal noto'g'ri tahlil qilinadi:
β¦,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",β¦
Jurnallarni tahlil qilishda foydalanish mumkin, masalan, IIS. Bunday holda, filtr bo'limi quyidagicha ko'rinishi mumkin:
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 konfiguratsiyasi foydalanishga imkon beradi , shuning uchun biz faqat filebeat yorlig'i bilan belgilangan jurnallarni dissect plaginiga yuborishimiz mumkin IIS. Plagin ichida biz maydon qiymatlarini ularning nomlari bilan moslashtiramiz, asl maydonni o'chirib tashlaymiz message, bu jurnaldagi yozuvni o'z ichiga olgan va biz, masalan, jurnallarni to'playdigan dastur nomini o'z ichiga olgan maxsus maydonni qo'shishimiz mumkin.
Jurnallarni kuzatishda csv plaginidan foydalanish yaxshidir, u murakkab maydonlarni to'g'ri qayta ishlay oladi:
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" }
}
}
Plagin ichida biz maydon qiymatlarini ularning nomlari bilan moslashtiramiz, asl maydonni o'chirib tashlaymiz message (shuningdek, maydonlar tenant-id ΠΈ schema-version), jurnaldagi yozuvni o'z ichiga olgan va biz, masalan, jurnallarni yig'adigan dastur nomini o'z ichiga olgan maxsus maydonni qo'shishimiz mumkin.
Filtrlash bosqichidan chiqishda biz hujjatlarni Kibanada vizualizatsiya qilishga tayyor bo'lgan birinchi taxminiy ko'rinishda olamiz. Bizga quyidagilar yetishmaydi:
- Raqamli maydonlar matn sifatida tan olinadi, bu ular ustida operatsiyalarni amalga oshirishga to'sqinlik qiladi. Ya'ni, dalalar
time-takenIIS jurnali, shuningdek, maydonlarrecipient-countΠΈtotal-bitesJurnalni kuzatish. - Standart hujjat vaqt tamg'asi server tomonida yozilgan vaqtni emas, balki jurnalga ishlov berilgan vaqtni o'z ichiga oladi.
- dala
recipient-addressxatlarni qabul qiluvchilarni hisoblash uchun tahlil qilishga imkon bermaydigan bitta qurilish maydonchasiga o'xshaydi.
Jurnalni qayta ishlash jarayoniga ozgina sehr qo'shish vaqti keldi.
Raqamli maydonlarni konvertatsiya qilish
Dissect plaginida imkoniyat bor convert_datatype, bu matn maydonini raqamli formatga aylantirish uchun ishlatilishi mumkin. Masalan, bu kabi:
dissect {
β¦
convert_datatype => { "time-taken" => "int" }
β¦
}
Shuni esda tutish kerakki, bu usul faqat maydonda albatta satr bo'lsa mos keladi. Variant maydonlardagi null qiymatlarni qayta ishlamaydi va istisno qiladi.
Jurnallarni kuzatish uchun maydonlardan beri shunga o'xshash aylantirish usulini ishlatmaslik yaxshiroqdir recipient-count ΠΈ total-bites bo'sh bo'lishi mumkin. Ushbu maydonlarni aylantirish uchun plagindan foydalanish yaxshiroqdir :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Qabul qiluvchi_manzilini alohida oluvchilarga bo'lish
Ushbu muammoni mutate plagin yordamida ham hal qilish mumkin:
mutate {
split => ["recipient_address", ";"]
}
Vaqt tamg'asini o'zgartirish
Jurnallarni kuzatish bo'lsa, muammo plagin tomonidan juda oson hal qilinadi , bu sohada yozishga yordam beradi timestamp maydondan kerakli formatdagi sana va vaqt date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
IIS jurnallarida biz maydon ma'lumotlarini birlashtirishimiz kerak bo'ladi date ΠΈ time mutate plaginidan foydalanib, bizga kerak bo'lgan vaqt mintaqasini ro'yxatdan o'tkazing va bu vaqt tamg'asini joylashtiring timestamp sana plaginidan foydalanish:
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" ]
}
chiqish
Chiqish qismi qayta ishlangan jurnallarni jurnal qabul qiluvchiga yuborish uchun ishlatiladi. To'g'ridan-to'g'ri Elastic-ga yuborilganda plagin ishlatiladi , bu yaratilgan hujjatni yuborish uchun server manzili va indeks nomi shablonini belgilaydi:
output {
elasticsearch {
hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
manage_template => false
index => "Exchange-%{+YYYY.MM.dd}"
}
}
Yakuniy konfiguratsiya
Yakuniy konfiguratsiya quyidagicha ko'rinadi:
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}"
}
}
Foydali havolalar:
Manba: www.habr.com