въведение

При внедряването на друга система се сблъскахме с необходимостта да обработваме голям брой различни регистрационни файлове. За инструмент беше избран 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. Ние не го пипаме, ние го използваме както е.

Така че нашата структура на директория е:

За момента приемаме, че това е 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. Мрежите и томовете са взети от оригиналния docker-compose.yml (този, където се стартира целият стек) и мисля, че не влияят много на общата картина тук.
2. Създаваме една услуга (услуги) 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 ще го вземе и ще го направи само за четене, за всеки случай.
5. Съпоставяме директорията ./config/pipelines, където имаме файловете за конфигуриране на канали, към директорията /usr/share/logstash/config/pipelines и я правим само за четене.

piping.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

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

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

logstash_one_channel | [2019-04-29T11:28:59,790][ГРЕШКА][logstash.licensechecker.licensereader] Не може да се извлече информация за лиценза от сървъра за лицензи {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore ::ResolutionFailure]elasticsearch", ...

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Конвейерът стартира успешно {:pipeline_id=>".monitoring-logstash", :thread=>"# »}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Тръбопроводи, работещи {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[ ]}
logstash_one_channel | [2019-04-29T11:29:00,015][ГРЕШКА][logstash.inputs.metrics] X-Pack е инсталиран на Logstash, но не и на Elasticsearch. Моля, инсталирайте X-Pack на Elasticsearch, за да използвате функцията за наблюдение. Може да са налични и други функции.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent] Успешно стартирана крайна точка на Logstash API {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][ИНФОРМАЦИЯ ][logstash.outputs.elasticsearch] Изпълнява се проверка на състоянието, за да се види дали връзка с Elasticsearch работи {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,487][ПРЕДУПРЕЖДЕНИЕ ][logstash.outputs.elasticsearch] Опитах се да възкреся връзката към мъртво копие на ES, но получих грешка. {:url=>"еластично търсене:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch"}
logstash_one_channel | [2019-04-29T11:29:04,704][ИНФОРМАЦИЯ ][logstash.licensechecker.licensereader] Изпълнява се проверка на състоянието, за да се види дали връзка с Elasticsearch работи {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,710][ПРЕДУПРЕЖДЕНИЕ ][logstash.licensechecker.licensereader] Опитах се да възкреся връзката към мъртво копие на ES, но получих грешка. {:url=>"еластично търсене:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch"}

И нашият дънер пълзи нагоре през цялото време.

Тук маркирах в зелено съобщението, че конвейерът е стартирал успешно, в червено съобщението за грешка и в жълто съобщението за опит за свързване еластично търсене: 9200.
Това се случва поради факта, че в logstash.conf, включен в изображението, има проверка за наличието на elasticsearch. В края на краищата 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, нека видим какво можем да получим, като работим с един канал.

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

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

Тествайте съобщения с помощта на сърдечен ритъм

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

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

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

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

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

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

Според описанието режимът на работа трябва да е подобен на 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

И променете секцията за въвеждане в 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 | }

Да, работи!

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

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

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

Страхотен! Файлът е взет, пътят е зададен правилно, всичко е наред.

Спрете 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 със секцията за въвеждане, която използвахме по-горе, няма да получим нищо. Само нови редове ще бъдат обработени от logstash.

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

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

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

Нека спрем на това, като изучаваме секцията за въвеждане. Има много повече опции, но засега имаме достатъчно за допълнителни експерименти.

Маршрутизиране и трансформиране на данни

Нека се опитаме да разрешим следния проблем, да кажем, че имаме съобщения от един канал, някои от тях са информационни, а други са съобщения за грешка. Те се различават по етикет. Някои са ИНФО, други са ГРЕШКА.

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

За да направите това, преминете от секцията за въвеждане към филтър и изход.

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

Разбор на съобщение с grok

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

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

За да направите това, трябва да вземете решение за формата на входните редове. Имам ги така:

1 ИНФО съобщение1
2 Съобщение за ГРЕШКА2

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

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

Ще изглежда така:

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

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

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

Те ще бъдат показани в изходния раздел.

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

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

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

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

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

Сега относно самото заключение.

Конзолен изход, тук всичко е ясно - stdout {}

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

Общо:

Изходният раздел на нашия файл изглежда така:


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

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

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