Úvod

Při nasazení dalšího systému jsme se setkali s potřebou zpracovávat velké množství různorodých logů. Jako nástroj jsme zvolili ELK. V tomto článku se podělíme o naše zkušenosti s nastavením tohoto stacku.

Nemáme za cíl popsat všechny jeho možnosti, ale chceme se soustředit na řešení praktických úloh. To je dáno tím, že při dostatku dokumentace a již připravených obrazů se objevuje dostatek zádrhelů, minimálně u nás se ukázaly.

Stack jsme nasazovali pomocí docker-compose. Měli jsme dobře napsaný docker-compose.yml, který nám umožnil téměř bez problémů zprovoznit stack. A zdálo se nám, že vítězství je již na dosah, teď už jen trochu přizpůsobíme svým potřebám a vše bude fungovat.

Bohužel, pokus o dodatečné nastavení systému pro příjem a zpracování logů z naší aplikace ihned neuspěl. Proto jsme se rozhodli, že se zaměříme na studium každé komponenty zvlášť, a poté se vrátíme k jejich provázanosti.

Takže jsme začali s Logstashem.

Prostředí, nasazení, spuštění Logstash v kontejneru

Pro nasazení používáme docker-compose, experimenty popsané zde byly prováděny na MacOS a Ubuntu 18.0.4.

Obraz Logstash, který jsme uvedli v našem původním docker-compose.yml, je docker.elastic.co/logstash/logstash:6.3.2

Ten použijeme pro naše experimenty.

Pro spuštění Logstash jsme napsali samostatný docker-compose.yml. Samozřejmě bylo možné spouštět obraz přímo z příkazové řádky, ale my jsme řešili konkrétní úlohu, kde se vše spouští z docker-compose.

Krátce o konfiguračních souborech

Jak vyplývá z popisu, Logstash lze spouštět jak pro jeden kanál, v takovém případě je třeba mu předat soubor *.conf, nebo pro více kanálů, kde je třeba mu poskytnout soubor pipelines.yml, který následně odkazuje na soubory .conf pro každý kanál.
Zvolili jsme druhou cestu. Ta nám přišla univerzálnější a škálovatelnější. Proto jsme vytvořili pipelines.yml a udělali adresář pipelines, do kterého budeme ukládat soubory .conf pro každý kanál.

Uvnitř kontejneru je ještě jeden konfigurační soubor — logstash.yml. Ten jsme neměnili, používáme ho tak, jak je.

Takže struktura našich adresářů:

Pro příjem dat zatím předpokládáme, že to bude tcp na portu 5046, a pro výstup použijeme stdout.

Toto je taková jednoduchá konfigurace pro první spuštění. Konečně, počáteční úkol je spustit.

Takže máme takový 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

Co zde vidíme?

1. Networks a volumes byly převzaty z původního docker-compose.yml (ten, kde se celý stack spouští) a myslím, že příliš neovlivňují celkový obraz.
2. Vytváříme jednu službu (services) logstash z obrazu docker.elastic.co/logstash/logstash:6.3.2 a pojmenujeme ji logstash_one_channel.
3. Předáváme dovnitř kontejneru port 5046 na stejný vnitřní port.
4. Zobrazujeme náš konfigurační soubor kanálů ./config/pipelines.yml jako soubor /usr/share/logstash/config/pipelines.yml uvnitř kontejneru, odkud jej logstash převezme, a děláme ho pouze pro čtení, jen pro jistotu.
5. Zobrazujeme adresář ./config/pipelines, kde máme soubory s nastavením kanálů, do adresáře /usr/share/logstash/config/pipelines a také ho děláme pouze pro čtení.

Soubor pipelines.yml

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

Tady je popsán jeden kanál s identifikátorem HABR a cesta k jeho konfiguračnímu souboru.

A nakonec soubor "./config/pipelines/habr_pipeline.conf"

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

Nebudeme se zatím zabývat jeho popisem, zkusíme spustit:

docker-compose up

Co vidíme?

Kontejner byl spuštěn. Můžeme zkontrolovat jeho funkčnost:

echo '13123123123123123123123213123213' | nc localhost 5046

A vidíme v konzoli kontejneru odpověď:

Ale přitom také vidíme:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Nelze získat informace o licenci ze serveru licencí {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch", …

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Pipeline byl úspěšně spuštěn {:pipeline_id=>".monitoring-logstash", :thread=>"#"}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Pipelines běží {: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 je nainstalován na Logstashi, ale ne na Elasticsearch. Nainstalujte X-Pack na Elasticsearch pro použití monitorovací funkce. Další funkce mohou být dostupné.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Úspěšně spuštěn koncový bod API Logstash {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Provádí se kontrola zdraví pro ověření fungování připojení k Elasticsearch {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Pokus o obnova připojení k neaktivní instanci ES selhal. {:url=«elasticsearch:9200/, :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch nedosažitelný: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Provádí se kontrola zdraví pro ověření fungování připojení k Elasticsearch {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Pokus o obnova připojení k neaktivní instanci ES selhal. {:url=«elasticsearch:9200/, :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch nedosažitelný: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

A naše logy stále stoupají.

Zde jsem zvýraznil zelenou barvou zprávu o úspěšném spuštění pipeline, červenou — zprávu o chybě a žlutou — zprávu o pokusu o spojení s elasticsearch:9200.
Toto se děje, protože v souboru logstash.conf, který je součástí obrazu, je ověřování dostupnosti elasticsearch. Logstash předpokládá, že funguje jako součást ELK stacku, který jsme oddělili.

Fungovat to může, ale není to pohodlné.

Řešením je vypnout tuto kontrolu pomocí proměnné prostředí XPACK_MONITORING_ENABLED.

Provedeme změnu v docker-compose.yml a znovu spustíme:

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

Teď už je vše v pořádku. Kontejner je připraven na experimenty.

Můžeme znovu zadat v sousední konzoli:

echo '13123123123123123123123213123213' | nc localhost 5046

A vidět:

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

Práce v rámci jednoho kanálu

Takže jsme se spustili. Teď se vlastně můžeme věnovat nastavení přímo logstash. Zatím se nebudeme dotýkat souboru pipelines.yml, podíváme se, co se dá získat prací s jedním kanálem.

Musím říct, že obecný princip práce se souborem konfigurace kanálu je dobře popsán v oficiálním průvodci, zde zde
Pokud si chcete číst v ruštině, použili jsme tenhle článek(ale syntaxe dotazů tam je stará, je potřeba to vzít v úvahu).

Pojďme postupovat sekvenčně od sekce Input. Práci s tcp jsme už viděli. Co dalšího by zde mohlo být zajímavého?

Testovací zprávy pomocí heartbeat

Existuje zajímavá funkce generování automatických testovacích zpráv.
K tomu je třeba do sekce input zahrnout plugin heartbeat.

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

Zapneme ho a začneme dostávat zprávy každou minutu.

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

Chceme dostávat zprávy častěji, musíme přidat parametr interval.
Takto budeme dostávat zprávy každých 10 sekund.

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

Získání dat ze souboru

Rozhodli jsme se také podívat na režim souboru. Pokud to sám funguje správně, možná ani nebude potřeba žádný agent, alespoň pro místní použití.

Podle popisu by měl být pracovní režim obdobný jako tail -f, tj. čte nové řádky nebo, jako možnost, čte celý soubor.

Takže, co chceme získat:

1. Chceme získávat řádky, které se přidávají do jednoho log souboru.
2. Chceme získávat data, která jsou zapisována do několika log souborů, přičemž chceme mít možnost rozlišit, odkud co pochází.
3. Chceme zkontrolovat, že při restartu logstash se tato data neobnoví.
4. Chceme zkontrolovat, že pokud logstash vypneme a data do souborů i nadále pokračují, tak po jeho spuštění tato data obdržíme.

Pro provedení experimentu přidáme další řádek do docker-compose.yml a otevřeme adresář, do kterého soubory ukládáme.

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

A změníme sekci input v habr_pipeline.conf

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

Spouštíme se:

docker-compose up

Pro vytváření a zapisování log souborů použijeme příkaz:

echo '1' >> logs/number1.log

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

Ano, funguje to!

Přitom vidíme, že se nám automaticky přidalo pole path. To znamená, že můžeme nově filtrovat záznamy podle něj.

Zkusíme ještě:

echo '2' >> logs/number1.log

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

A teď do jiného souboru:

 echo '1' >> logs/number2.log

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

Skvěle! Soubor byl načten, path byl uveden správně, všechno je v pořádku.

Zastavíme logstash a znovu spustíme. Počkáme. Ticho. To znamená, že tyto záznamy nebyly načteny znovu.

A teď nejodvážnější experiment.

Vložme logstash a provedeme:

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

Znovu spouštíme logstash a vidíme:

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

Hurá! Všechno bylo načteno.

Ale je třeba upozornit na následující. Pokud je kontejner s logstash odstraněn (docker stop logstash_one_channel && docker rm logstash_one_channel), nic nebude načteno. Uvnitř kontejneru byla uložena pozice souboru, do které byl přečten. Pokud se spouští „od nuly“, bude přijímat jen nové řádky.

Načítání již existujících souborů

Předpokládejme, že logstash spouštíme poprvé, ale již máme logy a chtěli bychom je zpracovat.
Pokud spustíme logstash s touto sekcí input, kterou jsme použili výše, nenahneme nic. Pouze nové řádky budou zpracovány logstash.

Aby se řádky z existujících souborů přetáhly, je třeba do sekce input přidat další řádku:

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

Je však třeba říci, že to platí pouze pro nové soubory, které logstash ještě neviděl. Pro soubory, které se již dostaly do zorného pole logstash, si již zapamatoval jejich velikost a nyní bude brát pouze nové záznamy v nich.

Na této příležitosti se zastavíme při studiu sekce input. Je tam ještě mnoho možností, ale pro naše další experimenty nám zatím postačí.

Směrování a převod dat

Zkusíme vyřešit následující úkol, řekněme, že máme zprávy z jednoho kanálu, část z nich je informativní, a část zprávy o chybách. Rozlišují se podle tagu. Některé INFO, jiné ERROR.

Musíme je na výstupu rozdělit. Tedy. Informativní zprávy píšeme do jednoho kanálu, a zprávy o chybách do druhého.

K tomu přecházíme z sekce input na filter a output.

Pomocí sekce filter rozložíme příchozí zprávu, získáme z ní hash (páry klíč-hodnota), se kterým už můžeme pracovat, tedy analyzovat podle podmínek. A v sekci output vybereme zprávy a každou pošleme do svého kanálu.

Analýza zprávy pomocí grok

Abychom mohli analyzovat textové řetězce a získat z nich sadu polí, existuje v sekci filter speciální plugin — grok.

Nepředstavujeme si, že bychom zde dali jeho podrobný popis (na to odkazuji na oficiální dokumentaci), uvedu svůj jednoduchý příklad.

Abychom to mohli udělat, je nutné si ujasnit formát vstupních řetězců. Mám je takové:

1 INFO message1
2 ERROR message2

Tedy. Identifikátor na prvním místě, poté INFO/ERROR, poté nějaké slovo bez mezer.
Není to složité, ale pro pochopení principu fungování je to dost.

Takže, v sekci filter, v pluginu grok musíme definovat vzor pro analýzu našich řetězců.

Bude vypadat takto:

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

V podstatě se jedná o regulární výraz. Používají se již hotové vzory, jako je INT, LOGLEVEL, WORD. Jejich popis, stejně jako další vzory, lze vidět zde zde

Nyní, když projde tímto filtrem, náš řetězec se promění v hash ze tří polí: message_id, message_type, message_text.

Tyto budou vycházet v sekci output.

Směrování zpráv v sekci output pomocí příkazu if

Ve sekci output, jak jsme si vzpomněli, jsme chtěli rozdělit zprávy do dvou toků. Jedny — které jsou iNFO, budeme posílat na konzoli, a chyby budeme zapisovat do souboru.

Jak tedy tyto zprávy rozdělit? Podmínka úkolu nám už naznačuje řešení — máme přece již vyčleněné pole message_type, které může nabývat pouze dvou hodnot INFO a ERROR. Přesně podle něj provedeme výběr pomocí operátoru if.

if [message_type] == "ERROR" {
        # Zde zapisujeme do souboru
       } else
     {
      # Zde posíláme na stdout
    }

Popis práce s poli a operátory lze vidět v této sekci oficiálního manuálu.

Nyní k samotnému výstupu.

Výstup na konzoli, tady je to jasné — stdout {}

A co se týče výstupu do souboru — připomínáme, že to vše spouštíme z kontejneru a aby byl soubor, do kterého zapisujeme výsledek, přístupný zvenčí, musíme otevřít tento adresář v docker-compose.yml.

Celkem:

Sekce output našeho souboru vypadá takto:

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

Do docker-compose.yml přidáváme další svazek pro výstup:

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

Spouštíme, zkoušíme, vidíme rozdělení na dva toky.

Zdroj: habr.com