Elasticsearch 7.9: transforms для агрегации событий безопасности без ETL-джобов

Elasticsearch 7.9 вышел на прошлой неделе, и среди нескольких новинок нас прежде всего интересовали две вещи, которые непосредственно касались текущей задачи: улучшенные transforms и runtime fields в beta. Рассказываем как это легло на практику.

Контекст: таблица аномалий по хостам

Один из клиентских проектов в рамках DWH и аналитики - это централизованный сбор событий безопасности с парка машин в Elasticsearch. Логи идут через Filebeat и Winlogbeat, около 800 хостов, суммарный поток в районе нескольких гигабайт в сутки. Elastic ML давно работает и детектирует аномалии: необычный трафик, нетипичные процессы, подозрительные паттерны логинов.

Проблема была в том, как эти аномалии агрегировать. Исходный индекс с аномалиями - это «сырые» документы по одной записи на событие. ML-детектор пишет туда result_type: anomaly_record, и документов там накапливается порядочно. Когда нужна сводная картина - «сколько аномалий по каждому хосту за последние сутки с разбивкой по severity» - это каждый раз агрегационный запрос по всему индексу. Дашборд в Kibana это терпит, но сырой API-запрос из скрипта работает медленнее чем хотелось бы.

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

Transforms как замена ручному ETL

В 7.9 transforms серьёзно переработали. Pivot transform - это по сути подписка на aggregation: указываешь source index, группировку (в нашем случае host.name + timestamp с bucket по 5 минут), метрики (count по result_type, max по record_score), и Elastic сам пишет результат в destination index и периодически его обновляет.

Конфиг transform выглядит примерно так:

PUT _transform/anomaly-summary-transform
{
  "source": { "index": ["ml-anomalies-*"] },
  "pivot": {
    "group_by": {
      "host": { "terms": { "field": "host.name" } },
      "bucket": {
        "date_histogram": {
          "field": "@timestamp",
          "calendar_interval": "5m"
        }
      }
    },
    "aggregations": {
      "anomaly_count": { "value_count": { "field": "record_score" } },
      "max_score":     { "max":         { "field": "record_score" } },
      "avg_score":     { "avg":         { "field": "record_score" } }
    }
  },
  "dest": { "index": "anomaly-summary-5m" },
  "frequency": "5m",
  "sync": {
    "time": {
      "field": "@timestamp",
      "delay": "60s"
    }
  }
}

sync.time включает continuous mode: transform следит за новыми документами по полю времени и обновляет destination по расписанию. Batch-режим (без sync) по-прежнему доступен, но для живой сводной таблицы нужен именно continuous. Для наших пяти минут это выглядит как живая сводная таблица.

Результат - anomaly-summary-5m с одной строкой на хост на пятиминутный bucket. Дашборд Kibana читает из него напрямую, запросы быстрые, никакого Python-скрипта на cron. Transform управляется через Kibana или API, логи и состояние видны прямо там же.

Runtime fields: когда маппинг не совпал с реальностью

Вторая находка - runtime fields, beta-фича в текущем релизе. Суть в том, что поле объявляется не в маппинге индекса, а в момент запроса (или в маппинге как runtime, но без физического хранения). Значение вычисляется на лету при поиске.

У нас конкретный кейс: в части логов Winlogbeat поле event.code приходит как строка ("4625"), в другой части - как integer (4625). Исторически маппинг выбрал keyword под первые документы, потом пришли агенты с другой версией и часть документов потеряла поле при индексации (dynamic mapping уже не менялся).

До 7.9 выход один - reindex с исправленным маппингом. Это долго, временно нужен алиас на оба индекса, и так далее. С runtime fields можно объявить вычисляемое поле прямо в запросе:

GET winlogbeat-*/_search
{
  "runtime_mappings": {
    "event.code.normalized": {
      "type": "long",
      "script": {
        "source": "if (doc['event.code'].size() != 0) { emit(Long.parseLong(doc['event.code'].value)) }"
      }
    }
  },
  "query": {
    "range": { "event.code.normalized": { "gte": 4600, "lte": 4700 } }
  }
}

Поле вычисляется в момент запроса через Painless-скрипт. Да, это медленнее чем фильтр по индексированному полю - Elastic это честно описывает в документации. Но для ad-hoc анализа аномальных логов, где запросов немного, разница некритична. Это инструмент для случаев «надо разобраться здесь и сейчас, а реиндексация займёт три часа».

Также runtime field можно добавить в маппинг индекса постоянно - и тогда он будет доступен во всех запросах без явного объявления. Физически данные не дублируются. При изменении логики достаточно обновить скрипт в маппинге.

EQL для threat hunting - пока осторожно

Event Query Language в 7.9 - тоже beta. Идея хорошая: EQL позволяет писать запросы по последовательностям событий (sequence by host.name [... event A ...] [... event B within 2m ...]), что для threat hunting очень естественно. Нас интересовало детектирование цепочек: аномальный логин, потом нетипичный процесс на том же хосте в течение минуты.

Потестировали - работает, результаты выглядят логично. Но пока не катим в production-дашборды именно из-за статуса beta. API меняется, и пару несовместимых изменений между 7.8 и 7.9 в EQL уже было. Подождём, пока статус сменится.

Что получилось

Python-скрипт на cron убрали. Сводная таблица аномалий по хостам обновляется каждые пять минут силами самого Elastic. Состояние и задержку transform видно в Kibana без лишних инструментов.

Runtime fields закрыли конкретную болячку с разнотипными event.code без реиндексации. Это не замена нормальному маппингу - скорее аварийный инструмент и инструмент исследования. Но в нужный момент он сэкономил несколько часов.

ILM (Index Lifecycle Management) в этой задаче тоже задействован: transform пишет в индекс anomaly-summary-5m, на него навешана политика rollover по размеру и delete-фаза через 90 дней. Это не новинка 7.9, но хорошо что transform и ILM нормально живут вместе без конфликтов.