В этом материале я расскажу о том, как в Яндекс.Деньгах организована система сбора и доставки серверных логов платежных сервисов. Доставкой логов я занимаюсь весь последний год, и за это время накопилось достаточно нюансов, чтобы поделиться опытом с общественностью.
Система построена на базе стека EHK (Elasticsearch/Heka/Kibana), с прицелом на работу практически в реальном времени. Особый упор сделаю на тонкие места и нюансы обработки миллиардов строк текста в сутки.
Я верю в три вещи: мониторинг, логи и бэкапы.
Для быстрого доступа к оперативным логам в Яндекс.Деньгах используется ElasticSearch, получающий данные из почти сотни микросервисов, которые и составляют платежный сервис.
Под «оперативными» я подразумеваю логи, доступные в реальном времени: задержка доставки новых записей от файла лога до кластера Elasticsearch сейчас составляет менее секунды. Служба мониторинга всегда точно знает состояние сервиса в любой момент, а разработчики могут оперативно оценивать и корректировать поведение новой версии каждого микросервиса сразу после релиза.
Но так было не всегда. Когда пять месяцев назад я пришел в компанию, мне поставили задачу наладить работу по доставке оперативных логов сервиса до кластера ElasticSearch (далее просто ES). На тот момент использовались четыре схемы их разбора и доставки до ES:
Heka → TCP output → Heka → ES.
Heka → Apache Kafka → Apache Flume → ES.
SyslogNg → ES.
- nxlog → nxlog → ES.
Почти все они работали неидеально: кластер Kafka был ненадежен, Flume периодически вис, отчего ES впадал в ступор.
Логи – это на самом деле очень простая вещь: множество файлов, которые на боевых серверах быстро растут в объемах. Поэтому простые решения в этом случае — самые надежные.
Так как «чем больше в зонтике сочленений — тем выше вероятность его поломки», то я выбросил все неработающие и неоправданно сложные схемы и остановился на одной, в которой роль обработчика и транспорта выполняла бы Heka.
Heka забирает логи и по протоколу TCP отправляет их в другую Heka для дальнейшей пересылки в ElasticSearch Cluster.
Идея следующая: на каждом сервере установлена Heka, логи всех сервисов этого сервера упаковываются в бинарный протокол Protobuf и отправляются на приемную Heka для разбора, фильтрации и отправки в ES.
Почему не классический стек ELK и почему deprecated Heka, а не Logstash
С точки зрения конфигурации и состава ПО наш кластер ES выглядит следующим образом:
ElasticSearch 2.2.
Kibana 4.4.2.
Две мастер-ноды: Intel Xeon 2x E5-2660, 64 ГБ ОЗУ, 2x 146 ГБ RAID-10.
Клиентская нода с установленной Kibana: Intel Xeon 2xE5-2660, 64 ГБ ОЗУ, 2x146 ГБ RAID-10.
- Четыре дата-ноды: Intel Xeon 2x E5-2640 v3; 512 ГБ ОЗУ, 3x16 ТБ RAID-10.
Все ноды ES находятся в одной локальной сети и подключены к одному маршрутизатору, что позволяет осуществлять общение внутри кластера по транспортному протоколу на максимальной скорости. Такая организация заметно ускоряет размещение индексов и rebalance кластера.
В современном мире стек Elasticsearch/Logstash/Kibana стал уже практически стандартом де-факто для работы с логами. И если против Elasticsearch и Kibana нет никаких возражений, то с Logstash есть один нюанс — он создан на jRuby (написанный на Java интерпретатор языка Ruby) и требует наличия в системе JVM. Учитывая, что Яндекс.Деньги – это множество микросервисов, размещенных на сотнях физических серверов и виртуальных контейнеров, ставить в каждый из них тяжеловесные Logstash и JAVA было бы неправильно. На Heka выбор пал из-за ее простоты, надежности, легкости, умения фильтровать проходящие сообщения и отличной буферизации данных.
Что касается статуса продукта (deprecated) – для нас это не аргумент. Лук и стрелы для военных целей тоже deprecated, но это никак не мешает вам выстрелить кому-нибудь в голову с гарантированным результатом. Если же понадобится, например, нестандартный плагин или обработчик, то доработать продукт поможет целый штат опытных разработчиков.
Но это всё была теория, а при переходе к практике начались проблемы.
Дьявол прятался в объеме данных
Учитывая финансовую специфику нашей работы, почти все сервисы пишут много разной информации в логи. Для примера и понимания масштаба: объем логов некоторых из компонентов системы доходит до 250 тысяч строк в минуту.
Ни одна Heka, на каком бы мощном железе она ни стояла, в одиночку такой объём не обработает — неизбежны проседания производительности и потеря данных. Ни то, ни другое, конечно же, совершенно недопустимо, поэтому на помощь приходит HAProxy. Итоговая схема получилась следующей:
На схеме изображено общее направление трафика логов от Heka-источников до связки HAProxy + Heka.
На каждом сервере стоит одна Heka, собирающая логи микросервисов этого сервера. Данные собираются, пакуются в Protobuf и по TCP уходят на балансировщик нагрузки, обслуживающий дата-центр. Бэкендами работают HAProxy, расположенные непосредственно на дата-нодах ES, за которыми стоят пулы Heka. В свою очередь, они принимают данные, переупаковывают их в ESJson и отправляют в локальную дата-ноду по протоколу HTTP.
… и в разных форматах лог-файлов
Несмотря на то, что основным языком в компании является Java и логи выводятся через стандартную библиотеку log4j, не было единого принятого формата на момент постройки «кластера мечты». Каждый микросервис писал логи собственного типа, включая вариации в наборе выводимых полей и форматах даты-времени. Не хотелось ждать, пока разработка приведёт логи к общим форматам, поэтому движение к цели было параллельным. Одновременно с разбором существующих форматов логов заводились задачи на доработку, а по мере релиза новых версий с правильными форматами менялись и настройки сборщиков.
Перейдем к самому соку — нюансам настройки
Поведение Heka описывается .toml файлами, которые при запуске собираются в единую конфигурацию, и в зависимости от описанных блоков Heka строит модель обработки данных.
Каждый блок проходит через несколько этапов:
Input — входной поток (это может быть файл, TCP/UDP input, данные, прочитанные из Kafka, событий Docker контейнеров и т.п.).
Splitter — тут мы указываем начало и конец каждого блока данных в потоке. Обязательный этап для многострочных данных вроде Java Stacktrace.
Decoder — описывает правила декодирования входящих данных. Мы в основном используем Regex декодеры.
Filters — этап фильтрации и изменения данных.
Encoders — кодирование потока данных под формат получателя.
- Output — здесь мы описываем, как и куда данные должны отправиться.
Все эти этапы — просто плагины на Go и Lua. При необходимости можно написать что-нибудь своё. Например, плагин-фильтр на Lua, который будет отсекать отправку в ES записей о мониторинговых запросах к сервису; или вырезать из логов конфиденциальные данные.
Хека в древнеегипетской мифологии — бог магии. И то, что Heka позволяет делать с логами — просто волшебно.
Параметры сервера-источника логов
Разберем конфигурацию Heka на примере сервера-источника логов и файла service.toml.
[money-service-log]
type = "LogstreamerInput"
log_directory = "/var/log/tomcat"
file_match = "money-service.log"
splitter = "RegexSplitter"
decoder = "service_decoder"
Простейший случай — один файл, ротация происходит системными средствами. Если файлов много и они различаются по форматам, то следует описывать каждый парой input/decoder. За более детальным описанием лучше обратиться к официальной документации. Если что-то остается непонятным — обязательно спрашивайте в комментариях.
[RegexSplitter]
delimiter = '\n(\[\d\d\d\d-\d\d-\d\d)'
delimiter_eol = false
Так как логи могут быть многострочными (те же stacktraces), не забываем про RegexSplitter, который даёт Heka понять, где заканчивается один блок текста и начинается другой.
[service_decoder]
type = "PayloadRegexDecoder"
match_regex = '^\[(?P<timestamp>\d{4}-\d{2}-\d{2}T[\d:.\+]+\])\s+(?P<level>[A-Z]+)\s+\[(?P<thread>.*)\]\s+\[(?P<context>\S*)\]\s+\[(?P<traceid>\S*)\]\s+\[(?P<unilabel>\S*)\]\s\[(?P<class>\S+)\]\s-\s(?P<msg>.*)'
log_errors = true
[service_decoder.message_fields]
@timestamp = "%timestamp%"
level = "%level%"
thread = "%thread%"
context = "%context%"
traceid = "%traceid%"
unilabel = "%unilabel%"
class = "%class%"
msg = "%msg%"
В match_regex описываем строки лога регулярным выражением в стандарте языка Go. Регулярные выражения в Go практически совпадают со стандартным PCRE, но есть ряд нюансов, из-за которых Heka может отказаться стартовать. Например, некоторые реализации PCRE простят такой синтаксис:
(?<groupname>.*)
А вот GOLANG — не простит.
С помощью параметра log_errors собираем все ошибки в отдельный лог — они понадобятся позже.
[ServiceOutput]
type = "TcpOutput"
address = "loadbalancer.server.address:port"
keep_alive = true
message_matcher = "Logger == 'money-appname'"
use_buffering = true
[ServiceOutput.buffering]
max_file_size = 100857600
max_buffer_size = 1073741824
full_action = "block"
Буферизация выходящего потока — одна из отличных возможностей Heka. По умолчанию она хранит выходной буфер по следующему пути:
/var/cache/hekad/output_queue/OutputName
В настройках мы ограничиваем размер каждого файла буфера объемом 100 МБ, а также устанавливаем суммарный размер кеша для каждого Output-модуля в 1 ГБ. Параметр full_action может принимать три значения:
shutdown — при переполнении буфера Heka останавливается;
drop — при переполнении буфера он начинает работать как стек, удаляя старые сообщения в очереди;
- block — при заполнении буфера Heka приостанавливает все операции и ждёт, пока не появится возможность отправить данные.
С block вы гарантированно не потеряете ни одной строки лога. Единственный минус в том, что при обрыве соединения или деградации канала вы получите резкий скачок трафика при возобновлении связи. Это связано с тем, что Heka отправляет накопленный буфер, пытаясь вернуться к обработке в реальном времени. Приёмный пул нужно проектировать с запасом, обязательно учитывая возможность таких ситуаций, иначе можно с лёгкостью провернуть DDoS самих себя.
Кстати, по поводу использования двойных и одинарных кавычек в конфигурации Heka — подразумевается следующее:
Значения переменных в одинарных кавычках по умолчанию рассматриваются как raw string. Спецсимволы экранировать не нужно.
- Значения переменных в двойных кавычках рассматриваются как обычный текст и требуют двойного экранирования при использовании в них регулярных выражений.
Этот нюанс в свое время попортил мне немало крови.
Конфигурация бэкендa
Бэкенд для балансировщика — это связка из HAProxy и трёх экземпляров Heka на каждой дата-ноде ES.
В HAProxy всё довольно просто и пояснений, как мне кажется, не требует:
listen pool_502
bind 0.0.0.0:502
balance roundrobin
default-server fall 5 inter 5000 weight 10
server heka_1502 127.0.0.1:1502 check
server heka_2502 127.0.0.1:2502 check
server heka_3502 127.0.0.1:3502 check
На каждом сервере запущено три инстанса Heka, отличающихся только портами:
[Input_502_1]
type = "TcpInput"
address = "0.0.0.0:1502"
keep_alive = true
keep_alive_period = 180
decoder = "MultiDecoder_502"
[MultiDecoder_502]
type = "MultiDecoder"
subs = ['Service1Decoder', 'Service2Decoder']
cascade_strategy = "first-wins"
log_sub_errors = true
В конфигурации используется MultiDecoder, так как через один порт проходят логи многих сервисов. Политика first-wins означает, что после первого совпадения дальнейший перебор декодеров прекращается.
[Service1Decoder]
type = "ProtobufDecoder"
[Service2Decoder]
type = "ProtobufDecoder"
[Service1Encoder]
type = "ESJsonEncoder"
index = "service1-logs-%{%Y.%m.%d}"
es_index_from_timestamp = false
type_name = "%{Logger}"
fields = [ "DynamicFields", "Payload", "Hostname" ]
dynamic_fields = ["@timestamp", "level", "thread", "context", "traceid", "unilabel", "class", "msg"]
[Service1Encoder.field_mappings]
Payload = "message"
Hostname = "MessageSourceAddress"
Параметр es_index_from_timestamp нужен для указания, что дата и время для формирования имени индекса берутся не из приходящих данных, а из локального времени сервера. Позволяет избежать бардака, когда серверы работают в разных временных зонах и кто-то пишет в логи время в UTC, а кто-то в MSK.
В параметре index реализуется принцип «один сервис – один индекс», новый индекс создается каждые сутки.
[Service1Output]
type = "ElasticSearchOutput"
message_matcher = "Logger == 'money-service1'"
server = "http://localhost:9200"
encoder = "Service1Encoder"
use_buffering = true
Плагины Output разбирают поток данных на основании параметра message_matcher, соответствующего имени файла лога. Чтобы не перегружать сеть, Heka отправляет данные в локальную дата-ноду, на которой установлена. А уже дальше ES сам рассылает индексированные данные по транспортному протоколу между дата-нодами кластера.
Заключение
Описанная выше схема успешно работает и индексирует по 25-30 тысяч записей в секунду. Запас прочности приемных пулов Heka позволяет выдерживать пики нагрузки до 100 тыс. записей/сек:
Статистика из Zabbix.
В ElasticSearch мы храним логи за последний 21 день. Опыт показывает, что к более старым данным оперативный доступ нужен крайне редко. Но если понадобится, их всегда можно запросить с лог-серверов, хранящих данные практически вечно.
Текущее состояние кластера по данным Kopf.
Я описал только ту часть системы, которая относится к сбору и доставке логов, поэтому в следующей статье собираюсь рассказать про сам кластер ElasticSearch и его настройку. Думаю рассказать, как мы его виртуализировали, как переезжали с версии 2.2 на 5.3 и перевозили с собой 24 миллиарда записей, не потеряв при этом веру в человечество.
Может быть, добавить к рассказу что-то еще, какие-то упущенные из виду детали? Поделитесь мнением в комментариях.