Поиск EQL

Язык запросов к событиям (EQL) — это язык запросов для данных временных рядов на основе событий, таких как журналы, метрики и трассировки.

Для выполнения поиска EQL, поток или индекс искомых данных должны содержать поля временная метка и категория события. По умолчанию EQL использует поля @timestamp и event.category из Общей схемы данных Elasticsearch (ECS). Чтобы использовать другое поле временной метки или категории событий, см. Укажите поле временной метки или категории события.

Используйте API поиска EQL для выполнения базового запроса EQL.

GET /my-data-stream/_eql/search
{
  "query": """
    process where process.name == "regsvr32.exe"
  """
}

По умолчанию базовые запросы EQL возвращают 10 самых последних соответствующих событий в свойстве hits.events. Эти совпадения сортируются по временной метке, преобразованной в миллисекунды с момента эпохи Unix, в порядке возрастания.

{
  "is_partial": false,
  "is_running": false,
  "took": 60,
  "timed_out": false,
  "hits": {
    "total": {
      "value": 2,
      "relation": "eq"
    },
    "events": [
      {
        "_index": ".ds-my-data-stream-2099.12.07-000001",
        "_id": "OQmfCaduce8zoHT93o4H",
        "_source": {
          "@timestamp": "2099-12-07T11:07:09.000Z",
          "event": {
            "category": "process",
            "id": "aR3NWVOs",
            "sequence": 4
          },
          "process": {
            "pid": 2012,
            "name": "regsvr32.exe",
            "command_line": "regsvr32.exe  /s /u /i:https://...RegSvr32.sct scrobj.dll",
            "executable": "C:\\Windows\\System32\\regsvr32.exe"
          }
        }
      },
      {
        "_index": ".ds-my-data-stream-2099.12.07-000001",
        "_id": "xLkCaj4EujzdNSxfYLbO",
        "_source": {
          "@timestamp": "2099-12-07T11:07:10.000Z",
          "event": {
            "category": "process",
            "id": "GTSmSqgz0U",
            "sequence": 6,
            "type": "termination"
          },
          "process": {
            "pid": 2012,
            "name": "regsvr32.exe",
            "executable": "C:\\Windows\\System32\\regsvr32.exe"
          }
        }
      }
    ]
  }
}

Используйте параметр size, чтобы получить меньшее или большее количество совпадений:

GET /my-data-stream/_eql/search
{
  "query": """
    process where process.name == "regsvr32.exe"
  """,
  "size": 50
}

Используйте синтаксис последовательности EQL синтаксис для поиска серии упорядоченных событий. Перечислите элементы событий в восходящем хронологическом порядке, поместив самое последнее событие в конец списка:

GET /my-data-stream/_eql/search
{
  "query": """
    sequence
      [ process where process.name == "regsvr32.exe" ]
      [ file where stringContains(file.name, "scrobj.dll") ]
  """
}

Свойство ответа hits.sequences содержит 10 самых последних соответствующих последовательностей.

{
  ...
  "hits": {
    "total": ...,
    "sequences": [
      {
        "events": [
          {
            "_index": ".ds-my-data-stream-2099.12.07-000001",
            "_id": "OQmfCaduce8zoHT93o4H",
            "_source": {
              "@timestamp": "2099-12-07T11:07:09.000Z",
              "event": {
                "category": "process",
                "id": "aR3NWVOs",
                "sequence": 4
              },
              "process": {
                "pid": 2012,
                "name": "regsvr32.exe",
                "command_line": "regsvr32.exe  /s /u /i:https://...RegSvr32.sct scrobj.dll",
                "executable": "C:\\Windows\\System32\\regsvr32.exe"
              }
            }
          },
          {
            "_index": ".ds-my-data-stream-2099.12.07-000001",
            "_id": "yDwnGIJouOYGBzP0ZE9n",
            "_source": {
              "@timestamp": "2099-12-07T11:07:10.000Z",
              "event": {
                "category": "file",
                "id": "tZ1NWVOs",
                "sequence": 5
              },
              "process": {
                "pid": 2012,
                "name": "regsvr32.exe",
                "executable": "C:\\Windows\\System32\\regsvr32.exe"
              },
              "file": {
                "path": "C:\\Windows\\System32\\scrobj.dll",
                "name": "scrobj.dll"
              }
            }
          }
        ]
      }
    ]
  }
}

Используйте with maxspan для ограничения соответствующих последовательностей временным интервалом:

GET /my-data-stream/_eql/search
{
  "query": """
    sequence with maxspan=1h
      [ process where process.name == "regsvr32.exe" ]
      [ file where stringContains(file.name, "scrobj.dll") ]
  """
}

Используйте by для сопоставления событий, которые имеют одинаковые значения полей:

GET /my-data-stream/_eql/search
{
  "query": """
    sequence with maxspan=1h
      [ process where process.name == "regsvr32.exe" ] by process.pid
      [ file where stringContains(file.name, "scrobj.dll") ] by process.pid
  """
}

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

GET /my-data-stream/_eql/search
{
  "query": """
    sequence by process.pid with maxspan=1h
      [ process where process.name == "regsvr32.exe" ]
      [ file where stringContains(file.name, "scrobj.dll") ]
  """
}

Свойство hits.sequences.join_keys содержит общие значения полей.

{
  ...
  "hits": ...,
    "sequences": [
      {
        "join_keys": [
          2012
        ],
        "events": ...
      }
    ]
  }
}

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

GET /my-data-stream/_eql/search
{
  "query": """
    sequence by process.pid with maxspan=1h
      [ process where process.name == "regsvr32.exe" ]
      [ file where stringContains(file.name, "scrobj.dll") ]
    until [ process where event.type == "termination" ]
  """
}

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

Вы можете использовать параметр запроса filter_path для фильтрации ответа API. Например, следующий запрос возвращает только временную метку и PID из _source каждого соответствующего события.

GET /my-data-stream/_eql/search?filter_path=hits.events._source.@timestamp,hits.events._source.process.pid
{
  "query": """
    process where process.name == "regsvr32.exe"
  """
}

API возвращает следующий ответ.

{
  "hits": {
    "events": [
      {
        "_source": {
          "@timestamp": "2099-12-07T11:07:09.000Z",
          "process": {
            "pid": 2012
          }
        }
      },
      {
        "_source": {
          "@timestamp": "2099-12-07T11:07:10.000Z",
          "process": {
            "pid": 2012
          }
        }
      }
    ]
  }
}

Вы также можете использовать параметр fields для получения и форматирования определённых полей в ответе. Это поле идентично параметру API поиска fields.

Поскольку он обращается к отображениям индекса, параметр fields предоставляет несколько преимуществ по сравнению с прямым использованием _source. В частности, параметр fields:

Следующий запрос использует параметр fields для получения значений поля event.type, всех полей, начинающихся с process., и поля @timestamp. Запрос также использует параметр filter_path для исключения поля _source каждого результата.

GET /my-data-stream/_eql/search?filter_path=-hits.events._source
{
  "query": """
    process where process.name == "regsvr32.exe"
  """,
  "fields": [
    "event.type",
    "process.*",                
    {
      "field": "@timestamp",
      "format": "epoch_millis"  
    }
  ]
}

Ответ включает значения в виде плоского списка в разделе fields для каждого результата.

{
  ...
  "hits": {
    "total": ...,
    "events": [
      {
        "_index": ".ds-my-data-stream-2099.12.07-000001",
        "_id": "OQmfCaduce8zoHT93o4H",
        "fields": {
          "process.name": [
            "regsvr32.exe"
          ],
          "process.name.keyword": [
            "regsvr32.exe"
          ],
          "@timestamp": [
            "4100324829000"
          ],
          "process.command_line": [
            "regsvr32.exe  /s /u /i:https://...RegSvr32.sct scrobj.dll"
          ],
          "process.command_line.keyword": [
            "regsvr32.exe  /s /u /i:https://...RegSvr32.sct scrobj.dll"
          ],
          "process.executable.keyword": [
            "C:\\Windows\\System32\\regsvr32.exe"
          ],
          "process.pid": [
            2012
          ],
          "process.executable": [
            "C:\\Windows\\System32\\regsvr32.exe"
          ]
        }
      },
      ....
    ]
  }
}

Используйте параметр runtime_mappings для извлечения и создания полей во время выполнения во время поиска. Используйте параметр fields для включения полей во время выполнения в ответ.

Следующий запрос создаёт поле во время выполнения day_of_week из полей @timestamp и возвращает его в ответе.

GET /my-data-stream/_eql/search?filter_path=-hits.events._source
{
  "runtime_mappings": {
    "day_of_week": {
      "type": "keyword",
      "script": "emit(doc['@timestamp'].value.dayOfWeekEnum.toString())"
    }
  },
  "query": """
    process where process.name == "regsvr32.exe"
  """,
  "fields": [
    "@timestamp",
    "day_of_week"
  ]
}

API возвращает:

{
  ...
  "hits": {
    "total": ...,
    "events": [
      {
        "_index": ".ds-my-data-stream-2099.12.07-000001",
        "_id": "OQmfCaduce8zoHT93o4H",
        "fields": {
          "@timestamp": [
            "2099-12-07T11:07:09.000Z"
          ],
          "day_of_week": [
            "MONDAY"
          ]
        }
      },
      ....
    ]
  }
}

API поиска EQL по умолчанию использует поля @timestamp и event.category из ECS. Чтобы указать другие поля, используйте параметры timestamp_field и event_category_field:

GET /my-data-stream/_eql/search
{
  "timestamp_field": "file.accessed",
  "event_category_field": "file.type",
  "query": """
    file where (file.size > 1 and file.type == "file")
  """
}

Поле категории событий должно быть отображено как поле типа keyword. Поле временной метки должно быть отображено как поле типа date. date_nanos поля временных меток не поддерживаются. Вы не можете использовать поле nested или подполя поля nested в качестве поля временной метки или категории события.

По умолчанию API поиска EQL возвращает соответствующие совпадения по временной метке. Если два или более события имеют одинаковую временную метку, Elasticsearch использует значение поля разрыва сортировки для сортировки событий в порядке возрастания. Elasticsearch помещает события без значения разрыва сортировки после событий со значением.

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

Чтобы указать поле разрыва сортировки, используйте параметр tiebreaker_field. Если вы используете ECS, мы рекомендуем использовать event.sequence в качестве поля разрыва сортировки.

GET /my-data-stream/_eql/search
{
  "tiebreaker_field": "event.sequence",
  "query": """
    process where process.name == "cmd.exe" and stringContains(process.executable, "System32")
  """
}

Параметр filter использует Query DSL для ограничения документов, к которым применяется запрос EQL.

GET /my-data-stream/_eql/search
{
  "filter": {
    "range": {
      "@timestamp": {
        "gte": "now-1d/d",
        "lt": "now/d"
      }
    }
  },
  "query": """
    file where (file.type == "file" and file.name == "cmd.exe")
  """
}

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

Чтобы избежать длительных ожиданий, выполните асинхронный поиск EQL. Установите wait_for_completion_timeout на желаемую продолжительность ожидания синхронных результатов.

GET /my-data-stream/_eql/search
{
  "wait_for_completion_timeout": "2s",
  "query": """
    process where process.name == "cmd.exe"
  """
}

Если запрос не завершится в течение заданного интервала времени, поиск становится асинхронным и возвращает ответ, который включает:

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

{
  "id": "FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=",
  "is_partial": true,
  "is_running": true,
  "took": 2000,
  "timed_out": false,
  "hits": ...
}

Для проверки прогресса асинхронного поиска используйте API для получения результатов асинхронного поиска EQL с идентификатором поиска. Укажите, как долго вы хотите ждать полных результатов в параметре wait_for_completion_timeout.

GET /_eql/search/FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=?wait_for_completion_timeout=2s

Если значение is_running ответа равно false, асинхронный поиск завершен. Если значение is_partial равно false, возвращенные результаты поиска завершены.

{
  "id": "FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=",
  "is_partial": false,
  "is_running": false,
  "took": 2000,
  "timed_out": false,
  "hits": ...
}

Другой, более лёгкий способ проверки прогресса асинхронного поиска — использование API для получения статуса асинхронного поиска EQL с идентификатором поиска.

GET /_eql/search/status/FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=

{
  "id": "FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=",
  "is_running": false,
  "is_partial": false,
  "expiration_time_in_millis": 1611690295000,
  "completion_status": 200
}

По умолчанию API поиска EQL хранит асинхронные запросы в течение пяти дней. По истечении этого периода все запросы и их результаты удаляются. Используйте параметр keep_alive для изменения этого периода хранения:

GET /my-data-stream/_eql/search
{
  "keep_alive": "2d",
  "wait_for_completion_timeout": "2s",
  "query": """
    process where process.name == "cmd.exe"
  """
}

Вы можете использовать параметр keep_alive API для получения результатов асинхронного поиска EQL для изменения периода хранения впоследствии. Новый период хранения начнёт действовать после выполнения запроса get.

GET /_eql/search/FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=?keep_alive=5d

Используйте API для удаления асинхронного поиска EQL для ручного удаления асинхронного поиска EQL до окончания периода keep_alive. Если поиск всё ещё выполняется, Elasticsearch отменяет запрос поиска.

DELETE /_eql/search/FmNJRUZ1YWZCU3dHY1BIOUhaenVSRkEaaXFlZ3h4c1RTWFNocDdnY2FSaERnUTozNDE=

По умолчанию API поиска EQL сохраняет только асинхронные запросы. Чтобы сохранить синхронный запрос, установите keep_on_completion в значение true:

GET /my-data-stream/_eql/search
{
  "keep_on_completion": true,
  "wait_for_completion_timeout": "2s",
  "query": """
    process where process.name == "cmd.exe"
  """
}

Ответ включает идентификатор поиска. is_partial и is_running равны false, указывая, что запрос EQL был синхронным и вернул полные результаты.

{
  "id": "FjlmbndxNmJjU0RPdExBTGg0elNOOEEaQk9xSjJBQzBRMldZa1VVQ2pPa01YUToxMDY=",
  "is_partial": false,
  "is_running": false,
  "took": 52,
  "timed_out": false,
  "hits": ...
}

Используйте API для получения результатов асинхронного поиска EQL для получения тех же результатов позже:

GET /_eql/search/FjlmbndxNmJjU0RPdExBTGg0elNOOEEaQk9xSjJBQzBRMldZa1VVQ2pPa01YUToxMDY=

Сохраненные синхронные запросы всё ещё подчиняются периоду хранения параметра keep_alive. По истечении этого периода запрос и его результаты удаляются.

Вы также можете проверить только статус сохранённого синхронного запроса без результатов, используя API для получения статуса асинхронного поиска EQL.

Вы также можете вручную удалить сохранённые синхронные запросы, используя API для удаления асинхронного поиска EQL.

API поиска EQL поддерживает поиск по нескольким кластерам. Однако локальный и удаленные кластеры должны использовать одну и ту же версию Elasticsearch. Локальные кластеры версии 7.17.7 и выше также поддерживают поиск по нескольким кластерам для удаленных кластеров версии 7.15.0 и выше.

Следующий запрос обновления настроек кластера добавляет два удалённых кластера: cluster_one и cluster_two.

PUT /_cluster/settings
{
  "persistent": {
    "cluster": {
      "remote": {
        "cluster_one": {
          "seeds": [
            "127.0.0.1:9300"
          ]
        },
        "cluster_two": {
          "seeds": [
            "127.0.0.1:9301"
          ]
        }
      }
    }
  }
}

Для указания потока данных или индекса на удалённом кластере используйте синтаксис <cluster>:<target>.

GET /cluster_one:my-data-stream,cluster_two:my-data-stream/_eql/search
{
  "query": """
    process where process.name == "regsvr32.exe"
  """
}

При выполнении запроса последовательности узлу, обрабатывающему запрос, необходимо хранить некоторые структуры в памяти, необходимые алгоритму, реализующему сопоставление последовательностей. При обработке больших объёмов данных и/или при запросе большого количества сопоставленных последовательностей пользователем (с помощью параметра запроса size) занимаемая этими структурами память может превысить доступную память JVM. Это приведёт к исключению OutOfMemory, которое приведёт к сбою узла.

Чтобы предотвратить это, используется специальный разрыв цепи, который ограничивает выделение памяти во время выполнения запроса последовательности. При срабатывании разрыва цепи возникает исключение org.elasticsearch.common.breaker.CircuitBreakingException и пользователю возвращается сообщение об ошибке.

Этот разрыв цепи можно настроить с помощью следующих параметров: