Настройка параметров обработки входящих данных лога с помощью правил обработки логов

Материал из Документация Ключ-АСТРОМ

Настройка параметров обработки входящих данных лога с помощью правил обработки логов

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

Для кого это предназначено?

Данная статья предназначена для администраторов Ключ-АСТРОМ, занимающихся настройкой правил обработки логов.

Чему вы научитесь?

В этом руководстве вы будете использовать правила обработки логов для:

  • Исправления нераспознанных атрибутов лога.
  • Создания метрики для данных лога.
  • Извлечения полей из JSON-контента.
  • Извлечения атрибутов из различных форматов.
  • Использования нескольких команд PARSE в одном правиле.
  • Использования специализированных средств сопоставления.
  • Изменения любого атрибута лога.
  • Добавления атрибута.
  • Выполнения простых математических операций над атрибутами.
  • Удаления атрибута.
  • Отбрасывания события в логе.
  • Маскирования атрибутов.
  • Переименования атрибутов.
  • Работы с типами данных входных полей.

Прежде чем начать

Предварительные требования

  • Вы настроили прием логов.
  • У вас есть необходимые права для настройки правил обработки логов.
  • У вас есть необходимые права для создания метрик логов.

Предварительные знания

  • Обработка логов с помощью классического конвейера
  • DQL-сопоставитель в логах
  • Команды обработки логов
  • Типы данных обработки логов
  • Функции обработки логов

Исправление нераспознанной метки времени и уровня логирования

Вы можете исправить неопознанные атрибуты timestamp и loglevel, видимые в средстве просмотра логов, на основе соответствующего источника лога.

В этом примере предположим, что вы видите сохраненное событие в логе, где log.source установлено в /var/log/myapp/application.log.#. Вы замечаете несколько вещей, которые хотите исправить:

  • В логе содержится нераспознанный формат метки времени, который вы хотите обрабатывать как метку времени события лога.
  • Надлежащим образом не обнаружен loglevel.

Итак, вы хотите преобразовать данные лога таким образом, чтобы они содержали правильные значения в полях timestamp и loglevel, и хотите добавить новый атрибут thread.name, содержащий правильно извлеченное значение.

Для разрешения неопознанных меток времени и уровня логирования:

  1. Перейдите в Настройки > Мониторинг логов > Обработка.
  2. Выберите Добавить правило и укажите следующие свойства правила обработки.
    • Название правила: Исправить метку времени и уровень логирования для MyApp
    • Сопоставитель: matchesValue(log.source, "/var/log/myapp/application.log.#").
    • Определение процессора: PARSE(content, "TIMESTAMP('MMMMM d, yyyy HH:mm:ss'):timestamp ' [' LD:thread.name '] ' UPPER:loglevel")

Объяснение определения

Данное определение процессора извлекает метку времени, имя потока и уровень логирования.

  • Функция сопоставления TIMESTAMP используется для поиска определенного формата даты и времени, а найденное значение устанавливается в качестве существующего атрибута timestamp.
  • Сопоставитель LD (Line Data) используется для сопоставления любых символов между литералами ' [' и '] '.
  • UPPER используется для сопоставления заглавных букв.
  • Оставшаяся часть содержимого не совпадает.

Вы можете протестировать свой DQL-сопоставитель перед использованием его в правиле обработки логов.

В поле Пример лога введите следующий фрагмент данных лога и выберите Проверить правило. 

{
   "event.type": "LOG",
   "content": "April 24, 2022 09:59:52 [myPool-thread-1] INFO Lorem ipsum dolor sit amet",
   "status": "NONE",
   "timestamp": "1650889391528",
   "log.source": "/var/log/myapp/application.log.#",
   "loglevel": "NONE"
}

Обработанные данные лога отображаются в разделе Результаты теста. Поля timestamp и loglevel имеют корректные значения. Дополнительный атрибут thread.name также извлекается корректно. 

{
   "content": "April 24, 2022 09:59:52 [myPool-thread-1] INFO Lorem ipsum dolor sit amet",
   "timestamp": "1650794392000",
   "event.type": "LOG",
   "status": "NONE",
   "log.source": "/var/log/myapp/application.log.#",
   "loglevel": "INFO",
   "thread.name": "myPool-thread-1"
}
  1. Нажмите Сохранить изменения, чтобы сохранить правило обработки логов.

По мере загрузки новых данных лога вы увидите обработанные данные в окне просмотра логов.

Создание метрики для данных лога

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

REPORT RequestId: 000d000-0e00-0d0b-a00e-aec0aa0000bc Duration: 5033.50 ms Billed Duration: 5034 ms Memory Size: 1024 MB Max Memory Used: 80 MB Init Duration: 488.08 ms

Кроме того, эта запись в логе содержит атрибут cloud.provider со значением aws.

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

  1. Перейдите в Настройки > Мониторинг логов > Обработка.
  2. Выберите Добавить правило и укажите следующие свойства правила обработки.
    • Название правила: Сервисы AWS - оплачиваемая продолжительность
    • Сопоставитель: matchesPhrase(content, "Billed Duration") and matchesValue(cloud.provider, "aws")
    • Определение процессора: PARSE(content, "LD 'Billed Duration:' SPACE? INT:aws.billed.duration"). Это определение извлекает значение оплачиваемой продолжительности.

Вы можете протестировать свой DQL-сопоставитель перед использованием его в правиле обработки логов.

В поле Пример лога введите следующий фрагмент данных лога и выберите Проверить правило. 

{
  "event.type": "LOG",
  "content": "REPORT RequestId: 000d000-0e00-0d0b-a00e-aec0aa0000bc\tDuration: 5033.50 ms\tBilled Duration: 5034 ms\tMemory Size: 1024 MB\tMax Memory Used: 80 MB\t\n",
  "status": "INFO",
  "timestamp": "1651062483672",
  "cloud.provider": "aws",
  "cloud.account.id": "999999999999",
  "cloud.region": "eu-central-1",
  "aws.log_group": "/aws/lambda/aws-dev",
  "aws.log_stream": "2022/04/27/[$LATEST]0d00000daa0c0c0a0a0e0ea0eccc000f",
  "aws.region": "central-1",
  "aws.account.id": "999999999999",
  "aws.service": "lambda",
  "aws.resource.id": "aws-dev",
  "aws.arn": "arn:aws:lambda:central-1:999999999999:function:aws-dev",
  "cloud.log_forwarder": "999999999999:central-1:astromkey-aws-logs",
  "loglevel": "INFO"
}

Обработанные данные лога отображаются в разделе Результаты теста. Они дополнены новым атрибутом aws.billed.duration. 

{
  "event.type": "LOG",
  "content": "REPORT RequestId: 000d000-0e00-0d0b-a00e-aec0aa0000bc\tDuration: 5033.50 ms\tBilled Duration: 5034 ms\tMemory Size: 1024 MB\tMax Memory Used: 80 MB\t\n",
  "status": "INFO",
  "timestamp": "1651062483672",
  "cloud.provider": "aws",
  "cloud.account.id": "999999999999",
  "cloud.region": "eu-central-1",
  "aws.log_group": "/aws/lambda/aws-dev",
  "aws.log_stream": "2022/04/27/[$LATEST]0d00000daa0c0c0a0a0e0ea0eccc000f",
  "aws.region": "central-1",
  "aws.account.id": "999999999999",
  "aws.service": "lambda",
  "aws.resource.id": "aws-dev",
  "aws.arn": "arn:aws:lambda:central-1:999999999999:function:aws-dev",
  "cloud.log_forwarder": "999999999999:central-1:astromkey-aws-logs",
  "loglevel": "INFO",
  "aws.billed.duration": "5034"
}

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

  1. Нажмите Сохранить изменения, чтобы сохранить правило обработки логов.
  2. Перейдите в Настройки > Мониторинг логов > Извлечение метрик.
  3. Выберите Добавить метрику лога и укажите следующие свойства метрики лога, чтобы создать метрику лога на основе полученного идентификатора продукта (aws.billed.duration).
    • Ключ к метрике: log.aws.billed.duration
    • Сопоставитель: matchesPhrase(content, "Billed Duration") and matchesValue(cloud.provider, "aws")
    • Измерение метрики: значение атрибута
    • Атрибут: aws.billed.duration
  4. Выберите Сохранить изменения, чтобы сохранить метрику лога.

Этот показатель отображается в Визуализации данных, и вы можете использовать его во всей системе Ключ-АСТРОМ, как и любой другой показатель. Вы можете добавить его на свою панель мониторинга, включить в анализ и даже использовать для создания оповещений.

Доступность метрик логов

Созданная метрика лога становится доступной только после того, как будут получены новые данные лога и они будут соответствовать запросу лога (Matcher), определенному вами при создании метрики. Убедитесь, что новые данные лога были получены, прежде чем использовать метрику лога в других областях Ключ-АСТРОМ.

Тестирование DQL-сопоставителей

Перед использованием в правиле обработки логов вы можете протестировать свой DQL-сопоставитель, чтобы убедиться в его корректности.

Для тестирования DQL-сопоставителя:

  1. Перейдите в расширенный режим фильтрации логов.
    • Последняя версия: Перейдите в раздел Логи. Рядом с текстовым полем Тип для фильтрации выберите (меню Действия) > Редактировать DQL-запрос.
    • Classic: Перейдите в раздел Логи и события Classic и включите расширенный режим в поле запроса.
  2. Введите DQL-запрос для поиска необходимых данных лога и выберите Выполнить запрос.
  3. Изменяйте DQL-запрос до тех пор, пока не получите ожидаемый результат.
  4. Удовлетворившись результатом фильтрации, скопируйте функцию matchesValue DQL-запроса в буфер обмена. Это ваш DQL-сопоставитель, который можно использовать в правилах обработки логов.

Изучите дополнительные примеры

Вы можете настроить правила обработки логов в соответствии со своими потребностями. Ниже приведены несколько примеров, которые могут подойти для некоторых ваших задач. Выполните все шаги, описанные в разделе «Исправление нераспознанной метки времени и уровня логирования», но установите свойства правила обработки логов на значения, описанные ниже. В частности, замените Определение обработчика предоставленным значением. Вы также можете использовать предоставленный пример лога для проверки вашего правила обработки логов.

Извлечение полей из JSON-содержимого

В этом примере вы видите строку лога, имеющую следующую JSON-структуру: 

{"intField": 13, "stringField": "someValue", "nested": {"nestedStringField1": "someNestedValue1", "nestedStringField2": "someNestedValue2"} }

Анализ полей из JSON в плоском режиме

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

Затем вы можете использовать команду FIELDS_RENAME, чтобы задать подходящие вам имена.

  • Определение процессора:
PARSE(content, "JSON{STRING:stringField}(flat=true)")
| FIELDS_RENAME(better.name: stringField)
  • Образец лога:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "better.name": "someValue"
}

Анализ вложенных полей из JSON

Вы также можете анализировать больше полей (включая вложенные), используя JSON-сопоставитель без плоского режима. В результате вы получаете объект VariantObject, который можно обрабатывать дальше. Например, вы можете создать атрибут верхнего уровня из его внутренних полей.

  • Определение процессора:
PARSE(content, "
JSON{
  STRING:stringField,
  JSON {STRING:nestedStringField1}:nested
}:parsedJson")
| FIELDS_ADD(top_level.attribute1: parsedJson["stringField"], top_level.attribute2: parsedJson["nested"]["nestedStringField1"])
| FIELDS_REMOVE(parsedJson)
  • Образец лога:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "top_level.attribute1": "someValue",
  "top_level.attribute2": "someNestedValue1"
}

Анализ всех полей из JSON в режиме автоматического обнаружения

Иногда вас интересуют все поля JSON. Вам не обязательно перечислять все атрибуты. Вместо этого можно использовать JSON-сопоставитель в режиме автоматического обнаружения. В результате вы получаете объект VARIANT_OBJECT, который можно дополнительно обработать. Например, вы можете создать атрибут верхнего уровня из его внутренних полей.

  • Определение процессора:
PARSE(content, "JSON:parsedJson")
| FIELDS_ADD(f1: parsedJson["intField"],
f2: parsedJson["stringField"],
f3: parsedJson["nested"]["nestedStringField1"],
f4: parsedJson["nested"]["nestedStringField2"])
| FIELDS_REMOVE(parsedJson)
  • Образец лога:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "f1": "13",
  "f2": "someValue",
  "f3": "someNestedValue1",
  "f4": "someNestedValue2"
}

Анализ любого поля из JSON, рассматривая его содержимое как обычный текст

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

  • Определение процессора:
PARSE(content, "LD '\"stringField\"' SPACE? ':' SPACE? DQS:newAttribute")
  • Образец лога:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "newAttribute": "someValue"
}

Сглаживание вложенных JSON-структур

В этом примере мы преобразуем вложенные JSON-структуры в плоскую структуру с помощью команды. Подробнее об используемых параметрах см. fieldsFlatten в разделе «Команды структурирования DQL».

  • Определение процессора:
| parse content, "JSON:parsedContent"
| fieldsFlatten parsedContent, prefix: "pref.", depth: 2
  • Образец лога:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"stringField\": \"someValue\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "pref.intField": "13",
  "pref.stringField": "someValue",
  "pref.nested.nestedStringField1": "someNestedValue1",
  "pref.nested.nestedStringField2": "someNestedValue2"
}

Извлечение атрибутов из различных форматов

В рамках одного шаблона можно извлекать атрибуты из разных форматов.

В этом примере одно или несколько приложений записывают идентификатор пользователя, который вы хотите извлечь в качестве отдельного атрибута лога. Формат лога непоследователен, поскольку он включает различные схемы для записи идентификатора пользователя: user ID=, userId=, userId:, или user ID =.

03/22 08:52:51 INFO user ID = 1234567 Call = 0319 Result = 0
03/22 08:52:51 INFO UserId = 1234567 Call = 0319 Result = 0
03/22 08:52:51 INFO userId=1234567 Call = 0319 Result = 0
03/22 08:52:51 INFO user ID: 1234567 Call = 0319 Result = 0
03/22 08:52:51 INFO User ID: 1234567 Call = 0319 Result = 0
03/22 08:52:51 INFO userid: 1234567 Call = 0319 Result = 0

С помощью необязательного модификатора (вопрос ?) и Alternative Groups можно охватить все подобные случаи одним выражением шаблона:

  • Определение процессора:
PARSE(content, "
  LD //сопоставляет любой текст в пределах одной строки
  ('user'| 'User') //user или литерал User
  SPACE? //необязательное пространство
  ('id'|'Id'|'ID') //соответствует любому из этих
  SPACE? //необязательное пространство
  PUNCTUATION? //необязательная пунктуация
  SPACE? //необязательное пространство
  INT:my.user.id
")

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

Использование нескольких команд PARSE в одном правиле

Вы можете обрабатывать различные форматы или выполнять дополнительный анализ уже обработанных атрибутов с помощью нескольких команд PARSE (соединенных каналами |) в рамках одного правила обработки.

  • Определение процессора:
PARSE(content, "JSON{STRING:message}(flat=true)") | PARSE(message, "LD 'user ' INT:user.id ': ' LD:error.message")

Здесь вы извлекаете поле сообщения, идентификатор пользователя и сообщение об ошибке.

  • Образец лога:
{
  "content": "{\"intField\": 13, \"message\": \"Error occurred for user 12345: Missing permissions\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }"
}
  • Результат теста:
{
  "content": "{\"intField\": 13, \"message\": \"Error occurred for user 12345: Missing permissions\", \"nested\": {\"nestedStringField1\": \"someNestedValue1\", \"nestedStringField2\": \"someNestedValue2\"} }",
  "message": "Error occurred for user 12345: Missing permissions",
  "user.id": "12345",
  "error.message": "Missing permissions"
}

Использование специализированных средств сопоставления

Мы предоставляем исчерпывающий список сопоставителей, упрощающих построение шаблонов.

  • Определение процессора:
PARSE(content, "ISO8601:timestamp SPACE UPPER:loglevel SPACE IPADDR:ip SPACE DQS:request SPACE INTEGER:code")
  • Образец лога:
{
  "content": "2022-05-11T13:23:45Z INFO 192.168.33.1 \"GET /api/v2/logs/ingest HTTP/1.0\" 200"
}
  • Результат теста:
{
  "content": "2022-05-11T13:23:45Z INFO 192.168.33.1 \"GET /api/v2/logs/ingest HTTP/1.0\" 200",
  "timestamp": "1652275425000",
  "loglevel": "INFO",
  "ip": "192.168.33.1",
  "request": "GET /api/v2/logs/ingest HTTP/1.0",
  "code": "200"
}

Изменение любого атрибута из лога

С помощью правил обработки логов вы можете изменять любой атрибут из лога, а не только его content.

Если не указано иное, правило обработки работает только с полем, доступным только для чтения content. Для его работы с другими атрибутами событий лога необходимо использовать команду USING.

  • Определение процессора:
USING(INOUT status:STRING, content)
| FIELDS_ADD(status:IF_THEN(status == 'WARN' AND content CONTAINS('error'), "ERROR"))

В этом определении объявляются два входных атрибута: статус записи и содержимое только для чтения. Далее проверяется, соответствует ли статус WARN и содержит ли содержимое текст error. Если оба условия верны, правило перезаписывает status значением ERROR.

  • Образец лога:
{
  "log.source": "using",
  "timestamp": "1656011002196",
  "status": "WARN",
  "content": "Error message"
}
  • Результат теста:
{
  "log.source": "using",
  "timestamp": "1656011002196",
  "status": "ERROR",
  "content": "Error message"
}

Добавление атрибута в структуру события лога

Вы можете добавить новый атрибут к текущей структуре событий лога. Команда FIELDS_ADD может использоваться для добавления дополнительных атрибутов верхнего уровня в лог.

  • Определение процессора:
FIELDS_ADD(content.length: STRLEN(content), content.words: ARRAY_COUNT(SPLIT(content,"' '")))

Это определение добавляет два атрибута: первый хранит длину, а второй — количество слов в поле содержимого.

  • Образец лога:
{
  "log.source": "new_attributes",
  "timestamp": "1656010654603",
  "content": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Duis."
}
  • Результат теста:
{
  "content": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Duis.",
  "timestamp": "1656010654603",
  "log.source": "new_attributes",
  "content.length": "62",
  "content.words": "9"
}

Выполнение простых математических операций над атрибутами

Благодаря множеству доступных функций и операторов, выполнять вычисления очень просто.

  • Определение процессора:
PARSE(content,"LD 'total: ' INT:total '; failed: ' INT:failed")
| FIELDS_ADD(failed.percentage: 100.0 * failed / total + '%')
| FIELDS_REMOVE(total, failed)

В этом определении процессора мы анализируем значения total и failed, вычисляем процент ошибок и объединяем полученное значение со знаком процента. Затем мы сохраняем его в новом атрибуте с именем failed.percentage и удаляем временные поля.

  • Образец лога:
{
  "timestamp": "1656011338522",
  "content": "Lorem ipsum total: 1000; failed: 250"
}
  • Результат теста:
{
  "content": "Lorem ipsum total: 1000; failed: 250",
  "timestamp": "1656011338522",
  "failed.percentage": "25.0%"
}

Удаление атрибута из сообщения лога

Чтобы удалить атрибут события, являющийся частью исходной записи, сначала необходимо объявить его как INOUT поле ввода, доступное для записи (опция), с помощью команды USING, а затем явно удалить его с помощью команды FIELDS_REMOVE, чтобы он не присутствовал в выходных данных преобразования.

  • Определение процессора:
USING(INOUT redundant.attribute:STRING)
| FIELDS_REMOVE(redundant.attribute)

В этом определении процессора мы объявляем redundant.attribute обязательный записываемый атрибут STRING типа, а затем удаляем его.

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

  • Образец лога:
{
  "redundant.attribute": "value",
  "timestamp": "1656011525708",
  "content": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nulla ac neque nisi. Nunc accumsan sollicitudin lacus."
}
  • Результат теста:
{
  "content": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nulla ac neque nisi. Nunc accumsan sollicitudin lacus.",
  "timestamp": "1656011525708"
}

Отбрасывание события в логе

Все события лога можно удалить с помощью команды FILTER_OUT. Событие удаляется, когда выполняется условие, переданное в качестве параметра команды.

Сброс на основе предварительного сопоставления

В большинстве случаев достаточно исключить все события, которые были предварительно сопоставлены.

Например, если мы хотим отбросить все события DEBUG и TRACE, мы можем настроить запрос сопоставления так, чтобы он соответствовал любому из этих статусов, а затем использовать команду FILTER_OUT для перехвата всего.

  • Matcher:
status=="DEBUG" or status=="TRACE"
  • Определение процессора:
FILTER_OUT(true)
  • Образец лога:
{
  "status": "DEBUG",
  "content": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nulla ac neque nisi. Nunc accumsan sollicitudin lacus."
}

Таким образом, все записи со статусом DEBUG или TRACE отбрасываются.

Усовершенствованное условие сброса

Также можно добавить дополнительную логику и не отбрасывать все предварительно сопоставленные события.

В следующем примере мы отбрасываем входящие события, время выполнения которых составляет менее 100 мс.

  • Определение процессора:
PARSE(content, "LD 'My monitored service call took ' INT:took 'ms'")
| FILTER_OUT(took < 100)
| FIELDS_REMOVE(took)
  • Образец лога:
{
  "content": "2022-06-23 06:52:35.280 UTC INFO My monitored service call took 97 ms"
}

Маскирование любого атрибута

При изменении содержимого или любого другого атрибута необходимо объявить его доступным для записи (INOUT) с помощью команды USING. REPLACE_PATTERN — это очень мощная функция, которая может быть полезна, когда нужно скрыть какую-либо часть атрибута.

Маскировка IP-адресов, вариант 1

В следующем примере мы маскируем IP-адрес, устанавливая значение 0 для последнего октета.

  • Определение процессора:
USING(INOUT ip)
| FIELDS_ADD(ip: IPADDR(ip) & 0xFFFFFF00l)
  • Образец лога:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009021053",
  "ip": "192.168.0.12"
}
  • Результат теста:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009021053",
  "ip": "192.168.0.0"
}

Маскировка IP-адресов, вариант 2

В следующем примере мы маскируем IP-адрес, устанавливая значение xxx для последнего октета.

  • Определение процессора:
USING(INOUT ip)
| FIELDS_ADD(ip: REPLACE_PATTERN(ip, "(INT'.'INT'.'INT'.'):not_masked INT", "${not_masked}xxx"))
  • Образец лога:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009021053",
  "ip": "192.168.0.12"
}
  • Результат теста:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009021053",
  "ip": "192.168.0.xxx"
}

Скрытие адресов электронной почты

В следующем примере мы маскируем весь адрес электронной почты с помощью sha1 (защищенного хеш-алгоритма).

  • Определение процессора:
USING(INOUT email)
| FIELDS_ADD(email: REPLACE_PATTERN(email, "LD:email_to_be_masked", "${email_to_be_masked|sha1}"))
  • Образец лога:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009924312",
  "email": "john.doe@astromkey.com"
}
  • Результат теста:
{
  "content": "Lorem ipsum",
  "timestamp": "1656009924312",
  "email": "9940e79e41cbf7cc452b137d49fab61e386c602d"
}

Маскировка IP-адресов, адресов электронной почты и номеров кредитных карт

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

  • Определение процессора:
USING(INOUT content)
| FIELDS_ADD(content: REPLACE_PATTERN(content, "
(LD 'ip: '):p1 // Lorem ipsum ip:
(INT'.'INT'.'INT'.'):ip_not_masked // 192.168.0.
INT // 12
' email: ':p2 // email:
LD:email_name '@' LD:email_domain // john.doe@astromkey.com
' card number: ':p3 // card number:
CREDITCARD:card // 4012888888881881
", "${p1}${ip_not_masked}xxx${p2}${email_name|md5}@${email_domain}${p3}${card|sha1}"))
  • Образец лога:
{
  "timestamp": "1656010291511",
  "content": "Lorem ipsum ip: 192.168.0.12 email: john.doe@astromkey.com card number: 4012888888881881 dolor sit amet"
}
  • Результат теста:
{
  "content": "Lorem ipsum ip: 192.168.0.xxx email: abba0b6ff456806bab66baed93e6d9c4@astromkey.com card number: 62163a017b168ad4a229c64ae1bed6ffd5e8fb2d dolor sit amet",
  "timestamp": "1656010291511"
}

Переименование атрибутов

С помощью команды FIELDS_RENAME мы можем переименовывать атрибуты, которые были частью исходного события лога, а также атрибуты, созданные внутри обработчика. При изменении любого атрибута из исходного события необходимо объявить его доступным для записи INOUT (writeable).

  • Определение процессора:
USING(INOUT to_be_renamed, content)
| FIELDS_RENAME(better_name: to_be_renamed)
| PARSE(content,"JSON{STRING:json_field_to_be_renamed}(flat=true)")
| FIELDS_RENAME(another_better_name: json_field_to_be_renamed)

С помощью этого определения обработчика мы переименовываем существующий атрибут. Кроме того, мы извлекаем поле из JSON в плоском режиме и переименовываем новый атрибут, созданный автоматически, используя имя поля из JSON.

  • Образец лога:
{
  "timestamp": "1656061626073",
  "content": "{\"json_field_to_be_renamed\": \"dolor sit amet\", \"field2\": \"consectetur adipiscing elit\"}",
  "to_be_renamed": "Lorem ipsum"
}
  • Результат теста:
{
  "content": "{\"json_field_to_be_renamed\": \"dolor sit amet\", \"field2\": \"consectetur adipiscing elit\"}",
  "timestamp": "1656061626073",
  "better_name": "Lorem ipsum",
  "another_better_name": "dolor sit amet"
}

Типы данных входных полей

Определение процессора работает со строго типизированными данными: функции и операторы принимают только объявленные типы данных. Тип присваивается всем входным полям, определенным в команде USING, а также переменным, созданным при разборе или использовании функций приведения типов.

  • Определение процессора:
USING(number:INTEGER, avg:DOUBLE, addr:IPADDR, arr:INTEGER[],bool:BOOLEAN, ts:TIMESTAMP)
| FIELDS_ADD(multi:number*10)
| FIELDS_ADD(avgPlus1:avg+1)
| FIELDS_ADD(isIP: IS_IPV6(addr))
| FIELDS_ADD(arrAvg: ARRAY_AVG(arr))
| FIELDS_ADD(negation: NOT(bool))
| FIELDS_ADD(tsAddYear: TIME_ADD_YEAR(ts,1))
  • Образец лога:
{
  "content": "Lorem ipsum",
  "number": "5",
  "avg": "123.5",
  "addr": "2a00:1450:4010:c05::69",
  "arr": ["1", "2"],
  "bool": "false",
  "ts": "1984-11-30 22:19:59.789 +0000"
}
  • Результат теста:
{
  "content": "Lorem ipsum",
  "number": "5",
  "avg": "123.5",
  "addr": "2a00:1450:4010:c05::69",
  "arr": [
    "1",
    "2"
  ],
  "bool": "false",
  "ts": "1984-11-30 22:19:59.789 +0000",
  "tsAddYear": "1985-11-30T22:19:59.789000000 +0000",
  "negation": "true",
  "arrAvg": "1.5",
  "isIP": "true",
  "avgPlus1": "124.5",
  "multi": "50"
}

Поздравляем!

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

Источник — https://doc.ruscomtech.ru/index.php?title=Настройка_параметров_обработки_входящих_данных_лога_с_помощью_правил_обработки_логов&oldid=6476