Vi er venner med ELK og Exchange. Del 2

Jeg fortsetter historien min om hvordan jeg får venner Exchange og ELK (begynner her). La meg minne deg på at denne kombinasjonen er i stand til å behandle et veldig stort antall logger uten å nøle. Denne gangen skal vi snakke om hvordan du får Exchange til å fungere med Logstash- og Kibana-komponenter.

Logstash i ELK-stakken brukes til å intelligent behandle logger og forberede dem for plassering i Elastic i form av dokumenter, på grunnlag av hvilke det er praktisk å bygge ulike visualiseringer i Kibana.

Installasjon

Består av to stadier:

• Installere og konfigurere OpenJDK-pakken.
• Installere og konfigurere Logstash-pakken.

Installere og konfigurere OpenJDK-pakken

OpenJDK-pakken må lastes ned og pakkes ut i en bestemt katalog. Deretter må banen til denne katalogen legges inn i variablene $env:Path og $env:JAVA_HOME i Windows-operativsystemet:

La oss sjekke Java-versjonen:

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)

Installere og konfigurere Logstash-pakken

Last ned arkivfilen med Logstash-distribusjonen derav. Arkivet må pakkes ut til roten av disken. Pakk ut til mappe C:Program Files Det er ikke verdt det, Logstash vil nekte å starte normalt. Deretter må du gå inn i filen jvm.options rettelser ansvarlig for tildeling av RAM for Java-prosessen. Jeg anbefaler å spesifisere halvparten av serverens RAM. Hvis den har 16 GB RAM ombord, er standardnøklene:

-Xms1g
-Xmx1g

må erstattes med:

-Xms8g
-Xmx8g

I tillegg er det lurt å kommentere linjen -XX:+UseConcMarkSweepGC. Mer om dette her. Det neste trinnet er å lage en standardkonfigurasjon i logstash.conf-filen:

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

Med denne konfigurasjonen leser Logstash data fra konsollen, sender dem gjennom et tomt filter og sender dem tilbake til konsollen. Bruk av denne konfigurasjonen vil teste funksjonaliteten til Logstash. For å gjøre dette, la oss kjøre den i interaktiv modus:

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 ble lansert på port 9600.

Det siste installasjonstrinnet: start Logstash som en Windows-tjeneste. Dette kan for eksempel gjøres ved å bruke pakken NSSM:

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

feiltoleranse

Sikkerheten til logger når de overføres fra kildeserveren er sikret av Persistent Queue-mekanismen.

Slik fungerer det

Oppsettet av køer under loggbehandling er: input → kø → filter + output.

Inngangspluginet mottar data fra en loggkilde, skriver det til en kø og sender bekreftelse på at dataene er mottatt til kilden.

Meldinger fra køen behandles av Logstash, sendes gjennom filteret og utdataplugin. Når du mottar bekreftelse fra utdata om at loggen er sendt, fjerner Logstash den behandlede loggen fra køen. Hvis Logstash stopper, forblir alle ubehandlede meldinger og meldinger som det ikke er mottatt bekreftelse på i køen, og Logstash vil fortsette å behandle dem neste gang den starter.

justering

Justerbar med taster i filen C:Logstashconfiglogstash.yml:

• queue.type: (mulige verdier - persisted и memory (default)).
• path.queue: (sti til mappen med køfiler, som er lagret i C:Logstashqueue som standard).
• queue.page_capacity: (maksimal køsidestørrelse, standardverdi er 64mb).
• queue.drain: (true/false - aktiverer/deaktiverer stopp av købehandling før du slår av Logstash. Jeg anbefaler ikke å aktivere det, fordi dette vil direkte påvirke hastigheten på serveravslutning).
• queue.max_events: (maksimalt antall hendelser i køen, standard er 0 (ubegrenset)).
• queue.max_bytes: (maksimal køstørrelse i byte, standard - 1024mb (1gb)).

Hvis konfigurert queue.max_events и queue.max_bytes, da slutter meldinger å bli akseptert i køen når verdien for noen av disse innstillingene er nådd. Lær mer om vedvarende køer her.

Et eksempel på delen av logstash.yml som er ansvarlig for å sette opp køen:

queue.type: persisted
queue.max_bytes: 10gb

justering

Logstash-konfigurasjonen består vanligvis av tre deler, ansvarlige for ulike faser av behandlingen av innkommende logger: mottak (inndataseksjon), parsing (filterseksjon) og sending til Elastic (utgangsseksjon). Nedenfor skal vi se nærmere på hver av dem.

Input

Vi mottar den innkommende strømmen med rålogger fra filebeat-agenter. Det er denne plugin-en vi angir i inngangsdelen:

input {
  beats {
    port => 5044
  }
}

Etter denne konfigurasjonen begynner Logstash å lytte til port 5044, og når du mottar logger, behandler de dem i henhold til innstillingene til filterdelen. Om nødvendig kan du pakke inn kanalen for mottak av logger fra filebit i SSL. Les mer om beats plugin-innstillinger her.

filtre

Alle tekstlogger som er interessante for behandling som Exchange genererer er i csv-format med feltene beskrevet i selve loggfilen. For å analysere csv-poster tilbyr Logstash oss tre plugins: dissekere, csv og grok. Den første er mest rask, men takler kun å analysere de enkleste loggene.
For eksempel vil den dele følgende post i to (på grunn av tilstedeværelsen av et komma inne i feltet), som er grunnen til at loggen vil bli analysert feil:

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

Den kan brukes når du analyserer logger, for eksempel IIS. I dette tilfellet kan filterdelen se slik ut:

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-konfigurasjon lar deg bruke betingede uttalelser, så vi kan bare sende logger som er merket med filebeat-taggen til dissect-pluginen IIS. Inne i pluginet matcher vi feltverdiene med navnene deres, slett det opprinnelige feltet message, som inneholdt en oppføring fra loggen, og vi kan legge til et tilpasset felt som for eksempel vil inneholde navnet på applikasjonen vi samler logger fra.

Når det gjelder sporingslogger, er det bedre å bruke csv-pluginen; den kan behandle komplekse felt på riktig måte:

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

Inne i pluginet matcher vi feltverdiene med navnene deres, slett det opprinnelige feltet message (og også felt tenant-id и schema-version), som inneholdt en oppføring fra loggen, og vi kan legge til et tilpasset felt, som for eksempel vil inneholde navnet på applikasjonen vi samler logger fra.

Ved utgangen fra filtreringsstadiet vil vi motta dokumenter i en første tilnærming, klare for visualisering i Kibana. Vi kommer til å mangle følgende:

• Numeriske felt vil bli gjenkjent som tekst, noe som forhindrer operasjoner på dem. Nemlig feltene time-taken IIS-logg, samt felt recipient-count и total-bites Loggsporing.
• Standard dokumenttidsstempel vil inneholde tidspunktet loggen ble behandlet, ikke tidspunktet den ble skrevet på serversiden.
• Feltet recipient-address vil se ut som én byggeplass, som ikke tillater analyse for å telle mottakerne av brev.

Det er på tide å legge til litt magi til loggbehandlingsprosessen.

Konvertering av numeriske felt

Dissekeringspluginen har et alternativ convert_datatype, som kan brukes til å konvertere et tekstfelt til et digitalt format. For eksempel slik:

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

Det er verdt å huske at denne metoden bare er egnet hvis feltet definitivt vil inneholde en streng. Alternativet behandler ikke nullverdier fra felt og gir et unntak.

For sporingslogger er det bedre å ikke bruke en lignende konverteringsmetode, siden feltene recipient-count и total-bites kan være tom. For å konvertere disse feltene er det bedre å bruke en plugin mutere:

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

Deler opp mottakeradresse i individuelle mottakere

Dette problemet kan også løses ved å bruke mutate plugin:

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

Endre tidsstemplet

Når det gjelder sporingslogger, løses problemet veldig enkelt med plugin data, som vil hjelpe deg å skrive i feltet timestamp dato og klokkeslett i ønsket format fra feltet date-time:

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

Når det gjelder IIS-logger, må vi kombinere feltdata date и time ved å bruke mutate-pluginen, registrer tidssonen vi trenger og sett inn dette tidsstemplet timestamp ved å bruke dato-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" ]
}

Produksjon

Utgangsdelen brukes til å sende behandlede logger til loggmottakeren. Ved sending direkte til Elastic brukes en plugin elasticsearch, som spesifiserer serveradressen og indeksnavnemalen for sending av det genererte dokumentet:

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

Endelig konfigurasjon

Den endelige konfigurasjonen vil se slik ut:

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

Nyttige lenker:

• Hvordan installere OpenJDK 11 på Windows?
• Last ned Logstash
• Elastic bruker avskrevet alternativ UseConcMarkSweepGC #36828
• NSSM
• Vedvarende køer
• Beats input plugin
• Logstash Dude, hvor er motorsagen min? Jeg må dissekere loggene mine
• Dissekere filter plugin
• conditionals
• Mutate filter plugin
• Datofilter-plugin
• Utgangsplugin for Elasticsearch

Kilde: www.habr.com