Við erum vinir ELK og Exchange. 2. hluti

Ég held áfram sögu minni um hvernig á að eignast vini Exchange og ELK (byrjun hér). Leyfðu mér að minna þig á að þessi samsetning er fær um að vinna mjög mikinn fjölda annála án þess að hika. Að þessu sinni munum við tala um hvernig á að fá Exchange til að vinna með Logstash og Kibana íhlutum.

Logstash í ELK staflanum er notað til að vinna úr annálum á skynsamlegan hátt og undirbúa þá fyrir staðsetningu í Elastic í formi skjala, á grundvelli þeirra er þægilegt að byggja upp ýmsar sjónmyndir í Kibana.

Uppsetning

Samanstendur af tveimur þrepum:

• Að setja upp og stilla OpenJDK pakkann.
• Að setja upp og stilla Logstash pakkann.

Að setja upp og stilla OpenJDK pakkann

OpenJDK pakkann verður að hlaða niður og pakka niður í tiltekna möppu. Síðan verður að slá inn slóðina að þessari möppu í $env:Path og $env:JAVA_HOME breyturnar í Windows stýrikerfinu:

Við skulum athuga Java útgáfuna:

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)

Að setja upp og stilla Logstash pakkann

Sæktu skjalasafnið með Logstash dreifingunni þess vegna. Safninu verður að pakka niður að rót disksins. Taktu niður í möppu C:Program Files Það er ekki þess virði, Logstash mun neita að byrja venjulega. Þá þarftu að slá inn í skrána jvm.options lagfæringar sem bera ábyrgð á úthlutun vinnsluminni fyrir Java ferlið. Ég mæli með að tilgreina helming af vinnsluminni miðlarans. Ef það er með 16 GB af vinnsluminni um borð, þá eru sjálfgefnir lyklar:

-Xms1g
-Xmx1g

verður að skipta út fyrir:

-Xms8g
-Xmx8g

Auk þess er ráðlegt að gera athugasemdir við línuna -XX:+UseConcMarkSweepGC. Meira um þetta hér. Næsta skref er að búa til sjálfgefna stillingu í logstash.conf skránni:

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

Með þessari uppsetningu les Logstash gögn úr stjórnborðinu, setur þau í gegnum tóma síu og sendir þau aftur til stjórnborðsins. Notkun þessarar stillingar mun prófa virkni Logstash. Til að gera þetta skulum við keyra það í gagnvirkum ham:

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 var hleypt af stokkunum með góðum árangri á höfn 9600.

Síðasta uppsetningarskrefið: ræstu Logstash sem Windows þjónustu. Þetta er til dæmis hægt að gera með því að nota pakkann NSSM:

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

bilanaþol

Öryggi annála þegar þær eru fluttar frá upprunaþjóninum er tryggt með viðvarandi biðröðum.

Hvernig það virkar

Skipulag biðraða við vinnslu skrár er: inntak → biðröð → sía + úttak.

Inntaksviðbótin tekur við gögnum frá annálgjafa, skrifar þau í biðröð og sendir staðfestingu á að gögnin hafi verið móttekin til upprunans.

Skilaboð úr biðröðinni eru unnin af Logstash, send í gegnum síuna og úttakstappið. Þegar staðfesting berst frá úttakinu um að annálinn hafi verið sendur, fjarlægir Logstash úrvinnsluskrána úr biðröðinni. Ef Logstash hættir verða öll óunnin skilaboð og skilaboð sem engin staðfesting hefur borist fyrir eftir í biðröðinni og Logstash mun halda áfram að vinna úr þeim næst þegar það byrjar.

aðlögun

Stillanlegt með lyklum í skránni C:Logstashconfiglogstash.yml:

• queue.type: (möguleg gildi - persisted и memory (default)).
• path.queue: (slóð að möppunni með biðröð skrám, sem eru sjálfgefið geymdar í C:Logstashqueue).
• queue.page_capacity: (hámarkssíðustærð biðraðar, sjálfgefið gildi er 64mb).
• queue.drain: (true/false - virkjar/slökkva á því að stöðva biðröðvinnslu áður en Logstash er lokað. Ég mæli ekki með því að virkja það, því þetta hefur bein áhrif á hraða lokunar netþjóns).
• queue.max_events: (hámarksfjöldi atburða í biðröð, sjálfgefið er 0 (ótakmarkað)).
• queue.max_bytes: (hámarksstærð biðraðar í bætum, sjálfgefið - 1024mb (1gb)).

Ef stillt er queue.max_events и queue.max_bytes, þá hættir að taka við skilaboðum í biðröðina þegar gildi einhverra þessara stillinga er náð. Lærðu meira um viðvarandi biðraðir hér.

Dæmi um þann hluta logstash.yml sem ber ábyrgð á að setja upp biðröðina:

queue.type: persisted
queue.max_bytes: 10gb

aðlögun

Logstash uppsetningin samanstendur venjulega af þremur hlutum, sem bera ábyrgð á mismunandi stigum vinnslu komandi annála: móttöku (inntakshluti), þáttun (síuhluti) og sendingu í Elastic (úttakshluti). Hér að neðan munum við skoða hvert þeirra nánar.

inntak

Við tökum á móti streymi sem kemur inn með hráum logs frá filebeat umboðsmönnum. Það er þessi viðbót sem við tilgreinum í inntakshlutanum:

input {
  beats {
    port => 5044
  }
}

Eftir þessa stillingu byrjar Logstash að hlusta á höfn 5044 og vinnur úr þeim í samræmi við stillingar síuhlutans þegar hann tekur á móti annálum. Ef nauðsyn krefur geturðu sett rásina fyrir móttöku annála frá filebit í SSL. Lestu meira um stillingar beats viðbóta hér.

síur

Allir textaskrár sem áhugaverðar eru til vinnslu sem Exchange býr til eru á csv sniði með þeim reitum sem lýst er í sjálfri annálaskránni. Til að flokka csv-skrár býður Logstash okkur upp á þrjú viðbætur: kryfja, csv og grok. Sá fyrsti er mestur skjót, en tekst á við að flokka aðeins einföldustu logs.
Til dæmis mun það skipta eftirfarandi færslu í tvennt (vegna kommu inni í reitnum), sem er ástæðan fyrir því að skráin verður ranglega þáttuð:

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

Það er hægt að nota við þáttun annála, til dæmis IIS. Í þessu tilviki gæti síuhlutinn litið svona út:

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 stillingar leyfa þér að nota skilyrtar yfirlýsingar, þannig að við getum aðeins sent logs sem voru merktir með filebeat taginu til dissect viðbótarinnar IIS. Inni í viðbótinni pössum við reitgildin við nöfn þeirra, eyðum upprunalega reitnum message, sem innihélt færslu úr skránni, og við getum bætt við sérsniðnum reit sem mun til dæmis innihalda nafn forritsins sem við söfnum annálum úr.

Þegar um er að ræða rakningarskrár er betra að nota csv viðbótina; það getur rétt unnið úr flóknum reitum:

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

Inni í viðbótinni pössum við reitgildin við nöfn þeirra, eyðum upprunalega reitnum message (og líka sviðum tenant-id и schema-version), sem innihélt færslu úr skránni, og við getum bætt við sérsniðnum reit, sem mun til dæmis innihalda nafn forritsins sem við söfnum annálum úr.

Við brottför frá síunarstigi munum við fá skjöl í fyrstu nálgun, tilbúin fyrir sjón í Kibana. Við munum sakna eftirfarandi:

• Tölureitir verða þekktir sem texti, sem kemur í veg fyrir aðgerðir á þeim. Nefnilega túnin time-taken IIS log, svo og reiti recipient-count и total-bites Log Rekja.
• Staðlaður tímastimpill skjalsins mun innihalda tímann sem vinnsluskráin var unnin, ekki tímann sem hún var skrifuð á miðlarahlið.
• Field recipient-address mun líta út eins og einn byggingarstaður, sem gerir ekki ráð fyrir greiningu til að telja viðtakendur bréfa.

Það er kominn tími til að bæta smá töfrum við vinnsluferlið annála.

Umbreytir talnareitum

Krufningsviðbótin hefur möguleika convert_datatype, sem hægt er að nota til að breyta textareit í stafrænt snið. Til dæmis, svona:

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

Það er þess virði að muna að þessi aðferð hentar aðeins ef reiturinn mun örugglega innihalda streng. Valmöguleikinn vinnur ekki úr núllgildum úr sviðum og sendir frá sér undantekningu.

Fyrir rekja logs, það er betra að nota ekki svipaða umbreyta aðferð, þar sem reitirnir recipient-count и total-bites gæti verið tómt. Til að umbreyta þessum reitum er betra að nota viðbót stökkbreyta:

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

Skiptir recipient_address í einstaka viðtakendur

Þetta vandamál er einnig hægt að leysa með því að nota mutate viðbótina:

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

Breyting á tímastimpli

Þegar um er að ræða rakningarskrár er vandamálið mjög auðvelt að leysa með viðbótinni dagsetning, sem mun hjálpa þér að skrifa á sviði timestamp dagsetningu og tíma á tilskildu sniði úr reitnum date-time:

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

Þegar um er að ræða IIS annála þurfum við að sameina svæðisgögn date и time Notaðu mutate viðbótina, skráðu tímabeltið sem við þurfum og settu þennan tímastimpil inn timestamp nota dagsetningarviðbótina:

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

Output

Úttakshlutinn er notaður til að senda unnar annála til móttakarans. Ef sent er beint til Elastic er viðbót notuð teygjanám, sem tilgreinir vistfang netþjóns og sniðmát fyrir vísitöluheiti til að senda myndað skjal:

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

Lokastilling

Endanleg uppsetning mun líta svona út:

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

Gagnlegar hlekkir:

• Hvernig á að setja upp OpenJDK 11 á Windows?
• Sækja Logstash
• Elastic notar afskrifaðan valkost UseConcMarkSweepGC #36828
• NSSM
• Viðvarandi biðraðir
• Beats inntak viðbót
• Logstash náungi, hvar er keðjusögin mín? Ég þarf að kryfja annálana mína
• Krufnasíuviðbót
• conditionals
• Stökkbreytingar síu viðbót
• Viðbót fyrir dagsetningarsíu
• Elasticsearch úttaksviðbót

Heimild: www.habr.com