Би Exchange болон ELK-тэй хэрхэн найзлах тухай түүхийг үргэлжлүүлж байна (эхлэл ). Энэхүү хослол нь маш олон тооны логийг ямар ч эргэлзээгүйгээр боловсруулах чадвартай гэдгийг танд сануулъя. Энэ удаад бид Exchange-г Logstash болон Kibana бүрэлдэхүүн хэсгүүдтэй хэрхэн ажиллуулах талаар ярилцах болно.
ELK стек дэх Logstash нь бүртгэлийг ухаалгаар боловсруулж, баримт бичгийн хэлбэрээр Elastic-д байрлуулахад бэлтгэхэд ашиглагддаг бөгөөд үүний үндсэн дээр Кибана дахь янз бүрийн дүрслэлийг бий болгоход тохиромжтой.
тохиргоо
Хоёр үе шатаас бүрдэнэ:
- OpenJDK багцыг суулгаж, тохируулж байна.
- Logstash багцыг суулгаж, тохируулж байна.
OpenJDK багцыг суулгаж, тохируулж байна
OpenJDK багцыг татан авч, тодорхой лавлах руу задлах ёстой. Дараа нь энэ директор руу очих замыг Windows үйлдлийн системийн $env:Path болон $env:JAVA_HOME хувьсагчдад оруулах ёстой.
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)
Logstash багцыг суулгаж, тохируулж байна
Logstash түгээлтийн тусламжтайгаар архивын файлыг татаж аваарай . Архивыг дискний үндэс рүү задлах ёстой. Хавтас руу задлах C:Program Files Энэ нь үнэ цэнэтэй зүйл биш, Logstash хэвийн эхлэхээс татгалзах болно. Дараа нь та файл руу орох хэрэгтэй jvm.options Java процесст RAM хуваарилах үүрэгтэй засварууд. Би серверийн RAM-ийн хагасыг зааж өгөхийг зөвлөж байна. Хэрэв 16 ГБ RAM-тай бол үндсэн товчлуурууд нь:
-Xms1g
-Xmx1g
солих шаардлагатай:
-Xms8g
-Xmx8g
Үүнээс гадна, энэ нь шугам гарч тайлбар өгөх нь зүйтэй юм -XX:+UseConcMarkSweepGC. Энэ талаар дэлгэрэнгүй . Дараагийн алхам бол logstash.conf файлд анхдагч тохиргоог үүсгэх явдал юм.
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
Энэ тохиргоог хийснээр Logstash нь консолоос өгөгдлийг уншиж, хоосон шүүлтүүрээр дамжуулж, консол руу буцааж гаргадаг. Энэ тохиргоог ашигласнаар Logstash-ийн ажиллагааг шалгах болно. Үүнийг хийхийн тулд интерактив горимд ажиллуулъя:
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 порт дээр амжилттай эхлүүлсэн.
Суулгах эцсийн алхам: Logstash-ийг Windows үйлчилгээ болгон ажиллуул. Үүнийг жишээ нь багцыг ашиглан хийж болно :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
алдааг тэсвэрлэх чадвар
Эх серверээс шилжүүлсэн бүртгэлийн аюулгүй байдлыг Байнгын дараалал механизмаар хангадаг.
Энэ нь хэрхэн ажилладаг вэ?
Лог боловсруулах явцад дарааллын зохион байгуулалт нь: оролт → дараалал → шүүлтүүр + гаралт.
Оролтын залгаас нь бүртгэлийн эх сурвалжаас өгөгдлийг хүлээн авч, дараалалд бичиж, өгөгдлийг эх сурвалж руу хүлээн авсны баталгааг илгээдэг.
Дараалалаас ирсэн мессежийг Logstash боловсруулж, шүүлтүүр болон гаралтын залгаасаар дамжуулдаг. Гаралтаас бүртгэл илгээсэн гэсэн баталгааг хүлээн авах үед Logstash боловсруулсан бүртгэлийг дарааллаас устгадаг. Хэрэв Logstash зогсвол боловсруулаагүй бүх мессежүүд болон баталгаажуулалт аваагүй мессежүүд дараалалд үлдэх бөгөөд Logstash дараагийн удаад тэдгээрийг үргэлжлүүлэн боловсруулах болно.
тохируулга
Файлын товчлууруудаар тохируулж болно C:Logstashconfiglogstash.yml:
queue.type: (боломжтой утгууд -persistedиmemory (default)).path.queue: (анхдагчаар C:Logstashqueue-д хадгалагдсан дарааллын файл бүхий хавтас руу очих зам).queue.page_capacity: (хамгийн их дарааллын хуудасны хэмжээ, анхдагч утга нь 64mb).queue.drain: (үнэн/худал - Logstash-ийг хаахаас өмнө дарааллын боловсруулалтыг зогсоохыг идэвхжүүлдэг/идэвхгүй болгодог. Би үүнийг идэвхжүүлэхийг зөвлөдөггүй, учир нь энэ нь сервер унтрах хурдад шууд нөлөөлнө).queue.max_events: (дараалал дахь үйл явдлын хамгийн их тоо, анхдагч нь 0 (хязгааргүй)).queue.max_bytes: (байт дахь дарааллын дээд хэмжээ, анхдагч - 1024mb (1gb)).
Хэрэв тохируулсан бол queue.max_events и queue.max_bytes, дараа нь эдгээр тохиргооны аль нэгийн утгад хүрсэн үед мессежийг дараалалд хүлээн авахаа болино. Байнгын дарааллын талаар нэмэлт мэдээлэл аваарай .
Дараалал тогтоох үүрэгтэй logstash.yml хэсгийн жишээ:
queue.type: persisted
queue.max_bytes: 10gb
тохируулга
Logstash тохиргоо нь ихэвчлэн гурван хэсгээс бүрдэх бөгөөд ирж буй бүртгэлийг боловсруулах янз бүрийн үе шатуудыг хариуцдаг: хүлээн авах (оролтын хэсэг), задлан шинжлэх (шүүлтүүр хэсэг) болон Elastic руу илгээх (гаралтын хэсэг). Доор бид тус бүрийг нарийвчлан авч үзэх болно.
Оролт
Бид filebeat агентуудаас ирж буй урсгалыг түүхий бүртгэлээр хүлээн авдаг. Энэ бол оролтын хэсэгт заасан залгаас юм:
input {
beats {
port => 5044
}
}
Энэ тохиргооны дараа Logstash нь 5044 портыг сонсож эхэлдэг бөгөөд бүртгэлийг хүлээн авахдаа шүүлтүүрийн хэсгийн тохиргооны дагуу боловсруулдаг. Шаардлагатай бол та SSL-д filebit-аас бүртгэл хүлээн авах сувгийг боож болно. Beats залгаасын тохиргооны талаар дэлгэрэнгүй уншина уу .
Шүүлтүүр
Exchange-ийн үүсгэсэн боловсруулахад сонирхолтой бүх текст лог нь лог файлд тодорхойлсон талбаруудтай csv форматтай байна. Csv бичлэгийг задлан шинжлэхийн тулд Logstash бидэнд гурван нэмэлт өргөтгөл санал болгодог: , csv болон grok. Эхнийх нь хамгийн их , гэхдээ зөвхөн хамгийн энгийн логуудыг задлан шинжилж чаддаг.
Жишээлбэл, энэ нь дараах бичлэгийг хоёр болгон хуваах болно (талбар дотор таслал байгаа тул) логийг буруу задлан шинжлэх болно:
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…
Үүнийг логийг задлан шинжлэхэд ашиглаж болно, жишээлбэл, IIS. Энэ тохиолдолд шүүлтүүрийн хэсэг дараах байдалтай байж болно.
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 тохиргоо нь танд ашиглах боломжийг олгодог , тиймээс бид зөвхөн filebeat шошготой логуудыг dissect залгаас руу илгээх боломжтой IIS. Залгаасны дотор талбарын утгыг нэрстэй нь тааруулж, анхны талбарыг устгана message, логоос оруулсан оруулгыг агуулсан бөгөөд бид жишээ нь бидний лог цуглуулдаг програмын нэрийг агуулсан тусгай талбар нэмж болно.
Бүртгэлийг хянах тохиолдолд csv залгаасыг ашиглах нь дээр бөгөөд энэ нь нарийн төвөгтэй талбаруудыг зөв боловсруулж чаддаг.
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" }
}
}
Залгаасны дотор талбарын утгыг нэрстэй нь тааруулж, анхны талбарыг устгана message (мөн талбайнууд tenant-id и schema-version), бүртгэлээс оруулсан оруулгыг агуулсан бөгөөд бид жишээлбэл, лог цуглуулдаг програмын нэрийг агуулсан тусгай талбар нэмж болно.
Шүүлтийн үе шатнаас гарахад бид баримт бичгийг Кибана хэл дээр харуулахад бэлэн эхний ойролцоо байдлаар хүлээн авах болно. Бидэнд дараах зүйлс дутагдах болно.
- Тоон талбаруудыг текст гэж хүлээн зөвшөөрөх бөгөөд энэ нь тэдгээр дээр ажиллахаас сэргийлнэ. Тухайлбал, талбайнууд
time-takenIIS бүртгэл, түүнчлэн талбаруудrecipient-countиtotal-bitesХяналтын бүртгэл. - Стандарт баримт бичгийн цагийн тэмдэг нь серверийн тал дээр бичигдсэн хугацааг биш харин бүртгэлийг боловсруулсан цагийг агуулна.
- талбар
recipient-addressзахидал хүлээн авагчдыг тоолох дүн шинжилгээ хийх боломжийг олгодоггүй нэг барилгын талбай шиг харагдах болно.
Бүртгэл боловсруулах үйл явцад бага зэрэг ид шид нэмэх цаг болжээ.
Тоон талбарыг хөрвүүлэх
Dissect залгаас нь сонголттой convert_datatype, текст талбарыг дижитал формат руу хөрвүүлэхэд ашиглаж болно. Жишээлбэл, иймэрхүү:
dissect {
…
convert_datatype => { "time-taken" => "int" }
…
}
Энэ арга нь тухайн талбарт заавал мөр агуулсан тохиолдолд л тохиромжтой гэдгийг санах нь зүйтэй. Сонголт нь талбаруудын Null утгыг боловсруулахгүй бөгөөд үл хамаарах зүйл үүсгэдэг.
Бүртгэлийг хянахын тулд талбаруудаас хойш ижил төстэй хөрвүүлэх аргыг ашиглахгүй байх нь дээр recipient-count и total-bites хоосон байж болно. Эдгээр талбаруудыг хөрвүүлэхийн тулд залгаасыг ашиглах нь дээр :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Хүлээн авагчийн хаягийг хүлээн авагч болгон хуваах
Энэ асуудлыг мөн mutate plugin ашиглан шийдэж болно:
mutate {
split => ["recipient_address", ";"]
}
Цагийн тэмдгийг өөрчлөх
Бүртгэлийг хянах тохиолдолд залгаасын тусламжтайгаар асуудлыг маш амархан шийддэг , энэ нь танд талбарт бичихэд тусална timestamp талбараас шаардлагатай форматаар огноо, цаг date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
IIS бүртгэлийн хувьд бид талбарын өгөгдлийг нэгтгэх шаардлагатай болно date и time mutate залгаасыг ашиглан бидэнд хэрэгтэй цагийн бүсийг бүртгэж, энэ цагийн тэмдгийг оруулна уу timestamp огнооны залгаасыг ашиглан:
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" ]
}
гаралтын
Гаралтын хэсэг нь боловсруулсан бүртгэлийг бүртгэлийн хүлээн авагч руу илгээхэд ашиглагддаг. Elastic руу шууд илгээх тохиолдолд залгаасыг ашигладаг , үүсгэсэн баримтыг илгээх серверийн хаяг болон индексийн нэрийн загварыг зааж өгдөг:
output {
elasticsearch {
hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
manage_template => false
index => "Exchange-%{+YYYY.MM.dd}"
}
}
Эцсийн тохиргоо
Эцсийн тохиргоо дараах байдлаар харагдах болно.
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}"
}
}
Хэрэгтэй холбоосууд:
Эх сурвалж: www.habr.com