ჩვენ ვმეგობრობთ ELK-თან და Exchange-თან. Მე -2 ნაწილი

მე ვაგრძელებ ჩემს ისტორიას იმის შესახებ, თუ როგორ შევქმნათ მეგობრები Exchange და ELK (დასაწყისი აქ). შეგახსენებთ, რომ ამ კომბინაციას შეუძლია უყოყმანოდ დაამუშაოს მორების ძალიან დიდი რაოდენობა. ამჯერად ვისაუბრებთ იმაზე, თუ როგორ უნდა ვიმუშაოთ Exchange Logstash და Kibana კომპონენტებთან.

Logstash ELK სტეკში გამოიყენება ჟურნალების ინტელექტუალურად დასამუშავებლად და დოკუმენტების სახით Elastic-ში განთავსებისთვის მოსამზადებლად, რის საფუძველზეც მოსახერხებელია Kibana-ში სხვადასხვა ვიზუალიზაციის აგება.

დაყენების

შედგება ორი ეტაპისგან:

• 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 აფიქსირებს პასუხისმგებელ RAM-ის გამოყოფას Java პროცესისთვის. მე გირჩევთ მიუთითოთ სერვერის 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 სერვისი. ეს შეიძლება გაკეთდეს, მაგალითად, პაკეტის გამოყენებით NSSM:

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: (რიდის გვერდის მაქსიმალური ზომა, ნაგულისხმევი მნიშვნელობა არის 64 მბ).
• queue.drain: (true/false - ჩართავს/გამორთავს რიგის დამუშავების შეჩერებას Logstash-ის გამორთვამდე. არ გირჩევთ მის ჩართვას, რადგან ეს პირდაპირ იმოქმედებს სერვერის გათიშვის სიჩქარეზე).
• queue.max_events: (მოვლენების მაქსიმალური რაოდენობა რიგში, ნაგულისხმევი არის 0 (შეუზღუდავი)).
• queue.max_bytes: (რიდის მაქსიმალური ზომა ბაიტებში, ნაგულისხმევი - 1024 მბ (1 გბ)).

თუ კონფიგურირებულია 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-ში. წაიკითხეთ მეტი 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 plugin-ში 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-taken IIS ჟურნალი, ისევე როგორც ველები recipient-count и total-bites ჟურნალის თვალყურის დევნება.
• სტანდარტული დოკუმენტის დროის შტამპი შეიცავს ჟურნალის დამუშავების დროს და არა სერვერის მხარეს დაწერის დროს.
• საველე recipient-address დაემსგავსება ერთ სამშენებლო ობიექტს, რომელიც არ იძლევა ანალიზს ასოების მიმღებთა დათვლის საშუალებას.

დროა დაამატოთ ცოტა მაგია ჟურნალის დამუშავების პროცესს.

რიცხვითი ველების კონვერტაცია

dissect plugin-ს აქვს ვარიანტი 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 {
  split => ["recipient_address", ";"]
}

დროის ანაბეჭდის შეცვლა

ლოგების თვალთვალის შემთხვევაში პრობლემა ძალიან მარტივად გვარდება მოდულის მიერ თარიღი, რომელიც დაგეხმარება ველში წერაში timestamp თარიღი და დრო ველიდან საჭირო ფორმატში date-time:

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

IIS ჟურნალების შემთხვევაში, ჩვენ დაგვჭირდება ველის მონაცემების გაერთიანება 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
• ელასტიური იყენებს გაუქმებულ ვარიანტს UseConcMarkSweepGC #36828
• NSSM
• მუდმივი რიგები
• Beats შეყვანის დანამატი
• ლოგსტაშ ძმაო, სად არის ჩემი ჯაჭვის ხერხი? მე უნდა დავშალო ჩემი მორები
• ფილტრის დაშლის დანამატი
• პირობითები
• ფილტრის მოდულის მუტაცია
• თარიღის ფილტრის დანამატი
• Elasticsearch გამომავალი მოდული

წყარო: www.habr.com