script

Описание

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

Обратите внимание!

Для работы источника script необходим настроенный SM Remote Executor(SME-RE).

Статья по настройке SME-RE

Синтаксис

...| script [intr=<script_interpreter>] <script_file_name> [<timeout>]

---

Обязательные аргументы

Параметр	Синтаксис	Описание
script_file_name	<string>	Название файла со скриптом.

---

Опциональные аргументы

Параметр	Синтаксис	По умолчанию	Описание
script_interpreter	<string>	Берется из sme.core.remote_script.base_interpreter_name.	Название интерпретатора.
timeout	<int>	Берется из sme.core.remote_script.timeout.	Таймаут выполнения запроса в миллисекундах.

---

Конфигурация источника

к сведению

Все настройки источника хранятся в _cluster/settings.

Пример настроек:

"sme" : {
    "core" : {
        "remote_script" : {
            "base_path" : "/home/test_user/test/",
            "port" : "18080",
            "base_interpreter_name" : "python3",
            "url" : "http://localhost",
            "timeout" : "60000",
            "interpreters" : [
            "python3::/usr/bin/python3",
            "bash::/usr/bin/bash"
            ]
        }
    }
}

---

Описание параметров конфигурации

• url — url SM Script Proxy коннектора
• port — порт SM Script Proxy коннектора
• interpreters - названия интерпретаторов и пути к их исполняемым файлам
• base_interpreter_name - название интерпретатора по умолчанию
• base_path - путь до директории, в которой лежат скрипты
• timeout- таймаут запроса, указывается в миллисекундах

---

Примеры запросов

Данные примеры показывают запуск скрипта без явного интерпретатора, с указанным intr и с настройкой таймаута.

Пример 1

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

| script test-script.py

Пример 2

В данном примере интерпретатор передается явно через intr=python3.

source tweets
| script intr=python3 test-script.py

Пример 3

В данном примере дополнительно задан timeout, чтобы ограничить время выполнения скрипта.

source tweets
| script intr=python3 test-script.py timeout=5000

Примеры запросов с передачей аргументов

Для корректного выполнения запроса вне зависимости от места использования команды script (в начале, в середине или в конце запроса), необходимо учитывать следующие особенности:

1. Подготовка данных для скрипта: перед вызовом скрипта формируется результат запроса, который будет передан на обработку. Передача данных не требует обязательного ограничения выборки по необходимым полям, однако для повышения производительности можно сократить объём данных с помощью fields, table или агрегирующих функций по требуемым полям. Результат запроса в виде JSON передаётся в скрипт через временный файл, путь к которому автоматически указывается как последний неявный аргумент

2. Получение данных скриптом: скрипт должен поддерживать прием аргументов, а также дальнейшую работу с ними. События могут быть как результат функций агрегации, так и просто результат поискового запроса. Главное условие для успешной передачи - наличие указанных полей в результате запроса. Для передачи аргументов необходимо перечислить список полей в строке вызова команды script. Входные данные извлекаются из блока resp.body.hits.hits временного файла, где они представлены в виде массива событий (JSON-объектов). Каждый элемент массива соответствует отдельному событию, с которым далее можно работать средствами используемого интерпретатора

3. Отправка данных из скрипта: Для отправки одного события достаточно вывести JSON-результат в stdout. Чтобы вывести несколько событий, необходимо сформировать payload и аналогичным образом отправить его в stdout

Пример 1

Рассмотрим следующий пример вызова transform_data.py. В качестве аргументов данному скрипту передаются необходимые поля из результата запроса, а для полей error_description и connect_status выполняется преобразование из числовых значений в текстовые. После данные передаются в SME и отображаются в поиске. Данный пример также демонстрирует, как сформировать данные для вывода нескольких событий.

Код python скрипта

    import sys
    import json

    def main():

        # Чтение аргументов и проверка их количества
        args = sys.argv[1:]
        if len(args) != 6:
            raise Exception(f"Expected 5 fields + payload file, got {len(args)-1}")

        # Последний аргумент - путь до временного файла
        timestamp_field, sensor_name_field, location_field, error_description_field, connect_status_field, payload_filepath = args


        # Загрузка JSON payload из временного файла SME
        with open(payload_filepath) as f:
            payload = json.load(f)

        # Извлечение списка hits - блока с данными
        hits = payload.get("resp", {}).get("body", payload).get("hits", {}).get("hits", [])
        result_hits = []


        # Обработка каждого документа из hits
        for i, hit in enumerate(hits, 1):
            source = hit.get("_source", {})

            error_description = source.get(error_description_field)
            connect_status = source.get(connect_status_field)

            # Преобразование error_description в человекочитаемый формат
            error_description = (
                "No data" if error_description == -1 else
                "No error" if 1 <= error_description <= 3 else
                "Medium error" if 4 <= error_description <= 7 else
                "Critical error" if 8 <= error_description <= 10 else
                "No data"
            )

            # Преобразование connect_status в человекочитаемый формат
            connect_status = (
                "Connected" if connect_status == 1 else
                "Not connected" if connect_status == 0 else
                "No data"
            )

            # Формирование итогового документа результата
            result_hits.append({
                "_index": hit.get("_index", "custom"),
                "_type": hit.get("_type", "custom"),
                "_id": hit.get("_id", str(i)),
                "_score": hit.get("_score", 1),
                "_source": {
                    "@timestamp": source.get(timestamp_field, "No data"),
                    "sensor_name": source.get(sensor_name_field, "No data"),
                    "location": source.get(location_field, "No data"),
                    "error_description": error_description,
                    "connect_status": connect_status,
                }
            })


        # Формирование payload для отправки
        result = {
            "hits": {
                "total": {
                    "value": len(result_hits),
                    "relation": "eq"
                },
                "max_score": 1,
                "hits": result_hits
            }
        }

        # Вывод payload в stdout
        print(json.dumps(result))

    if __name__ == "__main__":
        main()

Пример 2

Аналогичным образом осуществляется передача данных в bash скрипт. Следующий пример демонстрирует выполнение скрипта aggregation_value.sh, который выполняет отправку одного события. Перед использованием функции script выполняется сбор данных при помощи aggs. Далее осуществляет поиск уникальных значения для переданных полей, после чего выполняется преобразование для поля unique_location функцией eval и выводится результат через table.

Код bash скрипта

#!/bin/bash
# Получение временного файла результата запроса - последний неявный аргумент
payload_filepath="${!#}"

# Получение списка полей 
fields=("${@:1:$(($#-1))}")

# Чтение полного JSON из файла
payload=$(cat "$payload_filepath")

# Извлечение массива документов, который расположены в hits
hits=$(echo "$payload" | jq '.resp.body.hits.hits // .hits.hits // []')

# Обработка каждого переданного поля
jq_filter='{'

first=1

for field in "${fields[@]}"; do
  if [ $first -eq 1 ]; then
    first=0
  else
    jq_filter+=","
  fi

  jq_filter+="
    \"unique_$field\": (
      [ .[]?._source.\"$field\" ]
      | map(select(. != null))
      | unique
    )"
done

jq_filter+='
}'

# Вывод результата
echo "$hits" | jq "$jq_filter"