Логи, собственно, уже собираем. Следующая задача их нужно каким-то образом обработать логи: подготовить и отправить в базу данных.

Чем обрабатывать?

В принципе вариантов на эту тему масса. Первый вариант — использовать готовое решение, второй вариант — писать свое. Собственно сложность состоит в том, что хоть и логи +/- похожи друг на друга, но все же имеют отличия. Возьмем первый вариант и попробуем его реализовать.

Для получения и парсинга логов, дума, можно взять fluentd. Продукт открытый, документация объемная, готовые плагины тоже есть. Попробуем разобраться.

Парсим логи

Для начала его нужно установить. Идем на официальную страницу и смотрим установку:

curl -fsSL https://toolbelt.treasuredata.com/sh/install-debian-bullseye-td-agent4.sh | sh

Вроде как все просто.

Следующий момент — взять файл с логами и обработать его. Для этого будем использовать уже готовую директорию с логами, которую получаем с других серверов.

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

td-agent-gem install fluent-plugin-record-modifier --no-document

Этот плагин понадобится для удобной модификации записей. Теперь у нас практически все готово для работы. Приступим…

NGinx

Создаем файл /etc/td-agent/conf.d/nginx_site.conf:

<source>
  @type tail
  tag nginx.site
  path /var/log/10.x.x.x/nginx_snipchi.log
  pos_file /var/log/td-agent/nginx_snipchi.log.pos
  @log_level info
  @id nginx_site
  <parse>
    @type regexp
    expression ^(?<dt_syslog>[^ ]*\s+[^ ]*\s+[^ ]*)\s+(?<server>[^ ]*) (?<programm>[^ ]*) (?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*?) (?<proto>[^"]*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)"(?:\s+(?<http_x_forwarded_for>[^ ]+))?)?$
    time_format %d/%b/%Y:%H:%M:%S %z
  </parse>
</source>

<filter nginx.*>
    @type record_modifier

    <record>
        time ${time}
    </record>
</filter>

#<match nginx.*>
#    @type stdout
#</match>

<match nginx.*>
  @type http
  #open_timeout 2

  endpoint http://10.x.x.x:8123/?user=login&password=password&database=logs&query=INSERT%20INTO%20nginx%20FORMAT%20JSONEachRow

  <format>
    @type json
  </format>

  json_array true

  <buffer>
    flush_interval 10s
  </buffer>

</match>

В данном случае мы берем файл с диска и обрабатываем его построчно выделяя поля из записи с помощью регулярных выражений. Далее прогоняем через фильтр и уже в конце через директиву «match» собираем обработанные записи, накапливаем и отправляем уже в БД в формате JSON. Здесь я использую ClickHouse, но об это я напишу в следующей заметке. Здесь стоит запомнить лишь то, что fluent берет итоговый фвйл JSON и отправляет POST-запросом по протоколу http на указанный URI, а вот ClickHouse в свою очередь спокойно получает запрос и полученный JSON просто раскладывает по таблице.

Если присмотреться в раздел source, то в нашем случае важны следующие параметры:

@type — тип источника

tag — тэг для записей лога

path — путь к файлу лога

pos_file — что-то типа pipe-файла. Он нужен для fluent

@log_level — какие события отфильтровывать

Раздел parse в source объясняет источнику как распарсить полученную запись лога. Для этого используются следующие директивы:

@type — тип будущей записи

expression — регулярное выражение для парсинга записи и дальнейшего преобразования в JSON

time_format — формат времени чтобы система смогла определить формат и преобразовать строку даты/времени в числовое значение.

Еще стоит обрабтить внимание, что один из блоков match закомментирован. Его я использовал просто для отладки. В нем полученный результат записывается в лог самого fluent.

Squid

Создаем файл /etc/td-agent/conf.d/squid.conf:

<source>
  @type tail
  tag squid.access
  path /var/log/10.x.x.x/squid-access.log
  pos_file /var/log/td-agent/squid-access.log.pos
  @log_level info
  @id squid
  <parse>
    @type regexp
    expression ^(?<dt_syslog>[^ ]*\s+[^ ]*\s+[^ ]*)\s+(?<server>[^ ]*)\s+(?<programm>[^ ]*)\s+(?<dt>[^\.]*)\.(?<millis>[^ ]*)\s+(?<processTime>[^ ]*)\s+(?<ipClient>[^ ]*)\s+(?<statusName>[^/]*)/(?<statusCode>[^ ]*)\s+(?<size>[^ ]*)\s+(?<method>[^ ]*)\s+(?<url>[^ ]*)\s+(?<authName>[^ ]*)\s+(?<proxyResult>[^/]*)/(?<proxyAddress>[^ ]*)\s+(?<contentType>[^\s$]*)
    time_format %b %d %H:%M:%S %z
  </parse>
</source>

<filter squid.*>
    @type record_modifier
    remove_keys dt_syslog, programm
    <record>
        domain ${record['url']}
    </record>
    <replace>
        key domain
        #expression (http(s)?://)?(?<dom>[^\s\t/:$]*).*
        expression (?<schema>http(s)?:\/\/)
        replace ""
    </replace>
    <replace>
        key domain
        expression (?<port>:[^\/$]*)
        replace ""
    </replace>
    <replace>
        key domain
        expression \/.*$
        replace ""
    </replace>
</filter>

#<match squid.*>
#    @type stdout
#    <format>
#       @type json
#    </format>
#</match>

<match squid.*>
  @type http
  open_timeout 2

  endpoint http://10.x.x.x:8123/?user=login&password=password&database=logs&query=INSERT%20INTO%20squid%20FORMAT%20JSONEachRow

  <format>
    @type json
  </format>

  json_array true

  <buffer>
    flush_interval 10s
  </buffer>

</match>

Эта настройка предназначена для парсинга логов прокси-сервера Squid. По сути в нем так же ничего сверхъестественного нет. Вся разница заключается в разделе source, а конкретней в expression.

Электронная почта

В моем случае используется Postfix. Вот у него логи немного сложнее, но не намного. Стоит проявить немного внимания и уситчивости и все получится.

Создаем файл /etc/td-agent/conf.d/postfix.conf:

<source>
  type tail
  path /var/log/10.x.x.x/postfix.log
  tag postfix.relay
  #format /^(?<date>[^ ]+) (?<host>[^ ]+) (?<process>[^:]+): (?<message>((?<key>[^ :]+)[ :])? ?((to|from)=<(?<address>[^>]+)>)?.*)$/
  #format /(?<time>[\w]+\s+[\d]+\s[\d:]+)\s+(?<data>.+)/
  time_format %b %d %H:%M:%S
  #format none
  pos_file /var/log/td-agent/postfix-relay.log.pos
    <parse>
        @type regexp
        expression ^(?<time>[^ ]*\s+[^ ]*\s+[^ ]*)\s+(?<server>[^ ]*)\s+(?<daemon>[^/]*)/(?<process>[^\[]*)\[(?<process_id>[^\]]*)\]:\s+(connect to (?<connect_to_host>[^\[]+)\[(?<connect_to_ip>[^\]]+)\](:(?<port>[^:]+):\s(?<status>.+)?)?)?(connect\sfrom\s(?<connect_from>[^\[]+)\[(?<connect_ip_from>[^\]]+)\])?(disconnect\sfrom\s(?<disconnect_from>[^\[]+)\[(?<disconnect_ip_from>[^\]]+)\]\sehlo=(?<ehlo>[^ ]+)\sstarttls=(?<starttls>[^ ]+)\smail=(?<mail>[^ ]+)\srcpt=(?<rcpt>[^ ]+)\sdata=(?<data>[^ ]+)\squit=(?<quit>[^ ]+)\scommands=(?<commands>.+))?(lost connection after CONNECT from (?<lost_conn_host>[^\[]+)\[(?<lost_conn_ip>[^\]]+))?((?<queue_id>[^:]+): (removed(?<removed>))?(client=(?<client>[^\[]+)\[(?<client_ip>[^\]]+\]))?(message\-id=<(?<message_id>[^>]+))?)?(to=<(?<to>[^>]+)>, (orig_to=<(?<orig_to>[^>]+)>, )?(relay=(?<relay_host>[^\[]+)\[(?<relay_ip>[^\]]+)\](:(?<port>\d+))?)?(,\sdelay=(?<delay>[^,]+))?(,\sdelays=(?<delays>[^,]+))?(,\sdsn=(?<dsn>[^,]*))?(,\sstatus=(?<status>[^ ]*)\s)?(?<description>.+)?)?(from=<(?<from>[^>]+)>,\ssize=(?<size>[^,]+),\snrcpt=(?<nrcpt>[^ $]+)(\s(?<description>.+))?)?
        time_format %b %d %H:%M:%S
    </parse>
</source>

<filter postfix.*>
    @type record_modifier

    <record>
        time ${time}
    </record>
</filter>

#<match postfix.*>
#    @type stdout
#    <format>
#       @type json
#    </format>
#</match>

<match postfix.*>
  @type http
  #open_timeout 2

  endpoint http://10.x.x.x:8123/?user=login&password=password&database=logs&query=INSERT%20INTO%20postfix%20FORMAT%20JSONEachRow

  <format>
    @type json
  </format>

  json_array true

  <buffer>
    flush_interval 10s
  </buffer>

</match>

Собственно вся загвоздка в том, что по сути Postfix имеет многострочные логи, а если быть точнее, то логи однострочные, но форматы строк все разные (ну или почти). Но это тоже решается.

Немного велосипеда

Если смотреть в документации fluent, то можно обнаружить, что формат парсинга для того же Nginx уже есть, но он не будет работать, так как после приема логов через syslog в результирующий файл попадает еще несколько дополнительных полей в начало записи. Так что пришлось немного переписать. А вот как это поправить в rsyslog — я, честно, не разобрался.

Кроме всего прочего, судя по документации fluent умеет самостоятельно принимать логи в формате syslog и можно было бы отказаться от серверного rsyslog, но мы так прикинули, что пусть уж лучше будет что-то стандартное, тем более событий не так много, а временно хранящиеся файлы на диске в течении какого-то времени, как бы, хороший тон.

Работай!

После написания правил обработки данных их неплохо было бы включить, иначе работать просто не будут. Для этого редктируем файл /etc/td-agent/td-agent.conf:

@include conf.d/nginx_site.conf
@include conf.d/postfix.conf
@include conf.d/squid.conf

После можно перезапустить агента:

systemctl restart td-agent

Замечание

При написании правил порядок разделов имеет значение. Если сначала написать <match> а потом <source>, то работать не будет. соответственно и обработчики, например <match>, тоже должны идти последовательно в том порядке, который того требует обработка данных.

Результат

Сейчас есть обработчик на одном сервере, который собирает события откуда требуется, обрабатывает и отправляет их в базу. Теперь нужно решить вопрос о хранении.