ProHoster > Блог > Администриране > Практическо приложение на ELK. Настройваме logstash

Практическо приложение на ELK. Настройваме logstash

Въведение

При развертывании новой системы мы столкнулись с необходимостью обрабатывать большие объемы различных логов. В качестве инструмента мы выбрали ELK. В этой статье мы расскажем о нашем опыте настройки этого стека.

Мы не ставим целью описать все его возможности, но хотим сосредоточиться конкретно на решении практических задач. Это связано с тем, что, несмотря на наличие обширной документации и готовых образов, существуют множество подводных камней, которые, по крайней мере у нас, обнаружились.

Мы развертывали стек через docker-compose. Более того, у нас был хорошо написанный docker-compose.yml, который позволил практически без проблем поднять стек. Нам казалось, что победа уже близка, нужно было лишь немного адаптировать его под наши нужды.

К сожалению, попытка донастроить систему для получения и обработки логов от нашего приложения не увенчалась успехом. Поэтому мы решили изучить каждый компонент отдельно, а потом уже вернуться к их взаимодействиям.

Итак, мы начали с logstash.

Окружение, развертывание, запуск Logstash в контейнере

Для развертывания мы используем docker-compose, описанные здесь эксперименты проводились на MacOS и Ubuntu 18.0.4.

Образ logstash, который мы прописали в исходном docker-compose.yml, это docker.elastic.co/logstash/logstash:6.3.2

Его мы будем использовать для экспериментов.

Для запуска logstash мы написали отдельный docker-compose.yml. Конечно, можно было запустить образ из командной строки, но мы решали конкретную задачу, в которой всё запускается из docker-compose.

Кратко о конфигурационных файлах

Как следует из описания, logstash можно запускать как для одного канала, в этом случае ему нужно передать файл *.conf, так и для нескольких каналов, тогда ему необходимо передать файл pipelines.yml, который, в свою очередь, будет ссылаться на файлы .conf для каждого из каналов.
Мы выбрали второй путь. Он показался нам более универсальным и масштабируемым. Поэтому мы создали pipelines.yml и директорию pipelines, в которую будем помещать файлы .conf для каждого канала.

Внутри контейнера есть еще один конфигурационный файл — logstash.yml. Мы его не трогаем, используем как есть.

Итак, структура наших каталогов:

Практическо приложение на ELK. Настройваме logstash

Для получения входных данных пока считаем, что это tcp по порту 5046, а для вывода будем использовать stdout.

Ето такава проста конфигурация за първоначално стартиране. Понеже основната задача е да се стартира.

И така, имаме следния 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

Какво виждаме тук?

  1. Networks и volumes бяха взети от изходния docker-compose.yml (този, от който стартира целият стек) и мисля, че не влияят значително на общата картина.
  2. Създаваме един сервис (services) logstash, от образа docker.elastic.co/logstash/logstash:6.3.2 и му присвояваме име logstash_one_channel.
  3. Пробиваме порта 5046 в контейнера, към същия вътрешен порт.
  4. Отображаваме нашия файл с конфигурацията на каналите ./config/pipelines.yml на файла /usr/share/logstash/config/pipelines.yml вътре в контейнера, откъдето logstash ще го подхване и го правим read-only, просто за всеки случай.
  5. Отображаваме директорията ./config/pipelines, където имаме файлове с настройки на каналите, в директорията /usr/share/logstash/config/pipelines и също я правим read-only.

Практическо приложение на ELK. Настройваме logstash

Файл pipelines.yml

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

Тук е описан един канал с идентификатор HABR и пътя към неговия конфигурационен файл.

И накрая файлът "./config/pipelines/habr_pipeline.conf"

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

Не ще разглеждаме описанието му за сега, пробваме да го стартираме:

docker-compose up

Какво виждаме?

Контейнерът стартира. Можем да проверим неговата работа:

echo '13123123123123123123123213123213' | nc localhost 5046

И виждаме отговора в консолата на контейнера:

Практическо приложение на ELK. Настройваме logstash

Но също така виждаме:

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] Извършва се проверка на здравето, за да се види дали връзката с Elasticsearch работи {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Опит за възстановяване на връзка с мъртва ES инстанция, но получихме грешка. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch не е достижим: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Извършва се проверка на здравето, за да се види дали връзката с Elasticsearch работи {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Опит за възстановяване на връзка с мъртва ES инстанция, но получихме грешка. {:url=«elasticsearch:9200/», :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch не е достижим: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

И нашият лог постоянно се увеличава.

Тук подчертах със зелен цвят съобщението, че pipeline е стартиран успешно, с червен — съобщението за грешка и с жълт — съобщението за опит за свързване с elasticsearch:9200.
Това се дължи на проверката за достъпност на elasticsearch в logstash.conf, включен в образа. В края на краищата, logstash предполага, че работи в рамките на Elk стека, а ние сме го отделили.

Може да работите, но не е удобно.

Решението е да деактивирате тази проверка чрез променлива на средата XPACK_MONITORING_ENABLED.

Нека направим промяна в 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
    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

Сега всичко е наред. Контейнерът е готов за експерименти.

Можем отново да напишем в съседния терминал:

echo '13123123123123123123123213123213' | nc localhost 5046

И да видим:

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

Работа в рамките на един канал

И така, стартирахме. Сега можем да се съсредоточим върху конфигурирането на logstash. Засега няма да пипаме файла pipelines.yml, ще видим какво можем да получим, работейки с един канал.

Трябва да кажа, че общият принцип на работа с конфигурационния файл на канала е добре описан в официалното ръководство, ето го тук.
Ако искате да прочетете на български, използвахме тази статия(но синтаксисът на заявките там е остарял, трябва да се има предвид).

Нека да започнем последователно от секцията Input. Работата по tcp вече видяхме. Какво още може да бъде интересно тук?

Тестови съобщения, използвайки heartbeat

Има интересна възможност да генерираме автоматични тестови съобщения.
За това в секцията input трябва да включим плъгина heartbeat.

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

Включваме и започваме да получаваме веднъж в минута

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

Искаме да получаваме по-често, трябва да добавим параметъра interval.
Така ще получаваме съобщение на всеки 10 секунди.

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

Получаване на данни от файл

Също така решихме да разгледаме режима file. Ако работи нормално с файла, то може би няма да е необходим агент, поне за локално използване.

Според описанието, режимът на работа трябва да бъде аналогичен на tail -f, т.е. чете новите редове или, като опция, чете целия файл.

И така, какво искаме да получим:

  1. Искаме да получаваме редовете, които се добавят в един лог файл.
  2. Искаме да получаваме данни, които се записват в няколко лог файла, като същевременно имаме възможност да разделим откъде какво е получено.
  3. Искаме да проверим, че при рестартиране на logstash той няма да получи тези данни отново.
  4. Искаме да проверим, че ако logstash бъде изключен, а данните в файловете продължат да се записват, то когато го стартираме отново, ще получим тези данни.

За да проведем експеримента, ще добавим още един ред в 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
    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

И ще променим секцията input в habr_pipeline.conf

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

Стартираме:

docker-compose up

За създаване и записване на лог файлове ще използваме командата:

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

Да, работает!

При това, ние виждаме, че автоматично е добавено поле path. Това означава, че по-нататък можем да филтрираме записите по него.

Нека опитаме отново:

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

А сега в друг файл:

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

Супер! Файл бе подхванат, path е посочен правилно, всичко е наред.

Спираме logstash и го пускаме отново. Изчакваме. Тишина. Тоест, не получаваме отново тези записи.

А сега най-смелият експеримент.

Лягаме logstash и изпълняваме:

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

Отново стартираме logstash и виждаме:

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

Ура! Всичко беше подхванато.

Но трябва да предупредим за следното. Ако контейнерът с logstash бъде изтрит (docker stop logstash_one_channel && docker rm logstash_one_channel), нищо няма да се подхване. Вътре в контейнера е запазена позицията на файла, до която е бил прочетен. Ако се пуска "от нулата", той ще приема само новите редове.

Четене на вече съществуващи файлове

Да предположим, че стартираме logstash за първи път, но вече имаме логове и бихме искали да ги обработим.
Ако стартираме logstash с раздела input, който използвахме по-горе, няма да получим нищо. Само новите редове ще бъдат обработвани от logstash.

За да могат да бъдат включени редове от съществуващите файлове, трябва да добавим допълнителен ред в секция input:

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

Има обаче нюанс, това действа само за нови файлове, които logstash все още не е виждал. За файловете, които вече са попадали в обхвата на logstash, той вече е запомнил техния размер и сега ще взема само нови записи в тях.

Ще спрем с изучаването на секция input. Има още много опции там, но за нашите по-нататъшни експерименти в момента ни е достатъчно.

Маршрутизиране и преобразуване на данни

Нека опитаме да решим следната задача, да предположим, че получаваме съобщения от един канал, част от тях са информационни, а част са съобщения за грешки. Те се различават по таг.

Трябва да ги разделим на изхода. Т.е. Информационните съобщения изпращаме в един канал, а съобщенията за грешки в друг.

За целта, от секция input преминаваме към filter и output.

С помощта на секция filter ще анализираме входящото съобщение, получавайки от него hash (пара ключ-стойност), с който можем да работим, т.е. да го анализираме по условия. А в секция output, ще отберем съобщенията и ще изпратим всяко в своя канал.

Анализ на съобщението с помощта на grok

За да анализираме текстовите редове и да получим набор от полета, в секция filter има специален плъгин — grok.

Не цели да предоставя подробно описание тук (за това се отнасям към официалната документация), ще дам прост пример.

За целта, трябва да определим формата на входните редове. При мен те са такива:

1 INFO message1
2 ERROR message2

Т.е. Идентификатор на първо място, след това INFO/ERROR, след това някаква дума без интервали.
Не е сложно, но за разбирането на принципа на работа е достатъчно.

И така, в секция filter, в плъгина grok трябва да определим шаблон за анализа на нашите редове.

Той ще изглежда така:

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

Всъщност, това е регулярното изражение. Използват се вече готови шаблони, като INT, LOGLEVEL, WORD. Описание на тях, както и други шаблони, можете да видите тук тук.

Сега, преминавайки през този филтър, нашият ред ще се преобразува в hash от три полета: message_id, message_type, message_text.

Точно те ще бъдат изведени в секция output.

Маршрутиране на съобщения в секция output с помощта на командата if

В секцията output, както си спомняме, планирахме да разделим съобщенията на два потока. Едните — тези с тип iNFO, ще изведем на конзолата, а тези с грешки ще изведем в файл.

Как да разделим тези съобщения? Условията на задачата вече подсказват решението — имаме вече определено поле message_type, което може да приема само две стойности INFO и ERROR. Именно по него ще направим избора с помощта на оператора if.

if [message_type] == "ERROR" {
        # Тук извеждаме в файл
       } else
     {
      # Тук извеждаме в stdout
    }

Описание на работата с полета и оператори можете да видите в тази секция официалния наръчник.

Сега, за самия изход.

Изход на конзолата, тук всичко е ясно — stdout {}

А изходът в файл — нека припомним, че всичко това стартираме от контейнер и за да бъде достъпен файлът, в който записваме резултата, трябва да отворим тази директория в docker-compose.yml.

В обобщение:

Секцията output на нашия файл изглежда така:

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

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

Стартираме, пробваме, виждаме разделянето на два потока.

Източник: habr.com

Купете надежден хостинг за сайтове със защита от DDoS, VPS и VDS сървъри 🔥 Купете надежден хостинг за сайтове със защита от DDoS, VPS и VDS сървъри | ProHoster