Парсинг логов с помощью Logstash

Эта статья переведена с:Logstash Reference [6.5] » Getting Started with Logstash » Parsing Logs with Logstash, Личный уровень ограничен, пожалуйста, поправьте меня, если я не прав.

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

В этой главе вы создадите конвейер Logstash: Filebeat принимает данные веб-журнала Apache в качестве входных данных, анализирует входной журнал, создает определенные поля и записывает проанализированные данные в кластер Elasticsearch. Вы будете настраивать конвейер Logstash в файле конфигурации, а не в командной строке.

Далее нажмитездесьЗагрузите образец данных и разархивируйте файл.

Настройте Filebeat для отправки журналов в Logstash.

Перед созданием конвейера Logstash необходимо настроить Filebeat для отправки журналов в Logstash. Filebeat — это легкий инструмент сбора журналов с низким уровнем ресурсов, который собирает журналы с каждого сервера и отправляет эти журналы в экземпляр Logstash для обработки. Filebeat разработан для обеспечения надежности и низкой задержки. Filebeat занимает очень мало ресурсов хоста, поэтому плагин ввода Beat эффективно снижает требования к ресурсам экземпляра Logstash.

В типичном приложении Filebeat и Logstash развернуты на разных хостах. Здесь они развернуты на одном хосте для простоты руководства.

Установка Logstash по умолчанию уже содержитBeat inputплагин.BeatПлагин ввода позволяет Logstash конвертировать изElastic BeatФреймворк получает события, что означает, что все, что написано с помощьюBeatФреймворки, работающие с Beats, такие как Packetbeat и Metricbeat, могут отправлять данные о событиях в Logstash.

Чтобы установить Filebeat на хосте источника данных, вы можетеСтраница продуктаЗагрузите соответствующий инсталляционный пакет. смотрите такжеНачало работы с FilebeatПолучите дополнительные инструкции по установке.

После установки Filebeat его необходимо настроить. Найдите и откройте в каталоге установки Filebeatfilebeat.ymlфайл, замените элементы конфигурации в следующих строках. удостоверитьсяpathsЭтот элемент представляет собой путь к ранее загруженному образцу файла журнала Apache (logstash-tutorial.log).

filebeat.prospectors:
- type: log
  paths:
    - /path/to/file/logstash-tutorial.log  # 须填写绝对路径
output.logstash:
  hosts: ["localhost:5044"]

Сохраните конфигурацию. Чтобы упростить настройку, нет необходимости настраивать стандартные конфигурации TLS/SSL в средах сборки.

На хосте источника данных запустите Filebeat, введя следующую команду:

sudo ./filebeat -e -c filebeat.yml -d "publish"

Если вы запускаете Filebeat от имени пользователя root, вам потребуется изменить владельца файла конфигурации (см.Config File Ownership and Permissions).

Filebeat попытается установить соединение через порт 5044. До того, как Logstash запустит активный подключаемый модуль Beats, на этом порту нет ответа, поэтому сообщение, которое вы видите об ошибке соединения на порту 5044 до того, как входной подключаемый модуль неактивен, является нормальным явлением.

Настроить ввод Filebeat для Logstash

Далее вы создадите конфигурацию конвейера Logstash, которая принимает события от Beats с подключаемым модулем ввода Beats в качестве входных данных.

Следующий фрагмент содержимого является основой для конфигурации конвейера:

# The # character at the beginning of a line indicates a comment. Use
# comments to describe your configuration.
input {
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
}

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

Хорошо, продолжайте и создайте каталог logstashfirst-pipeline.confфайл и скопируйте и вставьте вышеприведенное содержимое структуры конфигурации конвейера.

Далее, добавив следующую строку вfirst-pipeline.confВ разделе ввода файла настройте экземпляр Logstash для использования Beats в качестве подключаемого модуля ввода:

    beats {
        port => "5044"
    }

Позже вы настроите Logstash для вывода в Elasticsearch. В настоящее время вы можете добавить следующие строки в раздел вывода, чтобы настроить Logstash для вывода на стандартный вывод при запуске.stdout:

    stdout { codec => rubydebug }

После того, как вышеуказанные шаги выполнены,first-pipeline.confКонфигурация файла должна быть следующей:

input {
    beats {
        port => "5044"
    }
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
    stdout { codec => rubydebug }
}

Приведенную выше конфигурацию можно проверить, выполнив следующую команду:

bin/logstash -f first-pipeline.conf --config.test_and_exit

--config.test_and_exitoption проверяет файл конфигурации и сообщает об ошибках.

Если файл конфигурации проходит проверочные тесты, выполните следующую команду, чтобы запустить Logstash:

bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automaticЭлемент будет периодически автоматически перезагружать конфигурацию, и вы можете изменить конфигурацию, не останавливая и не перезапуская Logstash.

Когда Logstash запустится, вы увидите одну или несколько строк об игнорировании Logstash.pipelines.ymlпредупреждение файла. Вы можете смело игнорировать это предупреждение.pipelines.ymlРоль файла конфигурации заключается в запуске нескольких конвейеров в экземпляре Logstash. В приведенном здесь примере запущен только один конвейер.

Если ваш конвейер работает правильно, он выведет на консоль множество событий, например:

{
    "@timestamp" => 2017-11-09T01:44:20.071Z,
        "offset" => 325,
      "@version" => "1",
          "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
          "host" => "My-MacBook-Pro.local",
    "prospector" => {
        "type" => "log"
    },
        "source" => "/path/to/file/logstash-tutorial.log",
       "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ]
}

Анализ веб-журналов с помощью подключаемого модуля фильтра Grok

Теперь у вас есть работающий конвейер, который может считывать строки журнала из Filebeat, но вы обнаружите, что формат журнала не очень идеален. Исходя из этого, вы хотите проанализировать сообщение, чтобы создать определенные поля. Для этого нужно использоватьgrokПлагин фильтра.

grokПлагин фильтра — один из нескольких плагинов, доступных по умолчанию в Logstash. Дополнительные сведения об управлении подключаемыми модулями Logstash см.Справочная документацияв разделе Управление плагинами.

grokПлагины фильтров позволяют преобразовывать неструктурированные данные журнала в структурированную, легко запрашиваемую форму.

grokПодключаемый модуль фильтра ищет соответствующий шаблон во входных данных журнала, поэтому вам необходимо настроить, как подключаемый модуль распознает соответствующий шаблон, в соответствии с вашими собственными требованиями к варианту использования. Репрезентативная строка в примере журнала веб-сервера выглядит так:

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

IP-адрес в начале строки журнала очень узнаваем, как и отметка времени в скобках. Чтобы проанализировать эти данные, вы можете использовать%{COMBINEDAPACHELOG}шаблон, который структурирует строки журнала Apache в виде следующей таблицы:

Information	Field Name
IP Address	clientip
User ID	ident
User Authentication	auth
timestamp	timestamp
HTTP Verb	verb
Request body	request
HTTP Version	httpversion
HTTP Status Code	response
Bytes served	bytes
Referrer URL	referrer
User agent	agent

если вы строитеgrokнужна помощь с шаблоном, попробуйте использоватьОтладчик Grok. Отладчик Grok — это функция X-Pack, которую можно использовать бесплатно.

редактироватьfirst-pipeline.confфайл конфигурации иfilterраздел заменяется следующим:

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

После того, как вышеуказанная работа будет завершена,first-pipeline.confСодержимое конфигурационного файла следующее:

input {
    beats {
        port => "5044"
    }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    stdout { codec => rubydebug }
}

Сохранить изменения. Поскольку автоматическая перезагрузка конфигурации задана в конфигурации ранее, при повторном изменении конфигурации вам не нужно перезапускать Logstash, чтобы конфигурация вступила в силу. Однако вам нужно заставить Filebeat читать файлы журнала с нуля. Для этого перейдите в окно терминала, где запущен Filebeat, и нажмитеCtrl + CЗакройте Filebeat. Затем удалите файл реестра Filebeat.registry. Например, запустите:

sudo rm data/registry

Поскольку Filebeat сохраняет состояние каждого файла в файле реестра после его чтения, удаление файла реестра заставит Filebea прочитать файл с нуля. Затем перезапустите Filebeat с помощью следующей команды:

sudo ./filebeat -e -c filebeat.yml -d "publish"

Если Filebeat необходимо дождаться, пока Logstash перезагрузит файл конфигурации, прежде чем он начнет обрабатывать события, может возникнуть небольшая задержка. Применяется в LogstashgrokПосле схемы событие будет иметь следующее представление JSON:

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
         "offset" => 325,
           "auth" => "-",
          "ident" => "-",
           "verb" => "GET",
     "prospector" => {
        "type" => "log"
    },
         "source" => "/path/to/file/logstash-tutorial.log",
        "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
     "@timestamp" => 2017-11-09T02:51:12.416Z,
       "response" => "200",
          "bytes" => "203023",
       "clientip" => "83.149.9.216",
       "@version" => "1",
           "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
           "host" => "My-MacBook-Pro.local",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000"
}

Обратите внимание, что события содержат необработанные сообщения, но сообщения журнала также разбиваются на определенные поля.

Улучшите свои данные с помощью плагинов фильтра Geoip

В дополнение к анализу данных журнала для улучшения поиска плагины фильтров также могут получать дополнительную информацию из существующих данных. Например,geoipПлагин ищет IP-адрес, получает информацию о геолокации из адреса и добавляет эту информацию о местоположении в журнал.

использоватьgeoipПлагин настраивает ваш экземпляр Logstash вfirst-pipeline.confДобавьте следующую строку в раздел фильтра файла конфигурации:

    geoip {
        source => "clientip"
    }

geoipКонфигурация плагина требует, чтобы вы указали имя исходного поля, которое содержит информацию об IP-адресе, который вы ищете. В этом примере поле clientip содержит информацию об IP-адресе.

Так как фильтры обрабатываются последовательно, в конфигурационном файле обязательноgeoipчасть вgrokчасть после, и обе вfilterвнутренний.

Когда вы завершили этот шаг,first-pipeline.confСодержание должно быть следующим:

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    stdout { codec => rubydebug }
}

Сохранить изменения. Чтобы заставить Filebeat прочитать файл журнала с самого начала, как вы делали это раньше, закройте Filebeat (Ctrl+C), удалите файл реестра и перезапустите Filebeat, выполнив следующую команду:

sudo ./filebeat -e -c filebeat.yml -d "publish"

Обратите внимание, что события теперь включают информацию о геолокации:

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "geoip" => {
              "timezone" => "Europe/Moscow",
                    "ip" => "83.149.9.216",
              "latitude" => 55.7485,
        "continent_code" => "EU",
             "city_name" => "Moscow",
          "country_name" => "Russia",
         "country_code2" => "RU",
         "country_code3" => "RU",
           "region_name" => "Moscow",
              "location" => {
            "lon" => 37.6184,
            "lat" => 55.7485
        },
           "postal_code" => "101194",
           "region_code" => "MOW",
             "longitude" => 37.6184
    },
    ...

Создайте индекс Elasticsearch

Теперь, когда вы разобрали данные веб-журнала по отдельным полям, вы можете экспортировать данные в Elasticsearch.

Вы можете запустить Elasticsearch на своем собственном хосте или использовать нашу размещенную службу Elasticsearch в Elastic Cloud. И AWS, и GCP предлагают услуги Elasticsearch.Попробуйте сервис Elasticsearch бесплатно.

Конвейеры Logstash могут индексировать данные в кластере Elasticsearch. редактироватьfirst-pipeline.confфайл конфигурации, замените раздел вывода следующим:

output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

В этой конфигурации Logstash использует протокол http для установления соединения с Elasticsearch. В приведенном выше примере предполагается, что Logstash работает на том же хосте, что и Elasticsearch. ты можешь пройтиhostsЭлементы конфигурации аналогичныhosts => [ "es-machine:9092" ]конфигурация для подключения к удаленному работающему экземпляру Elasticsearch.

сюда,first-pipeline.confВвод, фильтр и вывод в файле конфигурации настроены правильно.Содержимое конфигурации выглядит следующим образом:

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

sudo ./filebeat -e -c filebeat.yml -d "publish"

Трубопроводное тестирование

До сих пор конвейер Logstash был настроен для вывода данных в кластер Elasticsearch, поэтому вы можете запрашивать соответствующие данные в Elasticsearch.

на основеgrokОтфильтруйте поля, созданные плагином, и попробуйте проверить запрос на Elasticsearch. будет$ DATEЗаменить текущей датой в форматеYYYY.MM.DD:

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200'

Даты в именах индексов основаны на UTC, а не на часовом поясе, в котором работает Logstash. Если запрос возвращаетindex_not_found_exception,пожалуйста убедитесьlogstash-$DATAправильное имя индекса. Чтобы просмотреть список доступных в настоящее время индексов, вы можете использовать это для запроса:curl 'localhost:9200/_cat/indices?v'.

Вы получите несколько результатов. похожий:

{
  "took": 50,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 98,
    "max_score": 2.793642,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "3IzDnl8BW52sR0fx5wdV",
        "_score": 2.793642,
        "_source": {
          "request": "/presentations/logstash-monitorama-2013/images/frontend-response-codes.png",
          "agent": """"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "geoip": {
            "timezone": "Europe/Moscow",
            "ip": "83.149.9.216",
            "latitude": 55.7485,
            "continent_code": "EU",
            "city_name": "Moscow",
            "country_name": "Russia",
            "country_code2": "RU",
            "country_code3": "RU",
            "region_name": "Moscow",
            "location": {
              "lon": 37.6184,
              "lat": 55.7485
            },
            "postal_code": "101194",
            "region_code": "MOW",
            "longitude": 37.6184
          },
          "offset": 2932,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] "GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1" 200 52878 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"http://semicomplete.com/presentations/logstash-monitorama-2013/"""",
          "@timestamp": "2017-11-09T03:11:35.304Z",
          "response": "200",
          "bytes": "52878",
          "clientip": "83.149.9.216",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:13:45 +0000"
        }
      },
    ...

Попробуйте другой запрос с географической информацией, полученной из IP-адреса. будет$ DATEЗаменить текущей датой в форматеYYYY.MM.DD:

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=geoip.city_name=Buffalo'

В данных из Buffalo есть несколько записей журнала, и результаты запроса выглядят следующим образом:

{
  "took": 9,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 2,
    "max_score": 2.6390574,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "L4zDnl8BW52sR0fx5whY",
        "_score": 2.6390574,
        "_source": {
          "request": "/blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29",
          "agent": """"Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "geoip": {
            "timezone": "America/New_York",
            "ip": "198.46.149.143",
            "latitude": 42.8864,
            "continent_code": "NA",
            "city_name": "Buffalo",
            "country_name": "United States",
            "country_code2": "US",
            "dma_code": 514,
            "country_code3": "US",
            "region_name": "New York",
            "location": {
              "lon": -78.8781,
              "lat": 42.8864
            },
            "postal_code": "14202",
            "region_code": "NY",
            "longitude": -78.8781
          },
          "offset": 22795,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """198.46.149.143 - - [04/Jan/2015:05:29:13 +0000] "GET /blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29 HTTP/1.1" 200 9316 "-" "Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"-"""",
          "@timestamp": "2017-11-09T03:11:35.321Z",
          "response": "200",
          "bytes": "9316",
          "clientip": "198.46.149.143",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:29:13 +0000"
        }
      },
     ...

Если вы используете Kibana для визуализации данных журнала, вы также можете просмотреть данные, прочитанные Filebeat в Kibana:

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

Вы успешно создали конвейер, который использует Filebeat для чтения информации веб-журнала Apache в качестве входных данных, анализирует журнал для создания указанных полей и выводит проанализированные данные журнала в кластер Elasticsearch. Далее вы узнаете, как создать конвейер, использующий несколько подключаемых модулей ввода и вывода.

Обратите внимание на публичный аккаунт WeChat и регулярно публикуйте статьи!