Sisi ni marafiki na ELK na Exchange. Sehemu ya 2

Ninaendelea hadithi yangu kuhusu jinsi ya kutengeneza marafiki Exchange na ELK (mwanzo hapa) Napenda kukukumbusha kwamba mchanganyiko huu una uwezo wa kusindika idadi kubwa sana ya magogo bila kusita. Wakati huu tutazungumza kuhusu jinsi ya kufanya Exchange ifanye kazi na vijenzi vya Logstash na Kibana.

Logstash katika stack ELK hutumiwa kwa akili kusindika magogo na kuwatayarisha kwa kuwekwa kwa Elastic kwa namna ya nyaraka, kwa misingi ambayo ni rahisi kujenga taswira mbalimbali katika Kibana.

Ufungaji

Inajumuisha hatua mbili:

• Kufunga na kusanidi kifurushi cha OpenJDK.
• Kufunga na kusanidi kifurushi cha Logstash.

Kufunga na kusanidi kifurushi cha OpenJDK

Kifurushi cha OpenJDK lazima kipakuliwe na kufunguliwa kwenye saraka maalum. Kisha njia ya saraka hii lazima iingizwe kwenye $env:Path na $env:JAVA_HOME vigezo vya mfumo wa uendeshaji wa Windows:

Wacha tuangalie toleo la 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)

Kufunga na kusanidi kifurushi cha Logstash

Pakua faili ya kumbukumbu na usambazaji wa Logstash hivyo. Kumbukumbu lazima ifunguliwe kwenye mzizi wa diski. Fungua kwenye folda C:Program Files Sio thamani, Logstash itakataa kuanza kawaida. Kisha unahitaji kuingia kwenye faili jvm.options Marekebisho yanayowajibika kwa kutenga RAM kwa mchakato wa Java. Ninapendekeza kutaja nusu ya RAM ya seva. Ikiwa ina GB 16 ya RAM kwenye ubao, basi funguo za msingi ni:

-Xms1g
-Xmx1g

lazima ibadilishwe na:

-Xms8g
-Xmx8g

Kwa kuongeza, ni vyema kutoa maoni nje ya mstari -XX:+UseConcMarkSweepGC. Zaidi kuhusu hili hapa. Hatua inayofuata ni kuunda usanidi chaguo-msingi katika faili ya logstash.conf:

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

Kwa usanidi huu, Logstash husoma data kutoka kwa dashibodi, huipitisha kupitia kichujio tupu, na kuirudisha kwenye dashibodi. Kutumia usanidi huu kutajaribu utendakazi wa Logstash. Ili kufanya hivyo, wacha tuiendeshe katika hali ya maingiliano:

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 ilizinduliwa kwa mafanikio kwenye bandari 9600.

Hatua ya mwisho ya usakinishaji: zindua Logstash kama huduma ya Windows. Hii inaweza kufanyika, kwa mfano, kwa kutumia mfuko NSSM:

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

uvumilivu wa makosa

Usalama wa kumbukumbu zinapohamishwa kutoka kwa seva chanzo huhakikishwa na utaratibu wa Foleni Zinazoendelea.

Jinsi kazi

Mpangilio wa foleni wakati wa usindikaji wa kumbukumbu ni: ingizo → foleni → kichujio + pato.

Programu-jalizi ya ingizo hupokea data kutoka kwa chanzo cha kumbukumbu, huiandika kwenye foleni, na kutuma uthibitisho kwamba data imepokelewa kwa chanzo.

Ujumbe kutoka kwa foleni huchakatwa na Logstash, hupitishwa kupitia kichujio na programu-jalizi ya kutoa. Wakati wa kupokea uthibitisho kutoka kwa pato kwamba logi imetumwa, Logstash huondoa logi iliyochakatwa kutoka kwenye foleni. Logstash ikisimama, ujumbe na ujumbe wote ambao haujachakatwa ambao hakuna uthibitisho umepokelewa utabaki kwenye foleni, na Logstash itaendelea kuzichakata wakati mwingine itakapozianza.

marekebisho

Inaweza kubadilishwa kwa funguo kwenye faili C:Logstashconfiglogstash.yml:

• queue.type: (thamani zinazowezekana - persisted и memory (default)).
• path.queue: (njia ya folda iliyo na faili za foleni, ambazo zimehifadhiwa katika C:Logstashqueue kwa chaguo-msingi).
• queue.page_capacity: (kiwango cha juu cha ukubwa wa ukurasa wa foleni, thamani chaguo-msingi ni 64mb).
• queue.drain: (kweli/uongo - huwezesha/huzima kusimamisha usindikaji wa foleni kabla ya kuzima Logstash. Siipendekezi kuiwezesha, kwa sababu hii itaathiri moja kwa moja kasi ya kuzima kwa seva).
• queue.max_events: (idadi ya juu zaidi ya matukio kwenye foleni, chaguo-msingi ni 0 (bila kikomo)).
• queue.max_bytes: (kiwango cha juu cha ukubwa wa foleni katika ka, chaguo-msingi - 1024mb (1gb)).

Ikiwa imesanidiwa queue.max_events и queue.max_bytes, basi ujumbe huacha kukubaliwa kwenye foleni wakati thamani ya mipangilio hii inapofikiwa. Pata maelezo zaidi kuhusu Foleni Zinazoendelea hapa.

Mfano wa sehemu ya logstash.yml inayohusika na kusanidi foleni:

queue.type: persisted
queue.max_bytes: 10gb

marekebisho

Usanidi wa Logstash kawaida huwa na sehemu tatu, zinazowajibika kwa awamu tofauti za usindikaji wa kumbukumbu zinazoingia: kupokea (sehemu ya ingizo), kuchanganua (sehemu ya kichujio) na kutuma kwa Elastic (sehemu ya pato). Hapo chini tutazingatia kila mmoja wao kwa undani zaidi.

Pembejeo

Tunapokea mtiririko unaoingia na kumbukumbu ghafi kutoka kwa mawakala wa kupiga faili. Ni programu-jalizi hii ambayo tunaonyesha katika sehemu ya ingizo:

input {
  beats {
    port => 5044
  }
}

Baada ya usanidi huu, Logstash huanza kusikiliza bandari 5044, na wakati wa kupokea magogo, husindika kulingana na mipangilio ya sehemu ya chujio. Ikiwa ni lazima, unaweza kufunga kituo cha kupokea kumbukumbu kutoka kwa filebit katika SSL. Soma zaidi kuhusu mipangilio ya programu-jalizi ya beats hapa.

Chuja

Kumbukumbu zote za maandishi ambazo zinavutia kwa kuchakatwa ambazo Exchange inazalisha ziko katika umbizo la csv na sehemu zilizofafanuliwa katika faili ya kumbukumbu yenyewe. Kwa kuchanganua rekodi za csv, Logstash inatupa programu-jalizi tatu: kugawanya, csv na grok. Ya kwanza ndiyo iliyo nyingi zaidi haraka, lakini inakabiliana na kuchanganua magogo rahisi tu.
Kwa mfano, itagawanya rekodi ifuatayo kuwa mbili (kwa sababu ya uwepo wa koma ndani ya uwanja), ndiyo sababu logi itachanganuliwa vibaya:

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

Inaweza kutumika wakati wa kuchanganua magogo, kwa mfano, IIS. Katika kesi hii, sehemu ya kichungi inaweza kuonekana kama hii:

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

Usanidi wa Logstash hukuruhusu kutumia kauli zenye masharti, kwa hivyo tunaweza kutuma kumbukumbu ambazo zilitambulishwa na lebo ya mpigo wa faili kwenye programu-jalizi ya dissect IIS. Ndani ya programu-jalizi tunalinganisha maadili ya uwanja na majina yao, futa uwanja asili message, ambayo ilikuwa na ingizo kutoka kwa logi, na tunaweza kuongeza uga maalum ambao, kwa mfano, utakuwa na jina la programu ambayo tunakusanya kumbukumbu.

Kwa upande wa magogo ya kufuatilia, ni bora kutumia programu-jalizi ya csv; inaweza kusindika kwa usahihi sehemu ngumu:

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

Ndani ya programu-jalizi tunalinganisha maadili ya uwanja na majina yao, futa uwanja asili message (na pia mashamba tenant-id и schema-version), ambayo ilikuwa na kiingilio kutoka kwa logi, na tunaweza kuongeza uwanja maalum, ambao, kwa mfano, utakuwa na jina la programu ambayo tunakusanya kumbukumbu.

Wakati wa kutoka kwenye hatua ya kuchuja, tutapokea hati katika makadirio ya kwanza, tayari kwa taswira katika Kibana. Tutakosa yafuatayo:

• Sehemu za nambari zitatambuliwa kama maandishi, ambayo huzuia utendakazi juu yao. Yaani mashamba time-taken IIS logi, pamoja na mashamba recipient-count и total-bites Ufuatiliaji wa logi.
• Muhuri wa muda wa hati wa kawaida utakuwa na muda ambao kumbukumbu ilichakatwa, si wakati ilipoandikwa kwenye upande wa seva.
• Shamba recipient-address itaonekana kama tovuti moja ya ujenzi, ambayo hairuhusu uchambuzi kuhesabu wapokeaji wa barua.

Ni wakati wa kuongeza uchawi kidogo kwenye mchakato wa usindikaji wa logi.

Inabadilisha sehemu za nambari

Programu-jalizi ya dissect ina chaguo convert_datatype, ambayo inaweza kutumika kubadilisha uga wa maandishi kuwa umbizo la dijitali. Kwa mfano, kama hii:

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

Inafaa kukumbuka kuwa njia hii inafaa tu ikiwa shamba hakika litakuwa na kamba. Chaguo halichakati thamani za Null kutoka kwa sehemu na hutoa ubaguzi.

Kwa magogo ya kufuatilia, ni bora kutotumia njia sawa ya kubadilisha, tangu mashamba recipient-count и total-bites inaweza kuwa tupu. Ili kubadilisha sehemu hizi ni bora kutumia programu-jalizi mabadiliko:

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

Kugawanya_anwani_ya_mpokeaji katika wapokeaji binafsi

Shida hii pia inaweza kutatuliwa kwa kutumia programu-jalizi ya mutate:

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

Kubadilisha muhuri wa wakati

Katika kesi ya kufuatilia magogo, tatizo linatatuliwa kwa urahisi sana na programu-jalizi tarehe, ambayo itakusaidia kuandika kwenye shamba timestamp tarehe na wakati katika muundo unaohitajika kutoka kwa shamba date-time:

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

Katika kesi ya kumbukumbu za IIS, tutahitaji kuchanganya data ya shamba date и time kwa kutumia programu-jalizi ya mutate, sajili eneo la saa tunalohitaji na uweke muhuri wa wakati huu timestamp kwa kutumia programu-jalizi ya tarehe:

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

pato

Sehemu ya pato hutumiwa kutuma kumbukumbu zilizochakatwa kwa kipokea kumbukumbu. Katika kesi ya kutuma moja kwa moja kwa Elastic, programu-jalizi hutumiwa utaftaji wa elastic, ambayo inabainisha anwani ya seva na kiolezo cha jina la faharisi kwa kutuma hati iliyotolewa:

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

Usanidi wa mwisho

Mpangilio wa mwisho utaonekana kama hii:

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

Viungo muhimu:

• Jinsi ya kufunga OpenJDK 11 kwenye Windows?
• Pakua Logstash
• Elastic hutumia chaguo lililonyimwa UseConcMarkSweepGC #36828
• NSSM
• Foleni zinazoendelea
• Inashinda programu-jalizi ya kuingiza
• Logstash Jamani, msumeno wangu uko wapi? Ninahitaji kuchambua magogo yangu
• Chambua programu-jalizi ya kichujio
• Vifupisho
• Badilisha kichujio programu-jalizi
• Programu-jalizi ya kichujio cha tarehe
• Programu-jalizi ya pato la Elasticsearch

Chanzo: mapenzi.com