🥇Magkaibigan kami ni ELK at Exchange. Bahagi 2

Ipinagpapatuloy ko ang aking kwento tungkol sa kung paano makipagkaibigan kay Exchange at ELK (simula dito). Ipaalala ko sa iyo na ang kumbinasyong ito ay may kakayahang magproseso ng napakalaking bilang ng mga log nang walang pag-aalinlangan. Sa pagkakataong ito, pag-uusapan natin kung paano mapapagana ang Exchange sa mga bahagi ng Logstash at Kibana.

Ang logstash sa ELK stack ay ginagamit upang matalinong iproseso ang mga log at ihanda ang mga ito para sa paglalagay sa Elastic sa anyo ng mga dokumento, batay sa kung saan ito ay maginhawa upang bumuo ng iba't ibang mga visualization sa Kibana.

Instalasyon

Binubuo ng dalawang yugto:

Pag-install at pag-configure ng OpenJDK package.
Pag-install at pag-configure ng Logstash package.

Pag-install at pag-configure ng OpenJDK package

Ang OpenJDK package ay dapat ma-download at i-unpack sa isang partikular na direktoryo. Kung gayon ang landas sa direktoryo na ito ay dapat na maipasok sa mga variable na $env:Path at $env:JAVA_HOME ng Windows operating system:

Suriin natin ang bersyon ng 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)

Pag-install at pag-configure ng Logstash package

I-download ang archive file na may pamamahagi ng Logstash kaya. Ang archive ay dapat na i-unpack sa ugat ng disk. I-unpack sa folder C:Program Files Hindi katumbas ng halaga, tatanggi ang Logstash na magsimula nang normal. Pagkatapos ay kailangan mong pumasok sa file jvm.options mga pag-aayos na responsable para sa paglalaan ng RAM para sa proseso ng Java. Inirerekomenda kong tukuyin ang kalahati ng RAM ng server. Kung mayroon itong 16 GB ng RAM sa board, ang mga default na key ay:

-Xms1g
-Xmx1g

dapat palitan ng:

-Xms8g
-Xmx8g

Bilang karagdagan, ipinapayong magkomento sa linya -XX:+UseConcMarkSweepGC. Higit pa tungkol dito dito. Ang susunod na hakbang ay gumawa ng default na configuration sa logstash.conf file:

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

Gamit ang configuration na ito, binabasa ng Logstash ang data mula sa console, ipinapasa ito sa isang walang laman na filter, at inilalabas ito pabalik sa console. Ang paggamit ng configuration na ito ay susubukan ang functionality ng Logstash. Upang gawin ito, patakbuhin natin ito sa interactive na mode:

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}

Matagumpay na nailunsad ang Logstash sa port 9600.

Ang huling hakbang sa pag-install: ilunsad ang Logstash bilang isang serbisyo ng Windows. Magagawa ito, halimbawa, gamit ang package NSSM:

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

pagpaparaya sa kasalanan

Ang kaligtasan ng mga log kapag inilipat mula sa source server ay sinisiguro ng Persistent Queues na mekanismo.

Paano ito gumagana

Ang layout ng mga pila sa pagpoproseso ng log ay: input → queue → filter + output.

Ang input plugin ay tumatanggap ng data mula sa isang log source, isinusulat ito sa isang queue, at nagpapadala ng kumpirmasyon na ang data ay natanggap sa pinagmulan.

Ang mga mensahe mula sa queue ay pinoproseso ng Logstash, dumaan sa filter at sa output plugin. Kapag tumatanggap ng kumpirmasyon mula sa output na naipadala na ang log, inaalis ng Logstash ang naprosesong log mula sa pila. Kung hihinto ang Logstash, mananatili sa pila ang lahat ng hindi naprosesong mensahe at mensahe kung saan walang natanggap na kumpirmasyon, at patuloy na ipoproseso ng Logstash ang mga ito sa susunod na pagsisimula nito.

pag-aayos

Madaling iakma sa pamamagitan ng mga susi sa file C:Logstashconfiglogstash.yml:

queue.type: (mga posibleng halaga - persisted и memory (default)).
path.queue: (path sa folder na may mga queue file, na naka-imbak sa C:Logstashqueue bilang default).
queue.page_capacity: (maximum queue page size, default value ay 64mb).
queue.drain: (true/false - enable/disable ang paghinto sa pagpoproseso ng queue bago isara ang Logstash. Hindi ko inirerekomendang paganahin ito, dahil direktang makakaapekto ito sa bilis ng pagsara ng server).
queue.max_events: (maximum na bilang ng mga kaganapan sa queue, ang default ay 0 (walang limitasyon)).
queue.max_bytes: (maximum na laki ng queue sa bytes, default - 1024mb (1gb)).

Kung naka-configure queue.max_events и queue.max_bytes, pagkatapos ay hihinto ang pagtanggap ng mga mensahe sa pila kapag naabot na ang halaga ng alinman sa mga setting na ito. Matuto pa tungkol sa Persistent Queues dito.

Isang halimbawa ng bahagi ng logstash.yml na responsable para sa pag-set up ng queue:

queue.type: persisted
queue.max_bytes: 10gb

pag-aayos

Karaniwang binubuo ang configuration ng Logstash ng tatlong bahagi, na responsable para sa iba't ibang yugto ng pagproseso ng mga papasok na log: pagtanggap (seksyon ng input), pag-parse (seksyon ng filter) at pagpapadala sa Elastic (seksyon ng output). Sa ibaba ay susuriin natin ang bawat isa sa kanila.

input

Natanggap namin ang papasok na stream na may mga raw log mula sa mga ahente ng filebeat. Ito ang plugin na ito na ipinapahiwatig namin sa seksyon ng input:

input {
  beats {
    port => 5044
  }
}

Pagkatapos ng pagsasaayos na ito, nagsimulang makinig ang Logstash sa port 5044, at kapag tumatanggap ng mga log, pinoproseso ang mga ito ayon sa mga setting ng seksyon ng filter. Kung kinakailangan, maaari mong balutin ang channel para sa pagtanggap ng mga log mula sa filebit sa SSL. Magbasa pa tungkol sa mga setting ng beats plugin dito.

Filter

Ang lahat ng mga text log na kawili-wili para sa pagproseso na nabuo ng Exchange ay nasa csv format na may mga field na inilarawan sa log file mismo. Para sa pag-parse ng mga tala ng csv, nag-aalok sa amin ang Logstash ng tatlong plugin: masira, csv at grok. Ang una ay ang pinaka mabilis, ngunit nakayanan ang pag-parse lamang ng mga pinakasimpleng log.
Halimbawa, hahatiin nito ang sumusunod na tala sa dalawa (dahil sa pagkakaroon ng kuwit sa loob ng field), kaya naman mali ang pag-parse ng log:

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

Maaari itong magamit kapag nag-parse ng mga log, halimbawa, IIS. Sa kasong ito, maaaring ganito ang hitsura ng seksyon ng filter:

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

Nagbibigay-daan sa iyo ang pagsasaayos ng logstash na gamitin mga kondisyong pahayag, kaya maaari lang kaming magpadala ng mga log na na-tag ng filebeat tag sa dissect plugin IIS. Sa loob ng plugin itinutugma namin ang mga halaga ng field sa kanilang mga pangalan, tanggalin ang orihinal na field message, na naglalaman ng entry mula sa log, at maaari kaming magdagdag ng custom na field na, halimbawa, maglalaman ng pangalan ng application kung saan kami nangongolekta ng mga log.

Sa kaso ng mga tracking log, mas mainam na gamitin ang csv plugin; maaari itong magproseso nang tama ng mga kumplikadong field:

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

Sa loob ng plugin itinutugma namin ang mga halaga ng field sa kanilang mga pangalan, tanggalin ang orihinal na field message (at pati na rin ang mga patlang tenant-id и schema-version), na naglalaman ng entry mula sa log, at maaari kaming magdagdag ng custom na field, na, halimbawa, ay naglalaman ng pangalan ng application kung saan kami nangongolekta ng mga log.

Sa paglabas mula sa yugto ng pag-filter, makakatanggap kami ng mga dokumento sa unang pagtatantya, handa na para sa visualization sa Kibana. Mawawalan tayo ng mga sumusunod:

Makikilala ang mga numeric na field bilang text, na pumipigil sa mga operasyon sa mga ito. Ibig sabihin, ang mga patlang time-taken IIS log, pati na rin ang mga field recipient-count и total-bites Pagsubaybay sa Log.
Ang karaniwang timestamp ng dokumento ay maglalaman ng oras na naproseso ang log, hindi ang oras na isinulat ito sa gilid ng server.
Field recipient-address ay magmumukhang isang site ng konstruksiyon, na hindi nagpapahintulot para sa pagsusuri upang mabilang ang mga tatanggap ng mga titik.

Oras na para magdagdag ng kaunting magic sa proseso ng pagpoproseso ng log.

Pag-convert ng mga numeric na field

May opsyon ang dissect plugin convert_datatype, na maaaring magamit upang i-convert ang isang text field sa isang digital na format. Halimbawa, tulad nito:

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

Ito ay nagkakahalaga ng pag-alala na ang pamamaraang ito ay angkop lamang kung ang patlang ay tiyak na naglalaman ng isang string. Ang pagpipilian ay hindi nagpoproseso ng mga Null na halaga mula sa mga patlang at nagtatapon ng isang pagbubukod.

Para sa mga log ng pagsubaybay, mas mainam na huwag gumamit ng katulad na paraan ng pag-convert, dahil ang mga patlang recipient-count и total-bites maaaring walang laman. Upang i-convert ang mga patlang na ito ay mas mahusay na gumamit ng isang plugin mutate:

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

Hinahati ang recipient_address sa mga indibidwal na recipient

Ang problemang ito ay maaari ding malutas gamit ang mutate plugin:

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

Pagbabago ng timestamp

Sa kaso ng pagsubaybay sa mga log, ang problema ay napakadaling malutas ng plugin petsa, na tutulong sa iyo na magsulat sa field timestamp petsa at oras sa kinakailangang format mula sa field date-time:

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

Sa kaso ng mga log ng IIS, kakailanganin naming pagsamahin ang data ng field date и time gamit ang mutate plugin, irehistro ang time zone na kailangan namin at ilagay ang time stamp na ito timestamp gamit ang plugin ng petsa:

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

Pagbubuhos

Ang seksyon ng output ay ginagamit upang magpadala ng mga naprosesong log sa receiver ng log. Sa kaso ng direktang pagpapadala sa Elastic, isang plugin ang ginagamit nababanat, na tumutukoy sa address ng server at template ng pangalan ng index para sa pagpapadala ng nabuong dokumento:

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

Panghuling pagsasaayos

Ang huling pagsasaayos ay magiging ganito:

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

Mga kapaki-pakinabang na link:

Pinagmulan: www.habr.com

Magkaibigan kami ni ELK at Exchange. Bahagi 2

Instalasyon

Pag-install at pag-configure ng Logstash package

pagpaparaya sa kasalanan

Paano ito gumagana

pag-aayos

pag-aayos

input

Filter

Pag-convert ng mga numeric na field

Hinahati ang recipient_address sa mga indibidwal na recipient

Pagbabago ng timestamp

Pagbubuhos

Panghuling pagsasaayos

Magdagdag ng komento Kanselahin ang sumagot