យើងជាមិត្តនឹង ELK និង Exchange។ ផ្នែកទី 2

ខ្ញុំបន្តរឿងរបស់ខ្ញុំអំពីរបៀបបង្កើតមិត្តភក្តិ Exchange និង ELK (ចាប់ផ្តើម នៅទីនេះ) ខ្ញុំសូមរំលឹកអ្នកថា ការរួមផ្សំគ្នានេះមានសមត្ថភាពក្នុងការដំណើរការកំណត់ហេតុមួយចំនួនធំដោយមិនស្ទាក់ស្ទើរ។ លើកនេះយើងនឹងនិយាយអំពីរបៀបដើម្បីទទួលបាន Exchange ធ្វើការជាមួយ Logstash និងសមាសធាតុ Kibana ។

Logstash នៅក្នុងជង់ ELK ត្រូវបានប្រើដើម្បីដំណើរការកំណត់ហេតុដោយឆ្លាតវៃ និងរៀបចំពួកវាសម្រាប់ដាក់នៅក្នុង Elastic ក្នុងទម្រង់ជាឯកសារ ដោយឈរលើមូលដ្ឋានដែលវាងាយស្រួលក្នុងការបង្កើតការមើលឃើញផ្សេងៗនៅក្នុង Kibana ។

ការកំណត់

មានពីរដំណាក់កាល៖

• ការដំឡើង និងកំណត់រចនាសម្ព័ន្ធកញ្ចប់ OpenJDK ។
• ការដំឡើង និងកំណត់រចនាសម្ព័ន្ធកញ្ចប់ Logstash ។

ការដំឡើង និងកំណត់រចនាសម្ព័ន្ធកញ្ចប់ OpenJDK

កញ្ចប់ OpenJDK ត្រូវតែទាញយក និងពន្លាចូលទៅក្នុងថតជាក់លាក់មួយ។ បន្ទាប់មកផ្លូវទៅកាន់ថតនេះត្រូវតែបញ្ចូលទៅក្នុងអថេរ $env:Path និង $env:JAVA_HOME នៃប្រព័ន្ធប្រតិបត្តិការ Windows៖

តោះពិនិត្យមើលកំណែ 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 ជួសជុលទទួលខុសត្រូវក្នុងការបែងចែក RAM សម្រាប់ដំណើរការ Java ។ ខ្ញុំសូមផ្តល់អនុសាសន៍ឱ្យបញ្ជាក់ពាក់កណ្តាលនៃ RAM របស់ម៉ាស៊ីនមេ។ ប្រសិនបើវាមាន RAM 16 GB នៅលើយន្តហោះនោះ គ្រាប់ចុចលំនាំដើមគឺ៖

-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បន្ទាប់មក សារឈប់ត្រូវបានទទួលយកទៅក្នុងជួរ នៅពេលដែលតម្លៃនៃការកំណត់ទាំងនេះត្រូវបានឈានដល់។ មើល​បន្ថែមទៀត​អំពី Persistent Queues នៅទីនេះ.

ឧទាហរណ៍នៃផ្នែកនៃ logstash.yml ទទួលខុសត្រូវក្នុងការដំឡើងជួរ៖

queue.type: persisted
queue.max_bytes: 10gb

ការលៃតម្រូវ

ការកំណត់រចនាសម្ព័ន្ធ Logstash ជាធម្មតាមានបីផ្នែក ដែលទទួលខុសត្រូវចំពោះដំណាក់កាលផ្សេងគ្នានៃការដំណើរការកំណត់ហេតុចូល៖ ការទទួល (ផ្នែកបញ្ចូល) ការញែក (ផ្នែកតម្រង) និងការបញ្ជូនទៅកាន់ Elastic (ផ្នែកលទ្ធផល)។ ខាងក្រោម​នេះ​យើង​នឹង​ពិនិត្យ​ឱ្យ​កាន់តែ​ច្បាស់​អំពី​ពួកវា​នីមួយៗ ។

បញ្ចូល

យើងទទួលបានស្ទ្រីមចូលជាមួយនឹងកំណត់ហេតុឆៅពីភ្នាក់ងារ filebeat ។ វាគឺជាកម្មវិធីជំនួយនេះដែលយើងបង្ហាញនៅក្នុងផ្នែកបញ្ចូល៖

input {
  beats {
    port => 5044
  }
}

បន្ទាប់ពីការកំណត់រចនាសម្ព័ន្ធនេះ Logstash ចាប់ផ្តើមស្តាប់ច្រក 5044 ហើយនៅពេលទទួលកំណត់ហេតុ ដំណើរការពួកវាតាមការកំណត់នៃផ្នែកតម្រង។ បើចាំបាច់ អ្នកអាចរុំឆានែលសម្រាប់ការទទួលកំណត់ហេតុពី filebit ក្នុង SSL។ សូមអានបន្ថែមអំពីការកំណត់កម្មវិធីជំនួយ beats នៅទីនេះ.

តម្រង

កំណត់ហេតុអត្ថបទទាំងអស់ដែលគួរឱ្យចាប់អារម្មណ៍សម្រាប់ដំណើរការដែល Exchange បង្កើតគឺនៅក្នុងទម្រង់ csv ជាមួយនឹងវាលដែលបានពិពណ៌នានៅក្នុងឯកសារកំណត់ហេតុខ្លួនឯង។ សម្រាប់ការវិភាគ csv records 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) ដែលមានធាតុចូលពីកំណត់ហេតុ ហើយយើងអាចបន្ថែមវាលផ្ទាល់ខ្លួន ដែលនឹងផ្ទុកឈ្មោះកម្មវិធីដែលយើងប្រមូលកំណត់ហេតុ។

នៅ​ពេល​ចេញ​ពី​ដំណាក់​កាល​ត្រង យើង​នឹង​ទទួល​ឯកសារ​ក្នុង​ការ​ប៉ាន់ស្មាន​ដំបូង​ដែល​ត្រៀម​ខ្លួន​ជា​ស្រេច​សម្រាប់​ការ​មើល​ឃើញ​នៅ Kibana។ យើងនឹងបាត់ដូចខាងក្រោម៖

• វាលលេខនឹងត្រូវបានទទួលស្គាល់ជាអត្ថបទ ដែលរារាំងប្រតិបត្តិការលើពួកវា។ ពោលគឺវាលស្រែ time-taken កំណត់ហេតុ IIS ក៏ដូចជាវាល 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" ]
}

ការបំបែកអាសយដ្ឋាន recipient_address ទៅជាអ្នកទទួលបុគ្គល

បញ្ហានេះក៏អាចដោះស្រាយបានដោយប្រើកម្មវិធីជំនួយ mutate៖

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

ការផ្លាស់ប្តូរត្រាពេលវេលា

នៅក្នុងករណីនៃការតាមដានកំណត់ហេតុបញ្ហាត្រូវបានដោះស្រាយយ៉ាងងាយស្រួលដោយកម្មវិធីជំនួយ កាលបរិច្ឆេទដែលនឹងជួយអ្នកសរសេរក្នុងវិស័យនេះ។ timestamp កាលបរិច្ឆេទ និងពេលវេលាក្នុងទម្រង់ដែលត្រូវការពីវាល date-time:

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

ក្នុងករណី IIS logs យើងនឹងត្រូវការបញ្ចូលទិន្នន័យវាល date и time ដោយប្រើកម្មវិធីជំនួយផ្លាស់ប្តូរ សូមចុះឈ្មោះតំបន់ពេលវេលាដែលយើងត្រូវការ ហើយដាក់ត្រាពេលវេលានេះ។ 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}"
  }
}

បណ្តាញភ្ជាប់មានប្រយោជន៍:

• របៀបដំឡើង OpenJDK 11 នៅលើ Windows?
• ទាញយក Logstash
• Elastic ប្រើជម្រើសដែលមិនសមរម្យ UseConcMarkSweepGC #36828
• អិន។ អេស។ អេ
• ជួរបន្ត
• កម្មវិធីជំនួយបញ្ចូល Beats
• Logstash Dude, តើច្រវ៉ាក់របស់ខ្ញុំនៅឯណា? ខ្ញុំត្រូវការបំបែកកំណត់ហេតុរបស់ខ្ញុំ
• បំបែកកម្មវិធីជំនួយតម្រង
• ស្ថានភាព
• ផ្លាស់ប្តូរកម្មវិធីជំនួយតម្រង
• កម្មវិធីជំនួយតម្រងកាលបរិច្ឆេទ
• កម្មវិធីជំនួយលទ្ធផល Elasticsearch

ប្រភព: www.habr.com