Wy binne freonen mei ELK en Exchange. Diel 2

Ik gean troch mei myn ferhaal oer hoe't ik freonen meitsje kinne Exchange en ELK (begjin hjir). Lit my jo herinnerje dat dizze kombinaasje yn steat is om in heul grut oantal logs sûnder wifkjen te ferwurkjen. Dizze kear sille wy prate oer hoe't jo Exchange wurkje kinne mei Logstash- en Kibana-komponinten.

Logstash yn 'e ELK-stapel wurdt brûkt om logs yntelligint te ferwurkjen en ta te rieden foar pleatsing yn Elastic yn' e foarm fan dokuminten, op basis wêrfan it handich is om ferskate fisualisaasjes yn Kibana te bouwen.

ynstelling

Bestiet út twa stadia:

• It ynstallearjen en konfigurearjen fan it OpenJDK-pakket.
• Ynstallearje en konfigurearje fan it Logstash-pakket.

It ynstallearjen en konfigurearjen fan it OpenJDK-pakket

It OpenJDK-pakket moat wurde ynladen en útpakt yn in spesifike map. Dan moat it paad nei dizze map ynfierd wurde yn de $env:Pad en $env:JAVA_HOME fariabelen fan it Windows bestjoeringssysteem:

Litte wy de Java-ferzje kontrolearje:

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)

Ynstallearje en konfigurearje fan it Logstash-pakket

Download it argyfbestân mei de Logstash-distribúsje fan hjir. It argyf moat útpakt wurde nei de root fan 'e skiif. Utpakke nei map C:Program Files It is it net wurdich, Logstash sil wegerje om normaal te begjinnen. Dan moatte jo ynfiere yn it bestân jvm.options reparearret ferantwurdlik foar it tawizen fan RAM foar it Java-proses. Ik riede oan te jaan de helte fan de tsjinner fan RAM. As it 16 GB RAM oan board hat, dan binne de standert toetsen:

-Xms1g
-Xmx1g

moat wurde ferfongen troch:

-Xms8g
-Xmx8g

Dêrneist is it oan te rieden om kommentaar út 'e line -XX:+UseConcMarkSweepGC. Mear oer dit hjir. De folgjende stap is om in standertkonfiguraasje te meitsjen yn it logstash.conf-bestân:

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

Mei dizze konfiguraasje lêst Logstash gegevens fan 'e konsole, stjoert it troch in leech filter en jout it werom nei de konsole. It brûken fan dizze konfiguraasje sil de funksjonaliteit fan Logstash testen. Om dit te dwaan, litte wy it yn ynteraktive modus útfiere:

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 lansearre mei súkses op haven 9600.

De lêste ynstallaasjestap: starte Logstash as in Windows-tsjinst. Dit kin bygelyks mei it pakket NSSM:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

marzje foar flaters

De feiligens fan logs by oerdracht fan de boarne tsjinner wurdt garandearre troch de Persistent Queues meganisme.

Hoe't it wurket

De yndieling fan wachtrijen by logferwurking is: ynfier → wachtrige → filter + útfier.

De ynfier plugin ûntfangt gegevens fan in log boarne, skriuwt it nei in wachtrige, en stjoert befêstiging dat de gegevens binne ûntfongen nei de boarne.

Berjochten út 'e wachtrige wurde ferwurke troch Logstash, trochjûn troch it filter en de útfier plugin. By it ûntfangen fan befêstiging fan útfier dat it log is ferstjoerd, ferwideret Logstash it ferwurke log út 'e wachtrige. As Logstash ophâldt, bliuwe alle net-ferwurke berjochten en berjochten dêr't gjin befêstiging foar ûntfongen is yn 'e wachtrige, en Logstash sil trochgean mei ferwurkjen se de folgjende kear as it begjint.

oanpassing

Ferstelber troch kaaien yn it bestân C:Logstashconfiglogstash.yml:

• queue.type: (mooglike wearden - persisted и memory (default)).
• path.queue: (paad nei de map mei wachtrige triemmen, dy't standert wurde opslein yn C: Logstashqueue).
• queue.page_capacity: (maksimum wachtrige sidegrutte, standertwearde is 64mb).
• queue.drain: (wier/false - ynskeakelje / útskeakelje stopping wachtrige ferwurking foar it ôfsluten fan Logstash. Ik riede net ynskeakelje, omdat dit sil direkt ynfloed op de snelheid fan tsjinner shutdown).
• queue.max_events: (maksimum oantal eveneminten yn 'e wachtrige, standert is 0 (ûnbeheind)).
• queue.max_bytes: (maksimum wachtrige grutte yn bytes, standert - 1024mb (1gb)).

As konfigurearre queue.max_events и queue.max_bytes, dan stopje berjochten te akseptearjen yn 'e wachtrige as de wearde fan ien fan dizze ynstellings wurdt berikt. Learje mear oer persistente wachtrijen hjir.

In foarbyld fan it diel fan logstash.yml dat ferantwurdlik is foar it ynstellen fan de wachtrige:

queue.type: persisted
queue.max_bytes: 10gb

oanpassing

De Logstash-konfiguraasje bestiet normaal út trije dielen, ferantwurdlik foar ferskate fazen fan it ferwurkjen fan ynkommende logs: ûntfange (ynputseksje), parsearjen (filterseksje) en ferstjoeren nei Elastic (útfierdiel). Hjirûnder sille wy elk fan har tichterby besjen.

Ynfier

Wy ûntfange de ynkommende stream mei rau logs fan filebeat-aginten. It is dit plugin dat wy oanjaan yn 'e ynfier seksje:

input {
  beats {
    port => 5044
  }
}

Nei dizze konfiguraasje begjint Logstash te harkjen nei poarte 5044, en by it ûntfangen fan logs, ferwurket se neffens de ynstellings fan 'e filterseksje. As it nedich is, kinne jo it kanaal ynpakke foar it ûntfangen fan logs fan filebit yn SSL. Lês mear oer beats plugin ynstellings hjir.

Filter

Alle tekstlogboeken dy't ynteressant binne foar ferwurking dy't Exchange genereart binne yn csv-formaat mei de fjilden beskreaun yn it lochbestân sels. Foar it parsearjen fan csv-records biedt Logstash ús trije plugins: dissekearje, csv en grok. De earste is it meast быстрый, mar omgaat mei it parsearjen fan allinich de ienfâldichste logs.
Bygelyks, it sil de folgjende record yn twa splitse (fanwege de oanwêzigens fan in komma yn it fjild), dat is de reden dat it log ferkeard wurdt parseard:

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

It kin brûkt wurde by it parsearjen fan logs, bygelyks IIS. Yn dit gefal kin de filterseksje der sa útsjen:

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-konfiguraasje kinne jo brûke betingsten útspraken, sadat wy allinich logs kinne stjoere dy't tagged binne mei de filebeat-tag nei de dissect-plugin IIS. Binnen de plugin komme wy oerien mei de fjildwearden mei har nammen, wiskje it orizjinele fjild message, dy't befette in yngong út it log, en wy kinne tafoegje in oanpaste fjild dat sil, bygelyks, befetsje de namme fan de applikaasje dêr't wy sammelje logs.

Yn it gefal fan tracking-logs is it better om it csv-plugin te brûken; it kin komplekse fjilden korrekt ferwurkje:

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

Binnen de plugin komme wy oerien mei de fjildwearden mei har nammen, wiskje it orizjinele fjild message (en ek fjilden tenant-id и schema-version), dy't in yngong út it log befette, en wy kinne in oanpast fjild tafoegje, dat bygelyks de namme fan 'e applikaasje sil befetsje wêrfan wy logs sammelje.

By de útgong fan it filter poadium, wy sille ûntfange dokuminten yn in earste approximation, klear foar fisualisaasje yn Kibana. Wy sille it folgjende misse:

• Numerike fjilden wurde herkend as tekst, wat operaasjes op har foarkomt. Nammentlik de fjilden time-taken IIS-log, lykas fjilden recipient-count и total-bites Log Tracking.
• It standert dokumint tiidstempel sil befetsje de tiid dat it log waard ferwurke, net de tiid dat it waard skreaun op de tsjinner kant.
• fjild recipient-address sil lykje op ien bou site, dy't net tastean foar analyze te tellen de ûntfangers fan brieven.

It is tiid om in bytsje magy ta te foegjen oan it logferwurkingsproses.

Konvertearjen fan numerike fjilden

De dissect plugin hat in opsje convert_datatype, dy't brûkt wurde kin om in tekstfjild te konvertearjen nei in digitaal formaat. Bygelyks, lykas dit:

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

It is it wurdich te betinken dat dizze metoade allinich geskikt is as it fjild definityf in tekenrige sil befetsje. De opsje ferwurket gjin Null-wearden fan fjilden en smyt in útsûndering.

Foar tracking logs, is it better net te brûken in ferlykbere konvertearjen metoade, sûnt de fjilden recipient-count и total-bites kin leech wêze. Om dizze fjilden te konvertearjen is it better om in plugin te brûken mutearje:

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

Recipient_address splitsen yn yndividuele ûntfangers

Dit probleem kin ek oplost wurde mei de mutate plugin:

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

It feroarjen fan de tiidstempel

Yn it gefal fan tracking logs, it probleem is hiel maklik oplost troch de plugin datum, dy't jo helpe sil op it fjild skriuwen timestamp datum en tiid yn it fereaske formaat út it fjild date-time:

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

Yn it gefal fan IIS-logs moatte wy fjildgegevens kombinearje date и time mei de mutate plugin, registrearje de tiidsône dy't wy nedich binne en pleats dizze tiidstempel yn timestamp mei help fan de datum plugin:

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

útfier

De útfierseksje wurdt brûkt om ferwurke logs nei de logûntfanger te stjoeren. Yn gefal fan ferstjoeren direkt nei Elastic, wurdt in plugin brûkt elastyk sykje, dy't it serveradres en yndeksnammesjabloan spesifisearret foar it ferstjoeren fan it oanmakke dokumint:

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Finale konfiguraasje

De definitive konfiguraasje sil der sa útsjen:

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

Nuttige links:

• Hoe kinne jo OpenJDK 11 op Windows ynstallearje?
• Download Logstash
• Elastysk brûkt ôfskreaun opsje UseConcMarkSweepGC #36828
• NSSM
• Persistente wachtrige
• Beats ynfier plugin
• Logstash Dude, wêr is myn kettingzaag? Ik moat myn logs ûntsiferje
• Filter plugin dissectearje
• Betingsten
• Filter plugin mutearje
• Date filter plugin
• Elasticsearch útfier plugin

Boarne: www.habr.com