🥇Մենք ընկերներ ենք ELK-ի և Exchange-ի հետ։ Մաս 2

Ես շարունակում եմ իմ պատմությունը, թե ինչպես ընկերներ ձեռք բերել Exchange և ELK (սկիզբ այստեղ) Հիշեցնեմ, որ այս համադրությունը առանց վարանելու ընդունակ է մշակել շատ մեծ քանակությամբ գերաններ։ Այս անգամ մենք կխոսենք այն մասին, թե ինչպես ստիպել Exchange-ին աշխատել Logstash և Kibana բաղադրիչների հետ:

Logstash-ը ELK stack-ում օգտագործվում է տեղեկամատյանները խելամտորեն մշակելու և փաստաթղթերի տեսքով 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 գործընթացի համար RAM-ի հատկացման համար: Խորհուրդ եմ տալիս նշել սերվերի RAM-ի կեսը: Եթե այն ունի 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 նավահանգստում:

Տեղադրման վերջնական քայլը. գործարկել Logstash-ը որպես Windows ծառայություն: Դա կարելի է անել, օրինակ, օգտագործելով փաթեթը ԱՎSS:

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

սխալների հանդուրժողականություն

Աղբյուրի սերվերից փոխանցելիս տեղեկամատյանների անվտանգությունն ապահովված է Persistent Queues մեխանիզմով:

Ինչպես է այն աշխատում

Մատյանների մշակման ժամանակ հերթերի դասավորությունը հետևյալն է. մուտք → հերթ → զտիչ + ելք։

Մուտքային հավելվածը տվյալներ է ստանում տեղեկամատյան աղբյուրից, գրում է այն հերթում և ուղարկում հաստատում, որ տվյալները ստացվել են աղբյուրին:

Հերթից ստացված հաղորդագրությունները մշակվում են 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 գործակալներից: Հենց այս plugin-ն է, որը մենք նշում ենք մուտքագրման բաժնում.

input {
  beats {
    port => 5044
  }
}

Այս կոնֆիգուրացիայից հետո Logstash-ը սկսում է լսել 5044 նավահանգիստը և տեղեկամատյանները ստանալիս մշակում է դրանք ըստ զտիչի բաժնի կարգավորումների։ Անհրաժեշտության դեպքում կարող եք SSL-ով փաթեթավորել ալիքը՝ ֆայլբիթից տեղեկամատյաններ ստանալու համար: Կարդացեք ավելին beats հավելվածի կարգավորումների մասին այստեղ.

ֆիլտր

Բոլոր տեքստային տեղեկամատյանները, որոնք հետաքրքիր են մշակման համար, որոնք ստեղծում է Exchange-ը, csv ձևաչափով են՝ բուն log ֆայլում նկարագրված դաշտերով: 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 կոնֆիգուրացիան թույլ է տալիս օգտագործել պայմանական հայտարարություններ, այնպես որ մենք կարող ենք միայն այն տեղեկամատյանները, որոնք հատկորոշված են եղել filebeat պիտակով dissect plugin-ին: IIS. Պլագինի ներսում մենք համապատասխանեցնում ենք դաշտի արժեքները դրանց անունների հետ, ջնջում ենք բնօրինակ դաշտը message, որը պարունակում էր մուտքագրում մատյանից, և մենք կարող ենք ավելացնել հատուկ դաշտ, որը, օրինակ, կպարունակի հավելվածի անունը, որտեղից մենք հավաքում ենք տեղեկամատյանները։

Հետևելու տեղեկամատյանների դեպքում ավելի լավ է օգտագործել csv plugin-ը, այն կարող է ճիշտ մշակել բարդ դաշտերը.

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 նման կլինի մեկ շինհրապարակի, որը թույլ չի տալիս վերլուծություններ կատարել նամակներ ստացողներին հաշվելու համար։

Ժամանակն է մի փոքր կախարդանք ավելացնել տեղեկամատյանների մշակման գործընթացին:

Թվային դաշտերի փոխակերպում

The dissect plugin-ն ունի տարբերակ convert_datatype, որը կարող է օգտագործվել տեքստային դաշտը թվային ձևաչափի փոխակերպելու համար։ Օրինակ, այսպես.

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

Հարկ է հիշել, որ այս մեթոդը հարմար է միայն այն դեպքում, եթե դաշտը անպայման կպարունակի տող: Տարբերակը չի մշակում Null արժեքները դաշտերից և բացառություն է անում:

Տեղեկամատյանները հետևելու համար ավելի լավ է չօգտագործել փոխակերպման նմանատիպ մեթոդ, քանի որ դաշտերը recipient-count и total-bites կարող է դատարկ լինել: Այս դաշտերը փոխարկելու համար ավելի լավ է օգտագործել plugin մուտացիայի ենթարկել:

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

Recipient_address-ի բաժանումը առանձին հասցեատերերի

Այս խնդիրը կարող է լուծվել նաև մուտացիայի հավելվածի միջոցով.

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

Ժամացույցի փոփոխություն

Հետևելու տեղեկամատյանների դեպքում խնդիրը շատ հեշտությամբ լուծվում է plugin-ի կողմից ամսաթիվ, որը կօգնի ձեզ գրել դաշտում timestamp ամսաթիվը և ժամը դաշտից պահանջվող ձևաչափով date-time:

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

IIS տեղեկամատյանների դեպքում մեզ անհրաժեշտ կլինի միավորել դաշտային տվյալները date и time օգտագործելով mutate plugin-ը, գրանցենք մեզ անհրաժեշտ ժամային գոտին և տեղադրենք այս ժամային դրոշմը timestamp օգտագործելով ամսաթիվը 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" ]
}

Արտադրողականություն

Արդյունք բաժինը օգտագործվում է մշակված տեղեկամատյանները տեղեկամատյան ընդունիչին ուղարկելու համար: Ուղիղ Elastic ուղարկելու դեպքում օգտագործվում է plugin առաձգական որոնում, որը սահմանում է սերվերի հասցեն և ինդեքսի անվան ձևանմուշը՝ ստեղծված փաստաթուղթն ուղարկելու համար.

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

Օգտակար հղումներ

Source: www.habr.com

Մենք ընկերներ ենք ELK-ի և Exchange-ի հետ։ Մաս 2

Տեղակայում

Logstash փաթեթի տեղադրում և կարգավորում

սխալների հանդուրժողականություն

Ինչպես է այն աշխատում

հարմարեցում

հարմարեցում

Մուտքային

ֆիլտր

Թվային դաշտերի փոխակերպում

Recipient_address-ի բաժանումը առանձին հասցեատերերի

Ժամացույցի փոփոխություն

Արտադրողականություն

Վերջնական կոնֆիգուրացիա

Добавить комментарий Չեղարկել պատասխանել