Ние сме пријатели со 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 меморија за процесот Јава. Препорачувам да наведете половина од RAM меморијата на серверот. Ако има 16 GB RAM меморија, тогаш стандардните копчиња се:

-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, тогаш пораките престануваат да се прифаќаат во редот кога ќе се достигне вредноста на која било од овие поставки. Дознајте повеќе за Постојаните редици тука.

Пример за делот од logstash.yml одговорен за поставување на редот:

queue.type: persisted
queue.max_bytes: 10gb

прилагодување

Конфигурацијата Logstash обично се состои од три дела, одговорни за различни фази на обработка на дојдовните дневници: примање (влезен дел), парсирање (дел за филтер) и испраќање до Elastic (излезен дел). Подолу ќе разгледаме подетално секој од нив.

Внесете

Го добиваме дојдовниот пренос со необработени логови од агентите за фајлбит. Токму овој приклучок го посочуваме во делот за внесување:

input {
  beats {
    port => 5044
  }
}

По оваа конфигурација, Logstash започнува да ја слуша портата 5044 и кога прима дневници, ги обработува според поставките на делот за филтри. Доколку е потребно, можете да го завиткате каналот за примање дневници од битот на датотеки во SSL. Прочитајте повеќе за поставките на додатокот Beats тука.

филтри

Сите текстуални дневници кои се интересни за обработка што ги генерира Exchange се во формат csv со полињата опишани во самата датотека за евиденција. За парсирање на CSV записи, Logstash ни нуди три приклучоци: дисецира, цсв и грок. Првиот е најмногу брза, но се справува со парсирање само на наједноставните логови.
На пример, ќе го подели следниов запис на два (поради присуството на запирка во полето), поради што дневникот ќе се анализира погрешно:

…,"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 до додатокот за разделување 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 ќе изгледа како едно градилиште, кое не дозволува анализа за броење на примателите на буквите.

Време е да додадете малку магија во процесот на обработка на дневниците.

Конвертирање на нумерички полиња

Приклучокот за дисекција има опција 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
• НССМ
• Постојани редици
• Приклучок за внесување Beats
• Логсташ пријателе, каде ми е моторната пила? Треба да ги сецирам моите трупци
• Приклучок за филтер за разделување
• Услови
• Мутирај го приклучокот за филтер
• Приклучок за филтер за датум
• Излезен приклучок Elasticsearch

Извор: www.habr.com