Вовед

При распоредувањето на друг систем, се соочивме со потребата да обработиме голем број различни логови. ЕЛК беше избран како алатка. Оваа статија ќе го разгледа нашето искуство во поставувањето на овој оџак.

Не си поставуваме цел да ги опишеме сите негови способности, туку сакаме да се концентрираме конкретно на решавање на практични проблеми. Ова се должи на фактот дека иако има прилично голема количина документација и готови слики, има доста стапици, барем ние ги најдовме.

Го распоредивме стекот преку 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

Ќе го користиме за експерименти.

Напишавме посебен docker-compose.yml за да се изврши logstash. Се разбира, беше можно да се стартува сликата од командната линија, но ние решававме специфичен проблем, каде што работиме сè од 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 и исто така го правиме само за читање.

Датотека 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

И го гледаме одговорот во конзолата на контејнерот:

Но, во исто време, гледаме и:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Не може да се вратат информациите за лиценцата од серверот за лиценца {:message=>„Elasticsearch недостапно: [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][ERROR][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][INFO ][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::Излези::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch недостижно: [http://elasticsearch:9200/:][Resoureail:] elasticearch"}
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][ПРЕДУПРЕДУВАЊЕ ][logstash.licensechecker.licensereader] Се обиде да ја оживее врската со мртвиот примерок ES, но доби грешка. {:url=>“еластичен пребарувач:9200/", :error_type=>LogStash::Излези::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch недостижно: [http://elasticsearch:9200/:][Resoureail:] elasticearch"}

И нашиот дневник постојано се прикрадува.

Овде ја истакнав со зелено пораката дека гасоводот е успешно стартуван, со црвено пораката за грешка и со жолта пораката за обид за контакт еластичен пребарувач: 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
   }
  }

Враќање податоци од датотека

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

Според описот, режимот на работа треба да биде сличен на опашката -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, тој веќе ја запомни нивната големина и сега ќе зема само нови записи во нив.

Ајде да застанеме овде и да го проучиме делот за внесување. Има уште многу опции, но тоа ни е доволно за понатамошни експерименти засега.

Рутирање и трансформација на податоци

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

Треба да ги одвоиме на излезот. Оние. На еден канал пишуваме информативни пораки, а на друг пораки за грешки.

За да го направите ова, преместете се од делот за внесување до филтер и излез.

Користејќи го делот за филтри, ќе ја анализираме дојдовната порака, добивајќи хаш (парови клуч-вредност) од неа, со кои веќе можеме да работиме, т.е. расклопувајте според условите. И во делот за излез, ќе избереме пораки и ќе ја испратиме секоја на свој канал.

Парсирање на порака со грок

За да ги анализирате текстуалните низи и да добиете збир на полиња од нив, во делот за филтри има посебен додаток - грок.

Без да си поставувам цел да дадам детален опис за тоа овде (за ова се повикувам официјална документација), ќе го дадам мојот едноставен пример.

За да го направите ова, треба да одлучите за форматот на влезните низи. Ги имам вака:

1 ИНФО порака1
2 Порака за ГРЕШКА2

Оние. Прво доаѓа идентификаторот, потоа INFO/ERROR, па некој збор без празни места.
Не е тешко, но доволно е да се разбере принципот на работа.

Значи, во делот за филтри на приклучокот grok, мора да дефинираме шема за парсирање на нашите жици.

Ќе изгледа вака:

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

Во суштина тоа е редовен израз. Се користат готови обрасци, како што се INT, LOGLEVEL, WORD. Нивниот опис, како и други обрасци, може да се најдат овде тука

Сега, поминувајќи низ овој филтер, нашата низа ќе се претвори во хаш од три полиња: порака_ид, порака_тип, порака_текст.

Тие ќе бидат прикажани во делот за излез.

Рутирање на пораки до делот за излез користејќи ја командата if

Во делот за излез, како што се сеќаваме, требаше да ги поделиме пораките на два текови. Некои - кои се iNFO, ќе бидат излезни на конзолата, а со грешки, ќе излеземе во датотека.

Како да ги разделиме овие пораки? Состојбата на проблемот веќе сугерира решение - на крајот на краиштата, ние веќе имаме посветено поле за порака_тип, кое може да земе само две вредности: 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