ProHoster > Blog > Administrim > PĂ«rdorimi praktik i ELK. Konfiguroni logstash

PĂ«rdorimi praktik i ELK. Konfiguroni logstash

Hyrje

Duke hapjesh një sistem të ri, u përballëm me nevojën për të përpunuar një sasi të madhe logesh të ndryshme. Si mjet zgjodhëm ELK. Në këtë artikull do të flasim mbi përvojën tonë në konfigurimin e këtij staku.

Nuk kemi për qëllim të përshkruajmë të gjitha mundësitë e tij, por duam të përqendrohemi në zgjidhjen e problemeve praktike. Kjo është e nevojshme sepse, pavarësisht se ekziston një sasi e madhe dokumentacioni dhe imazhe tashmë të gatshme, ka mjaft pengesa, të paktën ato u zbuluan tek ne.

Ne e vendosëm stakun përmes docker-compose. Më shumë se kaq, kishim një docker-compose.yml të shkruar mirë, i cili na lejoi të ngrinim stakun pothuajse pa probleme. Dhe na dukej se fitorja ishte afër, tani duhet të bëjmë disa rregullime për nevojat tona dhe gjithçka do ishte në rregull.

Fatkeqësisht, përpjekja për të rregulluar sistemin për marrjen dhe përpunimin e logeve nga aplikacioni ynë, nuk arriti sukses menjëherë. Prandaj, vendosëm që të studiojmë çdo komponent veç e veç, dhe më pas të kthehemi te lidhjet e tyre.

KĂ«shtu, filluam me logstash.

Mjedisi, vendosja, fillimi i Logstash në kontejner

Për instalimin përdorim docker-compose, eksperimentet e përshkruara këtu u zhvilluan në MacOS dhe Ubuntu 18.0.4.

Imazhi logstash, i cili ishte i shkruar në docker-compose.yml tonë, është docker.elastic.co/logstash/logstash:6.3.2

Këtë do ta përdorim për eksperimentet.

Për të nisur logstash, shkruam një docker-compose.yml të veçantë. Sigurisht, mund të ishte nisur imazhi nga linja e komandave, por ne po zgjidhim një detyrë të caktuar, ku gjithçka niset nga docker-compose.

Përshkrim i shpejtë mbi skedarët e konfigurimit

Siç del nga përshkrimi, logstash mund të niset si për një kanal, në këtë rast, i nevojitet një skedar *.conf ose për disa kanale, në këtë rast i nevojitet një skedar pipelines.yml, i cili nga ana e tij do të referojë skedarët .conf për çdo kanal.
Ne zgjodhëm rrugën e dytë. Na dukej se ishte më universale dhe e shkallëzuar. Prandaj, krijuam pipelines.yml, dhe bëmë një direktor për pipelines, ku do të vendosim skedarët .conf për çdo kanal.

Brenda konteinerit ka njĂ« skedar tjetĂ«r konfigurimi — logstash.yml. Ne nuk e prekim, e pĂ«rdorim siç Ă«shtĂ«.

Kështu, struktura e katalogëve tanë është:

PĂ«rdorimi praktik i ELK. Konfiguroni logstash

Për marrjen e të dhënave, për tani e konsiderojmë se është tcp në portin 5046, ndërsa për daljen do të përdorim stdout.

Kjo është një konfigurim i thjeshtë për fillimin e parë. Sepse në fund të fundit detyra fillestare është të nisim.

Kështu, kemi një docker-compose.yml të tillë.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

ÇfarĂ« shohim kĂ«tu?

  1. Networks dhe volumes janë marrë nga docker-compose.yml origjinal (ai ku starti i plotë ekzekutohet) dhe mendoj se nuk ndikojnë shumë në pamjen e përgjithshme.
  2. Ne krijojmë një shërbim (services) logstash, nga imazhi docker.elastic.co/logstash/logstash:6.3.2 dhe i japim asaj emrin logstash_one_channel.
  3. Ne kalojmë brenda konteinerit portin 5046, në të njëjtin port të brendshëm.
  4. Ne po lidhim skedarin tonë të konfigurimit të kanaleve ./config/pipelines.yml në skedarin /usr/share/logstash/config/pipelines.yml brenda konteinerit, nga ku do të merret logstash dhe e bëjmë atë read-only, vetëm për të qenë të sigurt.
  5. Ne lidhim direktorinë ./config/pipelines, ku kemi skedarët me konfigurimet e kanaleve, në direktorine /usr/share/logstash/config/pipelines dhe e bëjmë atë gjithashtu read-only.

PĂ«rdorimi praktik i ELK. Konfiguroni logstash

Skedari pipelines.yml

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

Këtu përshkruhet një kanal me identifikuesin HABR dhe rruga në skedarin e tij të konfigurimit.

Dhe përfundimisht skedari "./config/pipelines/habr_pipeline.conf"

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

Nuk do të thellojmë përshkrimin e tij për momentin, le të provojmë ta nisim:

docker-compose up

ÇfarĂ« shohim?

Konteineri është nisur. Mund të kontrollojmë funksionimin e tij:

echo '13123123123123123123123213123213' | nc localhost 5046

Dhe shohim në konsolën e konteinerit përgjigjen:

PĂ«rdorimi praktik i ELK. Konfiguroni logstash

Por duke e parë gjithashtu:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Unable to retrieve license information from license server {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch", 


logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Pipeline started successfully {:pipeline_id=>".monitoring-logstash", :thread=>"#"}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Pipelines running {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[]}
logstash_one_channel | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics ] X-Pack is installed on Logstash but not on Elasticsearch. Please install X-Pack on Elasticsearch to use the monitoring feature. Other features may be available.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://elasticsearch:9200/, :path=>"/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Përpjekur për të ringjallur lidhjen me instancën e vdekur të ES, por u shfaq një gabim. {:url=>«elasticsearch:9200/», :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>«Elasticsearch e pa arritshme: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Po kryejmë kontrollin e shëndetit për të parë nëse një lidhje me Elasticsearch po punon {:healthcheck_url=>http://elasticsearch:9200/, :path=>"/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Përpjekur për të ringjallur lidhjen me instancën e vdekur të ES, por u shfaq një gabim. {:url=>«elasticsearch:9200/», :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>«Elasticsearch e pa arritshme: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

Dhe ditari ynë ngjitet vazhdimisht lart.

KĂ«tu e kam theksuar me ngjyrĂ« tĂ« gjelbĂ«r mesazhin qĂ« pipeline ka nisur me sukses, tĂ« kuqe — mesazhin e gabimit dhe tĂ« verdhĂ« — mesazhin e pĂ«rpjekjes pĂ«r t'u lidhur me elasticsearch:9200.
Kjo ndodh sepse në logstash.conf, që është përfshirë në imazh, ka një kontroll për aksesin në elasticsearch. Sepse logstash supozon se punon brenda stakut Elk, ndërsa ne e kemi ndarë.

Mund të punosh, por nuk është e përshtatshme.

Zgjidhja është të çaktivizosh këtë kontroll përmes variablit të mjedisit XPACK_MONITORING_ENABLED.

Do të bëjmë një ndryshim në docker-compose.yml dhe do ta nisim përsëri:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

Tani, gjithçka është në rregull. Konteineri është gati për eksperimentet.

Mund të shkruajmë përsëri në terminalin përballet:

echo '13123123123123123123123213123213' | nc localhost 5046

Dhe të shohim:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

Puna në kuadër të një kanali

Dhe kështu, kemi filluar. Tani, mund të dedikojmë kohë për konfigurimin e direkt logstash. Nuk do të prekim ende skedarin pipelines.yml, do të shohim se çfarë mund të marrim duke punuar me një kanal.

Duhet të themi se principi i përgjithshëm për punën me skedarin e konfigurimit të kanalit është përshkruar mirë në udhëzimin zyrtar, këtu këtu
Nëse dëshiron të lexosh në shqip, ne e kemi përdorur këtë artikullin(por sintaksa e kërkesave atje është e vjetër, duhet ta kemi parasysh).

Le tĂ« ecim pĂ«rpara nga seksioni Input. Ne kemi parĂ« punĂ«n me tcp. ÇfarĂ« mund tĂ« jetĂ« interesante kĂ«tu?

Mesazhe testuese, duke përdorur heartbeat

Ka ekziston një mundësi interesante për të gjeneruar mesazhe automatike testuese.
Për këtë, në seksionin input duhet të aktivizohet plugini heartbean.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  }

E aktivizojmë, fillojmë të marrim çdo minutë

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Përshëndetje Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

Dëshirojmë të marrim më shpesh, duhet të shtojmë parametrin interval.
Kështu do të marrim një mesazh çdo 10 sekonda.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

Marrja e të dhënave nga një skedar

Po ashtu, vendosëm të shikojmë modin file. Nëse funksionon mirë me skedarin, ndoshta nuk do të nevojitet asnjë agjent, së paku për përdorim lokal.

Përshkrimi thotë se mënyra e punës duhet të jetë e ngjashme me tail -f, pra lexon rreshtat e rinj ose, si opsion, lexon të gjithë skedarin.

Pra, çfarë duam të arrijmë:

  1. Dëshirojmë të marrim rreshtat që shkruhen në një skedar log.
  2. Dëshirojmë të marrim të dhënat që shkruhen në disa skedarë log, duke pasur mundësinë të ndajmë se çfarë është marrë nga cila burim.
  3. Dëshirojmë të verifikojmë se gjatë rivendosjes së logstash, ai nuk do të marrë këto të dhëna përsëri.
  4. Dëshirojmë të verifikojmë që nëse logstash është fikur dhe të dhënat vazhdojnë të shkruhen në skedarë, atëherë, kur ta fillojmë përsëri, do t'i marrim këto të dhëna.

Për të realizuar eksperimentin do të shtojmë një rresht të ri në docker-compose.yml, duke hapur drejtorinë ku i vendosim skedarët.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

Dhe do të modifikojmë seksionin input në habr_pipeline.conf

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

Nisemi:

docker-compose up

Për të krijuar dhe regjistruar skedarë log do të përdorim komandën:

echo '1' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Përshëndetje Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Eh, funksionon!

Në këtë mënyrë, ne shohim se është shtuar automatikisht fusha path. Kështu që në të ardhmen, do të mund të filtrojmë regjistrimet në bazë të saj.

Le të provojmë përsëri:

echo '2' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Tani në një file tjetër:

 echo '1' >> logs/number2.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

Shkëlqyer! Filei u mor, path u vendos saktë, gjithçka është mirë.

Të ndalojmë logstash dhe ta fillojmë përsëri. Të presim. Qetësi. Domethënë, regjistrimet nuk po marrim përsëri.

Dhe tani, eksperimenti më i guximshëm.

Vëmë logstash dhe ekzekutojmë:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

Përsëri fillojmë logstash dhe shohim:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

Urime! Gjithçka u mor.

Por, duhet të paralajmërojmë për këtë. Nëse kontejneri me logstash fshihet (docker stop logstash_one_channel && docker rm logstash_one_channel), atëherë asgjë nuk do të merret. Brenda kontejnerit ishte ruajtur pozita e files, deri ku ishte lexuar. Nëse e filloni "nga zero", atëherë do të pranojë vetëm rreshtat e rinj.

Leximi i skedave ekzistuese

Supozoni se ne e fillojmë për herë të parë logstash, por kemi tashmë loge dhe dëshirojmë t'i përpunojmë ato.
Nëse e fillojmë logstash me seksionin input që përdorëm më sipër, atëherë nuk do të marrim asgjë. Vetëm rreshtat e rinj do të përpunohen nga logstash.

Për të marrë rreshtat nga skedarët ekzistues, duhet të shtojmë një rresht tëmësues në seksionin input:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

Kaçtu, ka një nuancë, kjo vlen vetëm për skedarët e rinj që logstash nuk i ka parë më parë. Për ato skedare që tashmë kanë përfunduar në shikimin e logstash, ai e ka mbajtur mend madhësinë e tyre dhe tani do të marrë vetëm regjistrimet e reja në to.

Të ndalemi këtu për të studiuar seksionin input. Ka shumë mundësi të tjera, por për eksperimentet tona aktuale, kjo mjafton për tani.

Rrugëzimi dhe transformimi i të dhënave

Do të përpiqemi të zgjidhim detyrën e mëposhtme, le të supozojmë se kemi mesazhe nga një kanal, disa prej tyre janë informatike, ndërsa disa janë mesazhe për gabime. Ato dallohen nga tagu. Njëra INFO, tjetra ERROR.

Na nevojitet qĂ« nĂ« dalje t’i ndajmĂ« ato. Pra, mesazhet informatike i shkruajmĂ« nĂ« njĂ« kanal, dhe mesazhet pĂ«r gabime nĂ« njĂ« tjetĂ«r.

Për këtë, nga seksioni input kalojmë në filter dhe output.

Me ndihmën e seksionit filter ne do të analizojmë mesazhin hyrës, duke marrë prej tij hash (çifte çelës-vlerë), me të cilin mund të punojmë, pra, ta analizojmë sipas kushteve. Ndërsa në seksionin output, do të selektojmë mesazhet dhe do të dërgojmë secilën në kanalin e vet.

Analiza e mesazhit me ndihmën e grok

PĂ«r tĂ« analizuar vargjet tekstuale dhe pĂ«r tĂ« nxjerrĂ« prej tyre njĂ« grup fushash, nĂ« seksionin filter ka njĂ« plugin tĂ« veçantĂ« — grok.

Pa synuar të ofroj një përshkrim të detajuar të tij (për këtë e dërgoj tek dokumentacionin zyrtar), do të jap një shembull të thjeshtë.

Për këtë, duhet të përcaktojmë formatin e vargjeve hyrëse. Të miat janë të tilla:

1 INFO message1
2 ERROR message2

Pra, identifikuesi në vend të parë, pastaj INFO/ERROR, dhe më pas një fjalë pa hapësira.
Nuk është e komplikuar, por për të kuptuar principin e punës kjo mjafton.

Pra, në seksionin filter, në plugin-in grok duhet të përcaktojmë një model për analizimin e vargjeve tona.

Do të duket kështu:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  }

Në thelb, kjo është një shprehje e rregullt. Përdoren modele të gatshme si INT, LOGLEVEL, WORD. Përshkrimi i tyre, si dhe modele të tjera, mund të shikohet këtu këtu

Tani, duke kaluar përmes këtij filtri, vargu ynë do të shndërrohet në një hash me tre fusha: message_id, message_type, message_text.

Ato do të dalin në seksionin output.

Rrugëzimi i mesazheve në seksionin output me ndihmën e komandës if

NĂ« seksionin output, siç e kujtojmĂ«, ne kishim pĂ«r qĂ«llim tĂ« ndanim mesazhet nĂ« dy rrjedha. NjĂ«ra — ato iNFO, do t’i nxjerrim nĂ« konsolĂ«, ndĂ«rsa ato me gabime, do t’i nxjerrim nĂ« skedar.

Si ne i ndajmĂ« kĂ«to mesazhe? KĂ«rkesa e detyrĂ«s e sugjeron zgjidhjen — ne tashmĂ« kemi fushĂ«n e dedikuar message_type, e cila mund tĂ« marrĂ« vetĂ«m dy vlera INFO dhe ERROR. PikĂ«risht sipas kĂ«saj do tĂ« bĂ«jmĂ« zgjedhjen me ndihmĂ«n e operatorit if.

if [message_type] == "ERROR" {
        # Këtu e shkruajmë në skedarin
       } else
     {
      # Këtu e shkruajmë në stdout
    }

Përshkrimi i punës me fushat dhe operatorët mund të shihet në këtë seksion të manualit zyrtar.

Tani, rreth vetë daljes.

Dalja nĂ« konsolĂ«, kĂ«tu Ă«shtĂ« gjithçka e qartĂ« — stdout {}

Por dalja nĂ« skedar — kujtojmĂ«, se ne po e drejtojmĂ« kĂ«tĂ« nga kontejneri dhe qĂ« skedari ku e shkruajmĂ« rezultatin tĂ« jetĂ« i aksesueshĂ«m nga jashtĂ«, na nevojitet tĂ« hapim kĂ«tĂ« drejtor nĂ« docker-compose.yml.

Pra, në përfundim:

Seksioni output i skedarit tonë duket kështu:

output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "format i zakontë: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

Në docker-compose.yml shtojmë një vëllim tjetër për daljen:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

E fillojmë, provojmë, shohim ndarjen në dy rrjedha.

Burimi: habr.com

Blini hosting tĂ« besueshĂ«m pĂ«r faqe interneti me mbrojtje nga DDoS, serverĂ« VPS VDS đŸ”„ Blini hosting tĂ« besueshĂ«m pĂ«r faqe interneti me mbrojtje nga DDoS, serverĂ« VPS VDS | ProHoster