Использование Grok

Elastic Stack поставляется с многочисленными предварительно определёнными шаблонами Grok, которые упрощают работу с Grok. Синтаксис для повторного использования шаблонов Grok имеет один из следующих форматов:

Например, предположим, что у вас есть данные сообщения, которые выглядят так:

3.44 55.3.244.1

Первое значение — число, за которым, по всей видимости, следует IP-адрес. Вы можете сопоставить этот текст, используя следующее выражение Grok:

%{NUMBER:duration} %{IP:client}

Для облегчения миграции на Elastic Common Schema (ECS) помимо существующих шаблонов доступен новый набор шаблонов, совместимых с ECS. Новые определения шаблонов ECS фиксируют имена полей событий, совместимые со схемой.

Набор шаблонов ECS содержит все определения шаблонов из набора legacy и является прямым заменителем. Используйте параметр ecs-compatability, чтобы переключиться на этот режим.

В файлы, совместимые с ECS, будут добавлены новые функции и улучшения. В шаблоны legacy всё ещё могут поступать исправления ошибок, совместимые с предыдущими версиями.

Вы можете включить предварительно определённые шаблоны Grok в скрипты Painless для извлечения данных. Для тестирования скрипта используйте контексты полей API выполнения Painless или создайте поле runtime, содержащее скрипт. Runtime поля обеспечивают большую гибкость и принимают несколько документов, но API выполнения Painless — отличный вариант, если у вас нет доступа к записи в кластере, где вы тестируете скрипт.

Например, если вы работаете с данными логов Apache, вы можете использовать синтаксис %{COMMONAPACHELOG}, который понимает структуру логов Apache. Пример документа может выглядеть так:

"timestamp":"2020-04-30T14:30:17-05:00","message":"40.135.0.0 - -
[30/Apr/2020:14:30:17 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"

Чтобы извлечь IP-адрес из поля message, вы можете написать скрипт Painless, который использует синтаксис %{COMMONAPACHELOG}. Вы можете проверить этот скрипт с помощью ip контекста поля API выполнения Painless, но давайте вместо этого используем поле runtime.

На основе примера документа проиндексируйте поля @timestamp и message. Для сохранения гибкости используйте тип wildcard для поля message:

PUT /my-index/
{
  "mappings": {
    "properties": {
      "@timestamp": {
        "format": "strict_date_optional_time||epoch_second",
        "type": "date"
      },
      "message": {
        "type": "wildcard"
      }
    }
  }
}

Затем используйте Bulk API, чтобы проиндексировать некоторые данные логов в my-index.

POST /my-index/_bulk?refresh
{"index":{}}
{"timestamp":"2020-04-30T14:30:17-05:00","message":"40.135.0.0 - - [30/Apr/2020:14:30:17 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:30:53-05:00","message":"232.0.0.0 - - [30/Apr/2020:14:30:53 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:12-05:00","message":"26.1.0.0 - - [30/Apr/2020:14:31:12 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:19-05:00","message":"247.37.0.0 - - [30/Apr/2020:14:31:19 -0500] \"GET /french/splash_inet.html HTTP/1.0\" 200 3781"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:22-05:00","message":"247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:27-05:00","message":"252.0.0.0 - - [30/Apr/2020:14:31:27 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:28-05:00","message":"not a valid apache log"}

Теперь вы можете определить поле runtime в отображении, которое включает ваш скрипт Painless и шаблон Grok. Если шаблон совпадает, скрипт выводит значение соответствующего IP-адреса. Если шаблон не совпадает (clientip != null), скрипт просто возвращает значение поля без аварийного завершения.

PUT my-index/_mappings
{
  "runtime": {
    "http.clientip": {
      "type": "ip",
      "script": """
        String clientip=grok('%{COMMONAPACHELOG}').extract(doc["message"].value)?.clientip;
        if (clientip != null) emit(clientip);
      """
    }
  }
}

В качестве альтернативы вы можете определить то же поле runtime в контексте запроса поиска. Определение поля runtime и скрипт точно такие же, как определённые ранее в отображении индекса. Просто скопируйте это определение в запрос поиска в секцию runtime_mappings и добавьте запрос, который соответствует полю runtime. Этот запрос возвращает те же результаты, что и если вы определили запрос поиска для поля runtime http.clientip в отображении вашего индекса, но только в контексте этого конкретного поиска:

GET my-index/_search
{
  "runtime_mappings": {
    "http.clientip": {
      "type": "ip",
      "script": """
        String clientip=grok('%{COMMONAPACHELOG}').extract(doc["message"].value)?.clientip;
        if (clientip != null) emit(clientip);
      """
    }
  },
  "query": {
    "match": {
      "http.clientip": "40.135.0.0"
    }
  },
  "fields" : ["http.clientip"]
}

Используя поле runtime http.clientip, вы можете определить простой запрос для выполнения поиска по определённому IP-адресу и возврата всех связанных полей. Параметр fields в API _search работает для всех полей, даже тех, которые не были отправлены в составе исходного запроса _source:

GET my-index/_search
{
  "query": {
    "match": {
      "http.clientip": "40.135.0.0"
    }
  },
  "fields" : ["http.clientip"]
}

Ответ включает указанный в запросе поиска IP-адрес. Шаблон Grok внутри скрипта Painless извлёк это значение из поля message во время выполнения.

{
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "my-index",
        "_type" : "_doc",
        "_id" : "1iN2a3kBw4xTzEDqyYE0",
        "_score" : 1.0,
        "_source" : {
          "timestamp" : "2020-04-30T14:30:17-05:00",
          "message" : "40.135.0.0 - - [30/Apr/2020:14:30:17 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"
        },
        "fields" : {
          "http.clientip" : [
            "40.135.0.0"
          ]
        }
      }
    ]
  }
}