Біз ELK және Exchange компанияларымен доспыз. 2-бөлім

Мен Exchange және ELK достарын қалай жасауға болатындығы туралы әңгімемді жалғастырамын (басы осында). Естеріңізге сала кетейін, бұл комбинация өте көп журналдарды еш ойланбастан өңдеуге қабілетті. Бұл жолы Logstash және Kibana құрамдастары бар Exchange орнату жолы туралы сөйлесеміз.

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 процесі үшін жедел жадты бөлуге жауапты өңдеулер. Мен сервердің жедел жадының жартысын көрсетуді ұсынамын. Егер оның бортында 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 портында сәтті іске қосылды.

Соңғы орнату қадамы: Windows қызметі ретінде Logstash іске қосу. Мұны, мысалы, буманы пайдалану арқылы жасауға болады 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-де 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 Журналды бақылау.
• Құжаттың стандартты уақыт белгісі сервер жағында жазылған уақытты емес, журналдың өңделген уақытын қамтиды.
• өріс recipient-address хаттарды қабылдаушыларды санау арқылы талдауға мүмкіндік бермейтін бір құрылыс алаңына ұқсайды.

Тіркеу процесіне сиқыр қосудың уақыты келді.

Сандық өрісті түрлендіру

Dissect плагинінде опция бар convert_datatype, ол мәтін өрісін сандық пішімге түрлендіру үшін пайдаланылуы мүмкін. Мысалы, келесідей:

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

Бұл әдіс өрісте міндетті түрде жол болған жағдайда ғана жарамды екенін есте ұстаған жөн. Өрістердегі нөлдік мәндер опциямен өңделмейді және ерекше жағдайға тасталады.

Бақылау журналдары үшін ұқсас түрлендіру әдісін пайдаланбаған дұрыс, өйткені өрістер recipient-count и total-bites бос болуы мүмкін. Бұл өрістерді түрлендіру үшін плагинді қолданған дұрыс мутация:

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

Алушы_мекенжайын жеке алушыларға бөлу

Бұл тапсырманы мутация плагинінің көмегімен де шешуге болады:

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

Пайдалы сілтемелер:

• Windows жүйесінде OpenJDK 11 қалай орнатуға болады?
• Logstash жүктеп алыңыз
• Elastic қолданылған UseConcMarkSweepGC #36828 опциясын пайдаланады
• NSSM
• Тұрақты кезектер
• Beats енгізу плагині
• Logstash досым, менің шынжырлы арам қайда? Маған журналдарымды бөлу керек
• Бөлшек сүзгі плагині
• Шарттар
• Сүзгі плагинін өзгерту
• Күн сүзгісінің плагині
• Elasticsearch шығыс плагині

Ақпарат көзі: www.habr.com