Разбор данных

Шаблоны dissect состоят из переменных и разделителей. Всё, что определено знаком процента и фигурными скобками %{}, считается переменной, например, %{clientip}. Вы можете назначить переменные любой части данных в поле и затем вернуть только нужные части. Разделители — это любые значения между переменными, такие как пробелы, тире или другие разделители.

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

"message" : "247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0"

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

Первая часть данных похожа на IP-адрес, поэтому вы можете назначить переменную, например, %{clientip}. Следующие два символа — тире с пробелами по обеим сторонам. Вы можете назначить переменную для каждого тире или одну переменную для представления тире и пробелов. Далее идёт набор скобок, содержащих временную метку. Скобки являются разделителями, поэтому их следует включать в шаблон dissect. На данный момент данные и соответствующий шаблон dissect выглядят так:

247.37.0.0 - - [30/Apr/2020:14:31:22 -0500]  

%{clientip} %{ident} %{auth} [%{@timestamp}] 

Используя ту же логику, вы можете создать переменные для оставшихся частей данных. Двойные кавычки являются разделителями, поэтому их нужно включить в шаблон dissect. Шаблон заменяет GET переменной %{verb}, но сохраняет HTTP как часть шаблона.

\"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0

"%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}

Объединение двух шаблонов приводит к шаблону dissect, который выглядит так:

%{clientip} %{ident} %{auth} [%{@timestamp}] \"%{verb} %{request} HTTP/%{httpversion}\" %{status} %{size}

Теперь, когда у вас есть шаблон dissect, как его протестировать и использовать?

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

Например, протестируйте шаблон dissect с помощью API Painless execute, включив в него свой скрипт Painless и один документ, соответствующий вашим данным. Начните с индексирования поля message как типа данных wildcard:

PUT my-index
{
  "mappings": {
    "properties": {
      "message": {
        "type": "wildcard"
      }
    }
  }
}

Если вы хотите получить код HTTP-ответа, добавьте свой шаблон dissect в скрипт Painless, который извлекает значение response. Для извлечения значений из поля используйте эту функцию:

`.extract(doc["<field_name>"].value)?.<field_value>`

В этом примере, message является <field_name>, а response — <field_value>:

POST /_scripts/painless/_execute
{
  "script": {
    "source": """
      String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response)); 
    """
  },
  "context": "long_field", 
  "context_setup": {
    "index": "my-index",
    "document": {          
      "message": """247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] "GET /images/hm_nbg.jpg HTTP/1.0" 304 0"""
    }
  }
}

Результат включает код HTTP-ответа:

{
  "result" : [
    304
  ]
}

Если у вас есть рабочий шаблон dissect, вы можете добавить его во временное поле для обработки данных. Поскольку временные поля не требуют индексации полей, вы обладаете невероятной гибкостью в изменении вашего скрипта и его работы. Если вы уже протестировали шаблон dissect с помощью API Painless execute, вы можете использовать этот тот же самый скрипт Painless в вашем временном поле.

Для начала добавьте поле message как тип wildcard, как в предыдущем разделе, но также добавьте @timestamp как date, на случай, если вы захотите обработать это поле для других сценариев использования:

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

Если вы хотите извлечь код HTTP-ответа, используя свой шаблон dissect, вы можете создать временное поле, как http.response:

PUT my-index/_mappings
{
  "runtime": {
    "http.response": {
      "type": "long",
      "script": """
        String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response));
      """
    }
  }
}

После сопоставления нужных полей проиндексируйте несколько записей из ваших логов в Elasticsearch. Следующий запрос использует API bulk для индексирования сырых логов в my-index:

POST /my-index/_bulk?refresh=true
{"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"}

Вы можете определить простой запрос для поиска определённого HTTP-ответа и возвращения всех связанных полей. Используйте параметр fields API поиска для извлечения временного поля http.response.

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

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

GET my-index/_search
{
  "runtime_mappings": {
    "http.response": {
      "type": "long",
      "script": """
        String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response));
      """
    }
  },
  "query": {
    "match": {
      "http.response": "304"
    }
  },
  "fields" : ["http.response"]
}

{
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "my-index",
        "_type" : "_doc",
        "_id" : "D47UqXkBByC8cgZrkbOm",
        "_score" : 1.0,
        "_source" : {
          "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"
        },
        "fields" : {
          "http.response" : [
            304
          ]
        }
      }
    ]
  }
}