Как структурировать и фильтровать смешанные логи в Logstash?

Основная задача при работе со смешанными логами в Logstash — это их парсинг и обогащение, чтобы превратить неструктурированный текст в структурированные события (например, JSON). Это достигается с помощью конвейера фильтров (filters).

1. Парсинг с помощью grok

Фильтр grok использует готовые шаблоны (patterns) для извлечения полей из строки лога. Это первый и самый важный шаг для придания структуры.

filter {
  # Применяем grok-шаблон для логов Nginx
  if [type] == "nginx-access" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
}

2. Обработка многострочных логов (стектрейсов)

Для логов, занимающих несколько строк (например, Java stack traces), используется кодек multiline. Он объединяет строки, относящиеся к одному событию, в одно сообщение до парсинга.

input {
  file {
    path => "/var/log/myapp.log"
    codec => multiline {
      pattern => "^%{TIMESTAMP_ISO8601}" # Новое событие начинается с временной метки
      negate => true
      what => "previous"
    }
  }
}

3. Обогащение и фильтрация с помощью условной логики

Используйте условные операторы (if/else) для применения разных правил к разным типам логов. Часто это используется для добавления тегов, которые помогают в дальнейшей фильтрации и маршрутизации.

filter {
  if [message] =~ /error|exception/i {
    mutate {
      add_tag => ["error_log"]
    }
  }
  # Удаляем отладочные сообщения, чтобы не хранить их
  if [loglevel] == "DEBUG" {
    drop {}
  }
}

4. Маршрутизация в output на основе тегов

В блоке output можно использовать теги, чтобы отправлять разные логи в разные места. Например, ошибки можно отправлять в отдельный индекс Elasticsearch для быстрого поиска и алертинга.

output {
  if "error_log" in [tags] {
    elasticsearch {
      hosts => ["es-node-1:9200"]
      index => "logs-errors-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => ["es-node-1:9200"]
      index => "logs-generic-%{+YYYY.MM.dd}"
    }
  }
}