Микросервисный фреймворк Flogo — собираем систему из кирпичиков

Микросервисная архитектура, предполагающая разделение информационной системы на небольшие самостоятельные части, взаимодействующие посредством сетевых протоколов (чаще всего HTTP с пересылкой JSON-сообщений или через gRPC) имеет особое значение для создания масштабируемых отказоустойчивых приложений. Для реализации микросервисов существует множество библиотек и фреймворков, созданных для всех актуальных технологий разработки (исчерпывающий обновляемый список можно посмотреть например здесь).

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

Но особый интерес для разработки микросервисов представляют фреймворки, формирующие среду выполнения (и иногда и инструменты конфигурирования) для быстрого развертывания кода, основанного на модели реакции на внешние и внутренние события. Например, среди известных фреймворков можно назвать Akka (для запуска акторов на JVM), Oracle Helidon (предоставляет большой набор вспомогательных сервисов для мониторинга, контроля доступа и поддержки различных протоколов) и Vert.X (фреймворк для создания реактивных приложений на JVM). Перечисленные фреймворки ориентированы на использование кода, создаваемого на основе JVM-совместимого языка программирования, но аналогичные решения есть и для других технологий разработки. Например, для создания микросервисной архитектуры на Go, может использоваться фреймворк Flogo, основанный на идеях потока сообщений/данных между микросервисами и реакции на события (официальный сайт). В этой статье мы рассмотрим его возможности на примере простой задачи обработки данных телеметрии.

Сам фреймворк представляет из себя набор инструментов командной строки, веб-конструктор для создания action и набор библиотек, которые могут быть подключены к коду на Go для взаимодействия с возможностями фреймворка. Кроме того, что фреймворк обеспечивает механизм последовательных операций (некоторого конвейера), основанный на триггерах, а также в нем “из коробки” поддерживается использование предварительно обученных нейронных сетей на основе TensorFlow для принятия решений о выборе последовательности операций. 

Для конфигурирования фреймворка можно установить веб-интерфейс

docker run -itd -p 3303:3303 flogo/flogo-docker eula-accept

или для запуска наиболее актуальной (нестабильной) версии с поддержкой машинного обучения:

docker run -itd -p 3303:3303 flogo/flogo-docker:unstable-ml eula-accept

Также возможно установить набор инструментов командной строки

go get -u github.com/project-flogo/cli/…

Центральным объектом конфигурирования в Flogo является приложение (Application), оно включает в себя зарегистрированные действия (Actions) и связанные с ними триггеры для запуска действия (Triggers). Начнем с создания нового приложения, для этого выполним переход на страницу http://localhost:3303/ и выберем “New” (создание приложения)

Action может работать с потоком данных (Stream) и формировать на основе набора входных данных (с использованием механизма map-filter-reduce) новый поток данных через канал, который может стать источником событий для другого Action. Второй вид Action — поток выполнения (flow), который может рассматриваться как высокоуровневое описание координации микросервисов со связыванием входным и выходных данных к контексту Action.

Action состоит из последовательности активностей (Activity), которые могут быть установлен из git-репозиториев определенной структуры, которая будет описана ниже.

Определим основные сущности Flogo:

• модели (models) — описывают структуру входных и выходных параметров activity

• триггеры (triggers) — определяют условия запуска Flow

• активности (activities) — определяют реализацию шага бизнес-логики

Большое количество готовых Activity можно найти на странице https://tibcosoftware.github.io/flogo/showcases/ и https://github.com/project-flogo/contrib. Аналогично и models и triggers также могут быть установлены из внешнего источника через веб-интерфейс (Install Contribution) или через командную строку flogo install <url>. Среди встроенных Activity можно выделить следующие:

• Start a subflow — запустить вспомогательный flow при достижении этой activity

• Run an Action — запустить другой Action (flow/stream)

• Log — добавить сообщение в журнал

• Mapper / Filter / Aggregate — реализация обработки данных из входного потока (используется в action типа stream)

• Throw Error — вернуть ошибку

• Return — вернуть результат

• SQL Database Activity — действия по манипуляции данными

• REST Invoke — запустить внешний REST-сервис

Для запуска Activity может использовать один из встроенных триггеров или подключаемый из внешнего репозитория. Встроенные триггеры могут вызываться из командной строки (CLI Trigger), при получении HTTP-запроса (Receive HTTP Message), данных через TCP/UDP, а также по таймеру. Также можно обмениваться данными с функциями AWS Lambda.

Триггер связан с последовательностью активностей через input/output аргументы (input передаются из триггера в направлении первой активности, output отправляются в триггер и могут, например, выступить в роли кода ответа или содержания сообщения для HTTP-запроса). 

Activity может получать входящие данные с предыдущего шага (будут доступны через селектор $.name), а также создавать выходные данные для следующего шага. Активность Return создает результат выполнения последовательности активностей (в flow создается единственное значение результата, в stream может быть создан поток из результатов).

Для разработки триггеров, активностей и действий могут использоваться базовые структуры из github.com/project-flogo/core. Для описания взаимодействия с контекстом выполнения определяется список входных и выходных аргументов, а также настройки. Для описания полей могут использоваться как примитивные типы данных (string, int, bool), так и составные структуры и массивы (полный список типов приведен здесь). Точкой входа является функция init(), которая должна зарегистрировать структуру с описанием Activity/Action/Trigger. Также репозиторий должен содержать файл descriptor.json с описанием компонента. Метаданные описывают следующие атрибуты:

• name — название компонента

• type — тип компонента (flogo:activity, flogo:action, flogo:trigger)

• version — версия компонента

• title — название для списка компонентов

• description — описание компонента

• display.icon — пиктограмма для списка

• input — схема входных данных (например, конфигурации или входного потока) для activity

• output — схема выходных данных

• handler — настройки для trigger

Мы попробуем разработать Flow для сохранения истории изменения данных с внешнего датчика температуры. Кроме этого мы будем использовать значения температуры для управления внешним исполнителем (кондиционером), который будет реализован в виде подключаемого драйвера на Go, который мы преобразуем в Activity.

Рассмотрим создание следующего сценария обработки потока входных данных с измерениями температуры:

• данные поступают через HTTP-запросы;

• производится усреднение окна из 5 последовательных замеров для сглаживания резких выбросов температуры;

• усредненные замеры отправляются в InfluxDB;

• выполняется вызов активности для управления кондиционером.

Первая часть обработки реализуется в модели поточной обработки данных. Создаем новое действие типа Stream, назовем его Conditioner Control Flow. Для активации потока добавим триггер для запуска (знак + слева от последовательности активностей) с использованием REST с типом Receive HTTP Message. Перейдем в конфигурацию триггера и определим порт, метод и путь публикации REST-точки подключения.

Для передачи полученных данных добавим в Stream Interface в Inputs именованное поле message и свяжем его в триггере с параметрами http-запроса.

Дальше добавим обработку входных данных и выполним усреднение, для этого добавим активность Aggregate и выполним ее настройку. Будет необходимо указать функцию агрегации (“avg”), режим определения окна (windowType = sliding), размер окна (количество записей для усреднения) windowSize=5. Далее будет необходимо указать способ преобразования входных данных в данные для агрегации. В нашем случае будет необходимо сделать преобразование в число с плавающей точкой coerce.toFloat64($.message). Flogo представляет большой набор функций для работы со строками, датами, генерации случайных чисел и извлечения данных из JSON, при этом все переменные в текущем контексте доступны через служебный объект $. Последним действием Stream должна быть передача вычисленного значения в канал (Channel), для которого необходимо сконфигурировать входной параметр name (например, “smoothed”) и значение (в данном случае оно будет $.result)

Для обработки усредненных значений температуры необходимо создать новый Action с типом Flow, создать входную переменную value для Action, задать стартовый триггер типа Listen to Channel (конфигурация: название — smoothed, Map to flow inputs — связать переменную value c $.data). 

Для отправки данных в InfluxDB необходимо установить дополнительный компонент github.com/shaliniGovindaNayak/flogo-workspace/activity/influxdb. После добавления компонента необходимо выполнить конфигурацию активности (адрес, порт и данные аутентификации к серверу InfluxDB) и указать название переменной с источником данных (» class=»formula inline»>.value). 

Но наибольший интерес представляет создание активности из существующего кода микросервиса. Для этого мы будем использовать возможности базовой библиотеки Flogo. Начнем с создания файла описания активности.

{   "name": "actuator",   "type": "flogo:activity",   "version": "0.0.1",   "title": "Actuator",   "description": "Conditioner Actuator",   "homepage": "https://github.com/dzolotov/actuator",   "settings": [],   ],   "input": [     {       "name": "temperature",       "type": "integer",       "required": true     },     {       "name": "state",       "type": "boolean",       "required": true     }   ],   "output": [     {       "name": "result",       "type": "string",       "required": true     }   ] }

Для подключения ядра и дополнительных функций Flogo подключим дополнительные модули. Сначала создадим описание метаданных (metadata.go).

package actuator  import (     "github.com/project-flogo/core/data/coerce"     "strconv" )  type Settings struct { }  type Input struct { Temperature uint64 `md:"temperature,required"` State bool `md:"state,required"` }  func (r *Input) FromMap(values map[string]interface{}) error { temperature, _ := coerce.ToInt64(values["temperature"])             state, _ := coerce.ToBool(values[“state”]) r.Temperature = temperature             r.State = state return nil }  func (r *Input) ToMap() map[string]interface{} { return map[string]interface{}{ "temperature": strconv.Itoa(r.Temperature),                        "state": strconv.FormatBool(r.State) } }  type Output struct { Result string `md:"result"` }  func (o *Output) FromMap(values map[string]interface{}) error { strVal, _ := coerce.ToString(values["result"]) o.Result = strVal return nil }  func (o *Output) ToMap() map[string]interface{} { return map[string]interface{}{ "result": o.Result, } }

и бизнес-логику (activity.go)

import ( "github.com/project-flogo/core/activity" )  //Точкой входа в activity является функция init() func init() { _ = activity.Register(&ActuatorActivity{}) }  //определение метаданные (структура входных и выходных параметров и настроек) var activityMd = activity.ToMetadata(&Settings{}, &Input{}, &Output{})  type ActuatorActivity struct { }  //получить метаданные func (a *ActuatorActivity) Metadata() *activity.Metadata { return activityMd }  //логика активности func (a *ActuatorActivity) Eval(ctx activity.Context) (done bool, err error) { //получение входных данных input := &Input{} err = ctx.GetInputObject(input) if err != nil { return true, err } temperature := input.Temperature state := input.State //реализация бизнес-логики //… //возврат значения output := &Output{Result: temperature} err = ctx.SetOutputObject(output) if err != nil { return true, err } return true, nil }

Для установки компонента загрузим репозиторий на github и выполним установку дополнительного компонента с адресом репозитория. После этого в списке активностей появится новая активность Actuator (после добавления нужно будет выполнить связывание или определение констант для входных значений и связывание выходных значений).

Итоговая конфигурация может быть выгружена в Web UI и экспортирована в JSON (Export -> App), либо собрана в выполняемый образ для целевой платформы (Linux/Mac/Windows в WebUI, а также любая поддерживаемая целевая платформа при сборке через CLI. 

Сегодня вечером в OTUS пройдет demo-урок «Тестирование в микросервисной архитектуре». На занятии расскажем про различные типы тестов и инструментов, используемых в тестировании, а также поговорим о том, как микросервисная архитектура изменила подходы к тестированию. Регистрация доступна для всех желающих по ссылке.