ProHoster > Dienoraštis > Administravimas > Praktinis ELK pritaikymas. „Logstash“ nustatymas

Praktinis ELK pritaikymas. „Logstash“ nustatymas

įvedimas

Diegdami kitą sistemą susidūrėme su būtinybe apdoroti daugybę skirtingų žurnalų. Priemone pasirinkta ELK. Šiame straipsnyje bus aptarta mūsų patirtis kuriant šį krūvą.

Nekeliame tikslo apibūdinti visų jo galimybių, bet norime koncentruotis konkrečiai į praktinių problemų sprendimą. Taip yra dėl to, kad nors dokumentacijos ir paruoštų vaizdų yra gana daug, spąstų yra gana daug, bent jau mes jas radome.

Mes įdiegėme dėklą naudodami docker-compose. Be to, mes turėjome gerai parašytą docker-compose.yml, kuri leido mums beveik be problemų padidinti krūvą. Ir mums atrodė, kad pergalė jau arti, dabar šiek tiek pakoreguosime, kad atitiktų mūsų poreikius ir viskas.

Deja, bandymas sukonfigūruoti sistemą gauti ir apdoroti žurnalus iš mūsų programos nebuvo iš karto sėkmingas. Todėl nusprendėme, kad verta panagrinėti kiekvieną komponentą atskirai, o tada grįžti prie jų sąsajų.

Taigi, mes pradėjome nuo logstash.

Aplinka, diegimas, „Logstash“ paleidimas konteineryje

Diegimui naudojame docker-compose; čia aprašyti eksperimentai buvo atlikti naudojant MacOS ir Ubuntu 18.0.4.

Logstash vaizdas, užregistruotas mūsų originaliame docker-compose.yml yra docker.elastic.co/logstash/logstash:6.3.2

Naudosime eksperimentams.

Parašėme atskirą docker-compose.yml, kad paleistume logstash. Žinoma, buvo galima paleisti vaizdą iš komandinės eilutės, bet mes sprendėme konkrečią problemą, kur viską paleidžiame nuo docker-compose.

Trumpai apie konfigūracijos failus

Kaip matyti iš aprašymo, logstash gali būti paleistas vienam kanalui, tokiu atveju jis turi perduoti *.conf failą, arba keliems kanalams, tokiu atveju jis turi perduoti pipelines.yml failą, kuris savo ruožtu , bus nuoroda į kiekvieno kanalo .conf failus.
Ėjome antruoju keliu. Mums jis atrodė universalesnis ir keičiamo dydžio. Todėl sukūrėme pipelines.yml ir sukūrėme pipelines katalogą, kuriame kiekvienam kanalui įdėsime .conf failus.

Konteinerio viduje yra kitas konfigūracijos failas - logstash.yml. Neliečiame, naudojame tokį, koks yra.

Taigi, mūsų katalogo struktūra:

Praktinis ELK pritaikymas. „Logstash“ nustatymas

Norėdami gauti įvesties duomenis, kol kas darome prielaidą, kad tai yra tcp 5046 prievade, o išvestims naudosime stdout.

Čia yra paprasta pirmojo paleidimo konfigūracija. Kadangi pradinė užduotis yra paleisti.

Taigi, turime šį docker-compose.yml

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

Ką mes čia matome?

  1. Tinklai ir tomai buvo paimti iš originalaus docker-compose.yml (to, kur paleistas visas stackas) ir manau, kad jie neturi didelės įtakos bendram vaizdui.
  2. Sukuriame vieną (-as) logstash paslaugą (-as) iš docker.elastic.co/logstash/logstash:6.3.2 vaizdo ir pavadiname ją logstash_one_channel.
  3. Persiunčiame 5046 prievadą konteinerio viduje į tą patį vidinį prievadą.
  4. Mes susiejame vamzdžio konfigūracijos failą ./config/pipelines.yml su failu /usr/share/logstash/config/pipelines.yml konteinerio viduje, kur logstash jį pasiims ir padarys tik skaitomą, bet kuriuo atveju.
  5. Katalogą ./config/pipelines, kuriame turime failus su kanalo nustatymais, susiejame su /usr/share/logstash/config/pipelines katalogu ir padarome jį tik skaitomą.

Praktinis ELK pritaikymas. „Logstash“ nustatymas

Pipelines.yml failą

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

Čia aprašytas vienas kanalas su HABR identifikatoriumi ir kelias į jo konfigūracijos failą.

Ir galiausiai failas „./config/pipelines/habr_pipeline.conf“

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

Kol kas nesigilinkime į jo aprašymą, pabandykime jį paleisti:

docker-compose up

Ką mes matome?

Konteineris pradėtas. Galime patikrinti jo veikimą:

echo '13123123123123123123123213123213' | nc localhost 5046

Ir mes matome atsakymą konteinerio konsolėje:

Praktinis ELK pritaikymas. „Logstash“ nustatymas

Tačiau tuo pat metu mes taip pat matome:

logstash_one_channel | [2019-04-29T11:28:59,790][KLAIDA][logstash.licensechecker.licensereader] Nepavyko gauti licencijos informacijos iš licencijų serverio {:message=>„Elasticsearch nepasiekiamas: [http://elasticsearch:9200/][Manticore“ ::ResolutionFailure] elasticsearch", ...

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Dujotiekis sėkmingai paleistas {:pipeline_id=>.monitoring-logstash", :thread=>"# "}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Dujotiekiai veikia {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[ ]}
logstash_one_channel | [2019-04-29T11:29:00,015][KLAIDA][logstash.inputs.metrics] X-Pack įdiegtas Logstash, bet ne Elasticsearch. Įdiekite X-Pack Elasticsearch, kad galėtumėte naudoti stebėjimo funkciją. Gali būti ir kitų funkcijų.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Sėkmingai paleistas Logstash API galutinis taškas {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Vykdoma būklės patikra, siekiant išsiaiškinti, ar veikia Elasticsearch ryšys {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Bandyta atkurti ryšį su mirusiu ES egzemplioriumi, bet įvyko klaida. {:url=>“elastinga paieška:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch nepasiekiamas: [http://elasticsearch:9200/][Manticore::Re]solution elasticsearch"}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Vykdoma būklės patikra, siekiant išsiaiškinti, ar veikia Elasticsearch ryšys {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Bandyta atkurti ryšį su mirusiu ES egzemplioriumi, bet įvyko klaida. {:url=>“elastinga paieška:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch nepasiekiamas: [http://elasticsearch:9200/][Manticore::Re]solution elasticsearch"}

O mūsų rąstas visą laiką šliaužia.

Čia žaliai paryškinau pranešimą, kad dujotiekis sėkmingai paleistas, raudonai paryškinau klaidos pranešimą ir geltonai pranešimą apie bandymą susisiekti elastinga paieška: 9200.
Taip atsitinka todėl, kad logstash.conf, įtrauktas į vaizdą, yra elasticsearch prieinamumo patikrinimas. Galų gale, logstash daro prielaidą, kad jis veikia kaip Elk kamino dalis, bet mes jį atskirome.

Dirbti galima, bet tai nėra patogu.

Sprendimas yra išjungti šį patikrinimą naudojant XPACK_MONITORING_ENABLED aplinkos kintamąjį.

Pakeiskime docker-compose.yml ir paleiskime dar kartą:

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

Dabar viskas gerai. Talpykla paruošta eksperimentams.

Kitoje konsolėje galime įvesti dar kartą:

echo '13123123123123123123123213123213' | nc localhost 5046

Ir matyti:

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

Darbas viename kanale

Taigi paleidome. Dabar galite skirti laiko pačiam logstash konfigūravimui. Kol kas nelieskime pipelines.yml failo, pažiūrėkime, ką galime gauti dirbdami su vienu kanalu.

Turiu pasakyti, kad bendras darbo su kanalo konfigūracijos failu principas yra gerai aprašytas oficialiame vadove, čia čia
Jei norite skaityti rusiškai, naudojome šią straipsnis(tačiau užklausos sintaksė yra sena, turime į tai atsižvelgti).

Eikime nuosekliai iš įvesties skyriaus. Jau matėme darbą su TCP. Kas čia dar gali būti įdomaus?

Išbandykite pranešimus naudodami širdies plakimą

Yra tokia įdomi galimybė generuoti automatinius testinius pranešimus.
Norėdami tai padaryti, įvesties skiltyje turite įjungti širdies pupelių papildinį.

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

Įjunkite, pradėkite gauti kartą per minutę 

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

Jei norime gauti dažniau, turime pridėti intervalo parametrą.
Taip kas 10 sekundžių gausime pranešimą.

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

Duomenų gavimas iš failo

Taip pat nusprendėme pažvelgti į failo režimą. Jei jis gerai veikia su failu, galbūt nereikia jokio agento, bent jau vietiniam naudojimui.

Pagal aprašymą darbo režimas turėtų būti panašus į tail -f, t.y. skaito naujas eilutes arba, kaip parinktis, nuskaito visą failą.

Taigi, ką mes norime gauti:

  1. Norime gauti eilutes, kurios pridedamos prie vieno žurnalo failo.
  2. Norime gauti duomenis, įrašytus į kelis žurnalo failus, kartu atskirdami, kas iš kur gaunama.
  3. Norime užtikrinti, kad iš naujo paleidus „logstash“, jis daugiau negautų šių duomenų.
  4. Norime patikrinti, ar jei logstash išjungtas, o duomenys ir toliau įrašomi į failus, tada jį paleisdami gausime šiuos duomenis.

Norėdami atlikti eksperimentą, į docker-compose.yml pridėkime dar vieną eilutę, atidarydami katalogą, į kurį įdėjome failus.

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

Ir pakeiskite įvesties skyrių habr_pipeline.conf

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

Pradėkime:

docker-compose up

Norėdami sukurti ir rašyti žurnalo failus, naudosime komandą:


echo '1' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello 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 | }

Taip, tai veikia!

Tuo pačiu matome, kad automatiškai įtraukėme kelio lauką. Tai reiškia, kad ateityje galėsime pagal jį filtruoti įrašus.

Pabandykime dar kartą:

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

O dabar prie kito failo:

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

Puiku! Failas paimtas, kelias nurodytas teisingai, viskas gerai.

Sustabdykite logstash ir pradėkite iš naujo. Palaukime. Tyla. Tie. Daugiau šių įrašų negauname.

O dabar pats drąsiausias eksperimentas.

Įdiekite logstash ir vykdykite:

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

Dar kartą paleiskite logstash ir žiūrėkite:

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

Sveika! Viskas buvo paimta.

Tačiau turime jus įspėti dėl šių dalykų. Jei konteineris su logstash ištrintas (docker stop logstash_one_channel && docker rm logstash_one_channel), niekas nebus paimtas. Failo, iki kurio jis buvo nuskaitytas, padėtis buvo išsaugota konteinerio viduje. Jei paleisite jį nuo nulio, jis priims tik naujas eilutes.

Esamų failų skaitymas

Tarkime, logstash paleidžiame pirmą kartą, bet jau turime žurnalus ir norėtume juos apdoroti.
Jei paleisime logstash su įvesties skyriumi, kurį naudojome aukščiau, nieko negausime. Logstash apdoros tik naujas eilutes.

Norėdami, kad esamų failų eilutės būtų ištrauktos, į įvesties skyrių turėtumėte pridėti papildomą eilutę:

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

Be to, yra niuansas: tai turi įtakos tik naujiems failams, kurių logstash dar nematė. Tiems patiems failams, kurie jau buvo logstash matymo lauke, jis jau prisiminė jų dydį ir dabar juose įves tik naujus įrašus.

Sustokime čia ir išnagrinėkime įvesties skyrių. Dar yra daug variantų, bet kol kas to mums pakanka tolesniems eksperimentams.

Maršrutas ir duomenų transformavimas

Pabandykime išspręsti šią problemą, tarkime, kad turime pranešimų iš vieno kanalo, kai kurie iš jų yra informaciniai, o kiti yra klaidų pranešimai. Jie skiriasi pagal etiketę. Kai kurie yra INFO, kiti yra KLAIDA.

Turime juos atskirti prie išėjimo. Tie. Viename kanale rašome informacinius pranešimus, kitame – klaidų pranešimus.

Norėdami tai padaryti, iš įvesties skyriaus pereikite prie filtro ir išvesties.

Naudodami filtrų skiltį analizuosime gaunamą žinutę, iš jos gaudami maišą (rakto-reikšmių poras), su kuria jau galime dirbti, t.y. išmontuoti pagal sąlygas. O išvesties skiltyje parinksime pranešimus ir išsiųsime kiekvieną į savo kanalą.

Nagrinėja pranešimą su grok

Norint išanalizuoti teksto eilutes ir iš jų gauti laukų rinkinį, filtrų skiltyje yra specialus įskiepis - grok.

Nenustatydamas sau tikslo čia pateikti išsamų jo aprašymą (šiuo atveju remiuosi oficialius dokumentus), pateiksiu savo paprastą pavyzdį.

Norėdami tai padaryti, turite nuspręsti dėl įvesties eilučių formato. Turiu juos tokius:

1 INFO pranešimas1
2 KLAIDOS pranešimas2

Tie. Pirmiausia pateikiamas identifikatorius, tada INFO/ERROR, tada koks nors žodis be tarpų.
Tai nėra sunku, bet pakanka suprasti veikimo principą.

Taigi, grok įskiepio filtrų skiltyje turime apibrėžti eilučių analizavimo modelį.

Tai atrodys taip:

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

Iš esmės tai yra reguliari išraiška. Naudojami paruošti raštai, tokie kaip INT, LOGLEVEL, WORD. Jų aprašymą ir kitus modelius rasite čia čia

Dabar, praeinant per šį filtrą, mūsų eilutė pavirs į trijų laukų maišą: message_id, message_type, message_text.

Jie bus rodomi išvesties skiltyje.

Pranešimų nukreipimas į išvesties sekciją naudojant komandą if

Išvesties skyriuje, kaip prisimename, ketinome padalinti pranešimus į du srautus. Kai kurie - kurie yra iNFO, bus išvesti į konsolę, o su klaidomis išvesime į failą.

Kaip atskirti šias žinutes? Problemos sąlyga jau siūlo sprendimą – juk jau turime tam skirtą laukelį message_type, kuris gali turėti tik dvi reikšmes: INFO ir ERROR. Remdamiesi tuo, pasirinksime naudodami teiginį if.

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

Šiame skyriuje rasite darbo su laukais ir operatoriais aprašymą oficialus vadovas.

Dabar apie pačią faktinę išvadą.

Konsolės išvestis, čia viskas aišku – stdout {}

Tačiau išvestis į failą – atminkite, kad visa tai vykdome iš konteinerio ir kad failas, kuriame rašome rezultatą, būtų pasiekiamas iš išorės, turime atidaryti šį katalogą docker-compose.yml.

Iš viso:

Mūsų failo išvesties skyrius atrodo taip:


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

Į docker-compose.yml pridedame kitą išvesties tomą:

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

Paleidžiame, išbandome ir matome padalijimą į du srautus.

Šaltinis: www.habr.com

Добавить комментарий