Ретроспективная корреляция событий

Данная инструкция описывает процесс настройки ретроспективной корреляции событий в системе.

О ретроспективной корреляции событий

Ретроспективная корреляция событий — это процесс обработки ранее собранных и сохраненных событий с помощью правил корреляции для выявления инцидентов в исторических данных.

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

Настройка ретроспективной корреляции поддерживается только в том случае, если СУБД ClickHouse установлена в выделенном кластере.

Подготовка к настройке

Для ретроспективной нормализации и корреляции необходим драйвер БД ClickHouse версии 0.9.8 или выше. Чтобы его настроить:

  1. Если драйвер БД ClickHouse отсутствует в системе, добавьте его. Для этого:

    1. Скачайте файл драйвера clickhouse-jdbc-<version>-all-dependencies.jar из репозитория ClickHouse, где <version> — версия драйвера, не ниже 0.9.8.

      Например, драйвер для версии 0.9.8 можно скачать по следующей ссылке.

    2. В разделе Ресурсы → Драйверы БД добавьте скачанный драйвер в систему.

  2. В разделе Ресурсы → Секреты создайте секрет со строкой подключения к вашей БД ClickHouse:

    jdbc:clickhouse://<address>:<port>/default?user=reader&password=<password>&compress=0

    Здесь:

    • <address> — полное доменное имя (FQDN) или IP-адрес сервера, где развернут ClickHouse. Если вы указываете FQDN, убедитесь, что оно корректно разрешается в IP-адрес на машине, которая выполняет роль worker-узла в кластере Kubernetes, где развернута система.

    • <port> — HTTP-порт, который вы задавали при развертывании и настройке ClickHouse.

    • <password> — пароль от учетной записи reader, который вы задавали при развертывании и настройке ClickHouse.

Настройка ретроспективной корреляции

Для настройки ретроспективной корреляции событий выполните следующие шаги:

  1. При необходимости создайте новое хранилище событий, куда будут записываться события после ретроспективной корреляции.

  2. Перейдите в раздел Ресурсы → Коллекторы.

  3. Создайте новый коллектор для выполнения задачи ретроспективной корреляции и включите его.

  4. Добавьте новый конвейер в коллекторе.

  5. Откройте конфигурацию конвейера.

  6. Добавьте на конвейер элемент Точка входа со следующими параметрами:

    • Название: введите название точки входа.

    • Тип точки входа: выберите вариант Database.

    • Драйвер базы данных: выберите необходимый драйвер БД ClickHouse.

    • Адрес подключения: выберите созданную ранее строку подключения к БД ClickHouse.

    • SQL-запрос: введите SQL-запрос для выборки событий из хранилища.

      Пример запроса:
      SELECT
    * EXCEPT (
        baseEventIds,
        baseEventTimestamps,
        groupedBy,
        array_field1,
        array_field2,
        array_field3,
        array_address1,
        array_address2,
        mitre_tts,
        raw
    )
FROM EventStorage_<ID>_local
WHERE
  type IN (1, 2, 3, 4)
  AND timestamp > toDateTime64(toUInt32(?) + 1, 3)
ORDER BY timestamp ASC
LIMIT 10000

      Здесь:

      • <ID> — идентификатор хранилища, события из которого необходимо отправить на повторную корреляцию.

      Как получить идентификатор хранилища

      Через карточку хранилища:

      1. Перейдите в раздел Ресурсы → Хранилища событий.

      2. Выберите требуемое хранилище в списке. В правой части экрана отобразится карточка с подробной информацией.

      3. Скопируйте идентификатор хранилища.

      4. Замените в идентификаторе все символы - на _.

      Через диспетчер запросов:

      1. Перейдите в раздел Поиск.

      2. Выполните поиск событий в требуемом хранилище.

      3. В разделе Инструменты → Диспетчер запросов перейдите на вкладку Журнал запросов.

      4. Выберите выполненный запрос в списке. В правой части экрана отобразится карточка с подробной информацией.

      5. В поле Запрос SQL скопируйте идентификатор хранилища, указанный в операторе FROM.

    • Поле идентификатора: укажите ключ timestamp и задайте начальное значение времени в формате Unix. Это значение определяет момент времени, начиная с которого были собраны сырые события, требующие повторной корреляции.

      Для перевода даты и времени в формат Unix можно воспользоваться онлайн-конвертером.
      Пример настройки временного диапазона

      Предположим, что на повторную корреляцию отправляются события, собранные с 23 декабря 2025 года, что соответствует значению 1766498149 в формате Unix, и до текущего момента.

      Для этого в параметре Поле идентификатора необходимо добавить ключ timestamp со значением 1766498149.

      SQL-запрос будет выглядеть следующим образом:

      SELECT
    * EXCEPT (
        baseEventIds,
        baseEventTimestamps,
        groupedBy,
        array_field1,
        array_field2,
        array_field3,
        array_address1,
        array_address2,
        mitre_tts,
        raw
    )
FROM EventStorage_<ID>_local
WHERE
  type IN (1, 2, 3, 4)
  AND timestamp > toDateTime64(toUInt32(?) + 1, 3)
ORDER BY timestamp ASC
LIMIT 10000

      Здесь:

      • <ID> — идентификатор хранилища, события из которого необходимо отправить на повторную корреляцию.

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

      AND timestamp < toDateTime64(1766692377, 3)

      Дополнительно в операторе WHERE можно настроить фильтрацию по любым полям модели, кроме тех, которые перечислены в блоке EXCEPT в SQL-запросе.

    • Интервал запроса: укажите частоту получения событий из базы данных ClickHouse. Рекомендуемое значение — 5 секунд или более.

    • Максимальное количество строк: укажите значение, соответствующее параметру LIMIT в SQL-запросе.

  7. Добавьте на конвейер элемент VRL-трансформация со следующим кодом:

    . = parse_json(to_string(.raw) ?? .raw) ?? .raw

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

  8. Соедините добавленную точку входа и VRL-трансформацию.

  9. Добавьте на конвейер прочие элементы для обработки и корреляции событий.

    Рекомендации по настройке конвейеров обработки событий для различных источников приведены в руководстве по настройке источников на справочном портале R-Vision.

  10. При необходимости добавьте на конвейер элемент Сервис оповещений и укажите в нем хранилище событий, на основе которых будут формироваться оповещения.

  11. Добавьте на конвейер элемент Конечная точка типа Хранилище событий. Укажите в настройках элемента хранилище, в которое будут записываться события после ретроспективной корреляции.

  12. Установите конфигурацию конвейера.

Если при работе конвейера возникнут ошибки, обратитесь к разделу Устранение проблем при ретроспективной обработке событий.

Была ли полезна эта страница?

+7 (499) 755 55 70 sales@rvision.ru

О компании Пользовательское соглашение Политика cookies

Обратная связь