Сказ о некластерном тестировании сервисов

Эта статья является живым очерком о буднях современного тестировщика в agile-команде и рассматривает подход в тестировании, при котором описывается вариант решения проблемы настройки и разворачивания тестируемого приложения. В качестве примера возьмем сервис для поиска данных во временных рядах — Time Series Pattern, разрабатываемый нами на аутсорс. 

Меня зовут Роман Мостафин, и я работаю QA-инженером в компании Factory5. 
Начинаем. 

На данный момент многие команды сталкиваются с тем, что сложно обслуживать среду для тестирования в виде стендов, а также с многопользовательским доступом к одним и тем же данным и версионирование этих стендов. Порою эта проблема стоит довольно остро. Решением может быть поднятие сервисов и части систем на локальной машине. К его плюсам относится: легкость версионирования, быстрота разворачивания и настройки. Данное решение применяется в нашей компании для тестирования отдельных сервисов. И сейчас я расскажу, как мы его используем на примере нашего сервиса разрабатываемого на аутсорсе.  

Установка сервисов 

Для начала нам понадобится утилита Docker — система контейнеризации и виртуализации. 

Команда сделала сборку в виде докер-файла, в которую входят следующие сервисы: 

• Kafka 

• Kafdrop 

• Zookeeper 

• Kafka UI 

• Clickhouse 

• Flink-jobmanager 

• Flink-taskmanager 

• TSP 

Создадим директорию на локальной машине и перенесем туда docker-compose файл.

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

Пока поднимается данная сборка, расскажу про возможности самого сервиса. 

TSP (Time Series Patterns) — аналитический сервис для поиска шаблонов в данных временных рядов большого объема. Имеет гибкий DSL для описания паттернов (правил) и представляет собой оптимизированный движок потоковой обработки поверх Apache Flink. Этот сервис может применяться для потоковой обработки данных, а также для частичной обработки данных. 

Принцип работы такой: есть исходные данные, на них пишутся правила (определенные условия поиска данных). Далее эти правила применяются на данных с помощью запроса, в котором мы указываем адреса источников данных, диапазон данных и условие проверки. Как итог, мы получаем запись о том, что найдено в соответствии правилу на определенном отрезке данных. Для запуска потоковой обработки (stream) используется kafka в качестве источника данных и хранилища инцидентов. Для запуска обработки данных по частям (batch) в качестве хранилища данных используется база данных clickhouse. В этой статье мы рассмотрим оба варианта запуска сервиса. 

После того как будет доступен адрес http://localhost:8081/ на локальной машине, можно приступать к настройке. 

Настройка запуска в режиме batch 

Начнем с настройки режима с запуском правил на части данных. Для этого нам понадобится клиент базы данных. Я покажу на примере клиента DataGrip https://www.jetbrains.com/ru-ru/datagrip/. 

Создаем соединение со стандартными настройками clickhouse (username = default, host = localhost, database = default).

Открываем консоль и создаем две таблицы. 

Таблицу для исходных данных (телеметрии): 

create table if not exists test_telemetry   (   dt DateTime default toDateTime(ts),   ts Float64,   sensor_id String,   n UInt32,   value_float Float32,   equipment_id UInt32,   engine_no String   )   engine = ReplacingMergeTree()   PARTITION BY toYYYYMM(dt)   ORDER BY (equipment_id, sensor_id, n, ts)   SETTINGS index_granularity = 8192; 

И таблицу для хранения инцидентов: 

create table if not exists events_v2_event   (   uuid UUID default generateUUIDv4(),   date_from DateTime64(6) default '0000000000.000000',   date_to DateTime64(6) default '0000000000.000000',   processing_date DateTime64(6) default now(),   type_id UInt16,   equipment_id UInt32,   ph_node_id UInt32,   algorithm_id UInt32,   value Nullable(Float32),   context Nullable(String),   status UInt8 default 1,   duration Float64 materialized (toUnixTimestamp64Milli(date_to) - toUnixTimestamp64Milli(date_from)) / 1000   )   engine = MergeTree()   PARTITION BY (toYYYYMM(date_from), intDiv(equipment_id, 1000))   ORDER BY date_from   SETTINGS index_granularity = 8192; 

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

INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157169, 'test_sensor_2', 1, 10, 1, '1');   default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157170, 'test_sensor_2', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:23', 1635157171, 'test_sensor_2', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157169, 'test_sensor_1', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157170, 'test_sensor_1', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:23', 1635157171, 'test_sensor_1', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157169, 'test_sensor', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:21', 1635157170, 'test_sensor', 1, 10, 1, '1');   INSERT INTO default.test_telemetry (dt, ts, sensor_id, n, value_float, equipment_id, engine_no) VALUES ('2021-10-25 13:23:23', 1635157171, 'test_sensor', 1, 10, 1, '1'); 

Теперь мы проверим работоспособность нашей системы выполнив REST запрос в сам TSP. 

Его можно выполнить, используя клиент для Rest. Я использую POSTMAN 

Url: http://localhost:8080/streamJob/from-jdbc/to-jdbc/?run_async=1  

Body: 

{ "sink": {   "jdbcUrl": "jdbc:clickhouse://default:@clickhouse:8123/default",   "rowSchema": {   "toTsField": "date_to",   "fromTsField": "date_from",   "unitIdField": "equipment_id",   "appIdFieldVal": [   "type_id",   1   ],   "patternIdField": "algorithm_id",   "subunitIdField": "ph_node_id"   },   "tableName": "events_v2_event",   "driverName": "ru.yandex.clickhouse.ClickHouseDriver"   },   "uuid": "59338fc6-087a-4f49-a6f8-f0f96cc31bcd_1",   "source": {   "query": "SELECT ts, equipment_id AS equipment_id, sensor_id, value_float \nFROM test_telemetry \nWHERE equipment_id IN (1) AND ts >= 1635157169.0 AND ts < 1635157171.0 ORDER BY ts ASC",   "jdbcUrl": "jdbc:clickhouse://default:@clickhouse:8123/default",   "sourceId": 1,   "driverName": "ru.yandex.clickhouse.ClickHouseDriver",   "unitIdField": "equipment_id",   "datetimeField": "ts",   "partitionFields": [   "equipment_id"   ],   "dataTransformation": {   "type": "NarrowDataUnfolding",   "config": {   "keyColumn": "sensor_id",   "defaultTimeout": 60000,   "fieldsTimeoutsMs": {},   "defaultValueColumn": "value_float"   }   }   },   "patterns": [   {   "id": "1",   "payload": {   "subunit": "4"   },   "sourceCode": "\"test_sensor\" = 10 andThen (\"test_sensor_1\" = 20 or \"test_sensor_2\" = 10)",   "forwardedFields": [   "equipment_id"   ]   }   ]   } 

После получения успешного ответа мы переходим в ui-флинк и ждем смены статуса job с running на finished.

Далее мы переходим в базу данных, делаем запрос к таблице: select * from events_v2_event; и получаем строчку данных, которая соответствует условию:

Настройка запуска в режиме stream 

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

Для начала мы идем в kafka ui и создаем два топика: 

топик для исходных данных, в данном примере название “test” и топик для инцидентов — events_v2_event. 

Пример: 

Далее с помощью python скрипта и csv файла с тестовыми данными мы заполняем данными топик test. 

Для этого мы создаем файл test_telemetry.csv со следующими данными: 

"dt","ts","sensor_id","n","value_float","equipment_id","engine_no"  2021-10-25 13:23:21,1635157169,"test_sensor_2",1,10.0,1,"1"  2021-10-25 13:23:21,1635157170,"test_sensor_2",1,10.0, 1,"1"  2021-10-25 13:23:23,1635157171,"test_sensor_2",1,10.0,1,"1"  2021-10-25 13:23:21,1635157169,"test_sensor_1",1,10.0,1,"1"  2021-10-25 13:23:21,1635157170,"test_sensor_1",1,10.0,1,"1"  2021-10-25 13:23:23,1635157171,"test_sensor_1",1,10.0,1,"1"  2021-10-25 13:23:21,1635157169,"test_sensor",1,10.0,1,"1"  2021-10-25 13:23:21,1635157170,"test_sensor",1,10.0,1,"1"  2021-10-25 13:23:23,1635157171,"test_sensor",1,10.0,1,"1" 

Далее мы создаем python-file со следующим содержимым: 

import pandas as pd  from kafka import KafkaProducer  producer = KafkaProducer(bootstrap_servers='10.83.0.3:9092')    df = pd.read_csv('test_telemetry.csv')    df.apply(lambda x: producer.send(topic='test',value=x.to_json().encode('utf-8')), axis=1)   

Устанавливаем python >= 3.8, виртуальное окружение и библиотеку kafka-python. 

И запускаем сам скрипт для того, чтобы перенести эти данные в топик.  

 И теперь мы можем отправить запрос на запуск task со следующим body и url.

Url: http://localhost:8080/streamJob/from-kafka/to-kafka/?run_async=1 

Body:

{                           "sink": {                               "topic": "events_v2_event",                               "broker": "10.83.0.3:9092",                               "rowSchema": {                                   "toTsField": "date_to",                                   "fromTsField": "date_from",                                   "unitIdField": "equipment_id",                                   "appIdFieldVal": [                                       "type_id",                                       1                                   ],                                   "patternIdField": "algorithm_id",                                   "subunitIdField": "ph_node_id",                                   "incidentIdField": "uuid"                               },                               "parallelism": 1,                               "datetimeField": "ts",                               "eventsMaxGapMs": 60000,                               "defaultEventsGapMs": 2000,                               "numParallelSources": 1,                               "patternsParallelism": 1                           },                           "uuid": "test_status",                           "source": {                               "group": "local_1",                               "topic": "test",                               "brokers": "10.83.0.3:9092",                               "sourceId": 1,                               "fieldsTypes": {                                   "ts": "float64",                                   "sensor_id": "String",                                   "n": "int32",                                   "value_float": "float32",                                   "equipment_id": "int32",                                   "engine_no": "String"                               },                               "unitIdField": "equipment_id",                               "datetimeField": "ts",                               "partitionFields": [                                   "equipment_id"                               ],                               "dataTransformation": {                                   "type": "NarrowDataUnfolding",                                   "config": {                                       "keyColumn": "sensor_id",                                       "defaultTimeout": 60000,                                       "fieldsTimeoutsMs": {},                                       "defaultValueColumn": "value_float",                                       "valueColumnMapping": {}                                   }                               }                           },                           "patterns": [                               {                                   "id": "1",                                   "subunit": 2,                                   "sourceCode": "\"test_sensor\" = 10 andThen (\"test_sensor_1\" = 20 or \"test_sensor_2\" = 10)",                                   "forwardedFields": [                                       "equipment_id"                                  ]                               }                           ]                       } 

Подождав некоторое время, мы можем перейти в топик с инцидентами и обнаружим там запись об инциденте:

Итоги: 

На примере настройки и установки сервиса через docker мы рассмотрели решение проблемы тестирования приложений на локальной машине и убедились, что это возможно при наличии определенных знаний по системам виртуализации. Конечно, тестирование в реальной среде (в кластере) имеет свои особенности, но большую часть функционала можно проверить именно на локальной машине. Я намеренно упустил шаги по созданию самой конфигурации, так как на просторах интернета есть множество статей как это сделать.  

Всем спасибо за внимание!