script_mc

Описание

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

предупреждение

Для работы источника script_mc необходим настроенный SM Remote Executor (SME-RE) и сервер Memcached. Оптимально устанавливать сервер Memcached с SM Remote Executor (либо промежуточный), а не на сервер Smart Monitor.

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

Синтаксис

...| script_mc "<path_to_script_engine> <path_to_script>"

---

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

Параметр	Синтаксис	Описание
path_to_script_engine	<string>	Путь до исполняемого файла.
path_to_script	<string>	Путь до скрипта.

---

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

к сведению

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

"sme" : {
    "core" : {
        "memcached": {
            "port": "11211",
            "ip": "127.0.0.1"
        }
    }
}

---

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

Пример 1

В данном примере script_mc запускает скрипт через явное указание интерпретатора и пути к файлу в одной строке.

| script_mc "python /home/user/dev/test-script/test-script.py"

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

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

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

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

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

Пример 1

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

source temperature_sensors
| script_mc "python /home/user/test-script/test-script.py sensor_name location error_description"

Код 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_mc выполняется выборка данных с помощью table. Далее осуществляется получение уникальных значений для переданных полей, после чего выполняется преобразование поля unique_location с помощью функции eval, и результат выводится через table.

source temperature_sensors
| table  sensor_name,location,error_description
| script "bash /home/user/test-script/aggregation_value.sh sensor_name location error_description"
| eval unique_location = upper(unique_location)
| 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"