Биз ELK жана Exchange менен доспуз. 2 бөлүк

Мен 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 бөлүштүрүү үчүн жооптуу оңдоолор. Мен сервердин оперативдик эс тутумунун жарымын көрсөтүүнү сунуштайм. Анын бортунда 16 ГБ оперативдүү эс тутум болсо, демейки баскычтар:

-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!

катага сабырдуулук

Булак серверинен өткөрүлүп берилген журналдардын коопсуздугу Persistent Queues механизми менен камсыз кылынат.

Бул кандайча иштейт

Журналды иштетүүдө кезектердин схемасы: киргизүү → кезек → чыпка + чыгаруу.

Киргизүү плагини журнал булагынан маалыматтарды кабыл алып, аны кезекке жазат жана маалымат булакка кабыл алынгандыгын тастыктоо менен жөнөтөт.

Кезектен келген билдирүүлөр Logstash тарабынан иштелип чыгып, чыпкадан жана чыгаруу плагининен өтөт. Чыгуудан журнал жөнөтүлгөнү тууралуу ырастоону алганда, Logstash иштетилген журналды кезектен алып салат. Logstash токтоп калса, бардык иштетилбеген билдирүүлөр жана эч кандай ырастоо алынбаган билдирүүлөр кезекте калат жана Logstash кийинки жолу башталганда аларды иштетүүнү улантат.

тууралоо

Файлдагы баскычтар менен жөнгө салынат C:Logstashconfiglogstash.yml:

• queue.type: (мүмкүн болгон баалуулуктар - persisted и memory (default)).
• path.queue: (демейки боюнча C:Logstashqueueде сакталган кезек файлдары бар папкага жол).
• queue.page_capacity: (максималдуу кезек барак өлчөмү, демейки маани 64mb).
• 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де 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 конфигурациясы колдонууга мүмкүндүк берет шарттуу билдирүүлөр, ошондуктан биз dissect плагинине filebeat теги менен белгиленген журналдарды гана жөнөтө алабыз 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 Log Tracking.
• Стандарттык документтин убакыт белгиси сервер тарабында жазылган убакытты эмес, журналдын иштетилген убактысын камтыйт.
• талаа 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 плагини аркылуу да чечсе болот:

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ке жөнөтүлгөн учурда, плагин колдонулат ElasticSearch, ал түзүлгөн документти жөнөтүү үчүн сервердин дарегин жана индекстин аталышын көрсөтөт:

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

Пайдалуу шилтемелер:

• Windowsта OpenJDK 11ди кантип орнотуу керек?
• Logstash жүктөп алыңыз
• Эластик UseConcMarkSweepGC #36828 деприкацияланган опцияны колдонот
• NSSM
• Туруктуу кезектер
• Beats киргизүү плагини
• Logstash досум, менин араам кайда? Мен журналдарымды кесишим керек
• Чыпкалоо плагини
• Conditionals
• Мутация фильтринин плагини
• Дата чыпкасы плагини
• Elasticsearch чыгаруу плагини

Source: www.habr.com