Кибана - Как извлечь поля из существующих логов Kubernetes

У меня есть своего рода стек ELK, с fluentd вместо logstash, работающий как DaemonSet в кластере Kubernetes и отправляющий все журналы из всех контейнеров в формате logstash на сервер Elasticsearch.

Из множества контейнеров, работающих в кластере Kubernetes, некоторые являются контейнерами nginx, которые выводят журналы следующего формата:

121.29.251.188 - [16/Feb/2017:09:31:35 +0000] host="subdomain.site.com" req="GET /data/schedule/update?date=2017-03-01&type=monthly&blocked=0 HTTP/1.1" status=200 body_bytes=4433 referer="https://subdomain.site.com/schedule/2589959/edit?location=23092&return=monthly" user_agent="Mozilla/5.0 (Windows NT 6.1; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0" time=0.130 hostname=webapp-3188232752-ly36o

Поля, видимые в Кибане, как на этом скриншоте:

журнал kibana nginx

Можно ли извлечь поля из этого типа журнала после его индексации?

Коллектор fluentd сконфигурирован со следующим источником, который обрабатывает все контейнеры, поэтому применение формата на этом этапе невозможно из-за очень разных выходных данных из разных контейнеров:

<source>
  type tail
  path /var/log/containers/*.log
  pos_file /var/log/es-containers.log.pos
  time_format %Y-%m-%dT%H:%M:%S.%NZ
  tag kubernetes.*
  format json
  read_from_head true
</source>

В идеальной ситуации я хотел бы дополнить поля, видимые на скриншоте выше, мета-полями в поле "log", такими как "host", "req", "status" и т. Д.

Источник

2 ответа

Решение

После нескольких дней исследований и привыкания к стеку EFK я пришел к специальному решению для EFK, в отличие от ответа в Darth_Vader, который возможен только в стеке ELK.

Подводя итог, я использую Fluentd вместо Logstash, поэтому любое решение grok будет работать, если вы также установите плагин Fluentd Grok, чего я решил не делать, потому что:

Как выясняется, Fluentd имеет свою собственную функцию извлечения полей за счет использования фильтров синтаксического анализатора. Чтобы решить проблему в моем вопросе, прямо перед <match **> после того, как объект строки журнала уже был обогащен полями метаданных и метками kubernetes, я добавил следующее:

<filter kubernetes.var.log.containers.webapp-**.log>
  type parser
  key_name log
  reserve_data yes
  format /^(?<ip>[^-]*) - \[(?<datetime>[^\]]*)\] host="(?<hostname>[^"]*)" req="(?<method>[^ ]*) (?<uri>[^ ]*) (?<http_version>[^"]*)" status=(?<status_code>[^ ]*) body_bytes=(?<body_bytes>[^ ]*) referer="(?<referer>[^"]*)" user_agent="(?<user_agent>[^"]*)" time=(?<req_time>[^ ]*)/
</filter>

Объяснить:

<filter kubernetes.var.log.containers.webapp-**.log> - применить блок ко всем строкам, соответствующим этой метке; в моем случае контейнеры компонента веб-сервера называются webapp-{что-то}

type parser - говорит fluentd применить фильтр парсера

key_name log - применить шаблон только на log свойство строки журнала, а не всей строки, которая является строкой json

reserve_data yes - очень важно, если не указано, весь объект строки журнала заменяется только свойствами, извлеченными из format, так что если у вас уже есть другие свойства, такие как добавленные kubernetes_metadata фильтр, они удаляются, если не добавлять reserve_data вариант

format - регулярное выражение, которое применяется к значению log ключ для извлечения именованных свойств

Обратите внимание, что я использую Fluentd 1.12, поэтому этот синтаксис не полностью совместим с более новым синтаксисом 1.14, но принцип будет работать с небольшими изменениями в объявлении синтаксического анализатора.

Источник

Чтобы извлечь строку журнала в поля, вам, возможно, придется использовать фильтр grok. Что вы можете сделать, так это иметь шаблон регулярного выражения, чтобы он точно соответствовал той части строки журнала, которая вам нужна. Фильтр Grok может выглядеть примерно так:

grok {
    patterns_dir => ["pathto/patterns"]
    match => { "message" => "^%{LOGTIMESTAMP:logtimestamp}%{GREEDYDATA:data}" }         
}                                                 ^-----------------------^ are the fields you would see in ES when log is being indexed

-------------------------------------------------- - ^ LOGTIMESTAMP должно быть определено в вашем файле шаблонов что-то вроде:

LOGTIMESTAMP %{YEAR}%{MONTHNUM}%{MONTHDAY} %{TIME}

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

if "something" in [message]{
     mutate {
         add_field => { "new_field" => %{logtimestamp} }
     }          
}

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

Сообщение в блоге, может быть удобно! Надеюсь это поможет.

Источник
Другие вопросы по тегам