Управління потоками в Node-RED¶

Мета: Опанування навичок ефективного управління потоками даних у середовищі Node-RED, зокрема освоєння принципів керування потоками, використання вбудованих та користувацьких вузлів для забезпечення оптимального оброблення інформації та забезпечення швидкої реакції на події в системі.

Виконавець:
студент групи КН-41
Кривобоков Микита Олександрович

Завдання¶

Додати фільтрацію повідомлень для потоку повідомлень. Фільтрувати за критерієм location_uid = 11 (Закарпатська область).
Відсортувати вхідні повідомлення за полем updated_at у порядку зростання.
Додати функціонал обчислення тривалості тривоги для кожного повідомлення.
Додати функціонал загальної аналітики (в рамках серії повідомлень) за критеріями: загальна тривалість тривог (total duration), середня тривалість тривоги (average duration), мінімальна та максимальна тривалості тривоги (min duration, max duration).
Винести створений потік як підпотік. Підпотік повинен мати 2 порти: перший повертатиме послідовність повідомлень, другий — об'єкт з аналітикою.
Сформувати шаблони для виведення інформації за прикладом попередньої практичної роботи (можливе повторне використання).
Попрацювати над оформленням потоків та обробкою помилок, за потреби згрупувати функціональні вузли, які працюють у рамках спільних цілей.
Скласти звіт про виконану роботу, у якому необхідно перелічити використані вузли та їх застосування.

Хід роботи¶

Компонент нормалізації¶

Сортування повідомлень відбувається лише для даних, що є масивом, і були початково надані для Джерела 2 (P3).

Наразі всі повідомлення проходять з обох джерел, як у статистику, так і у вивід як повідомлення (пізніше буде виправлено з поділом на тип джерела).

Рефактор

Код попередніх компонентів був модифікований, що включає форматер часу та обчислення тривалості.

Форматувальник часуОбчислення тривалостіМапер дати закінчення

function convert(item) {
    return {
        ...item,
        started_time: item.started_at
            ? new Date(item.started_at).toLocaleTimeString()
            : null,

        finished_time: item.isActive || !item.finished_at
            ? null
            : new Date(item.finished_at).toLocaleTimeString()
    };
}

if (Array.isArray(msg.payload)) {
    msg.payload = msg.payload.map(convert);
} else if (msg.payload && typeof msg.payload === "object") {
    msg.payload = convert(msg.payload);
}

return msg;

function calculate(item) {
    if (item.isActive || !item.started_at || !item.finished_at) {
        return {
            ...item,
            duration: null
        };
    }

    return {
        ...item,
        duration: Math.ceil(
            (new Date(item.finished_at).getTime() - new Date(item.started_at).getTime()) / 1000
        )
    };
}

if (Array.isArray(msg.payload)) {
    msg.payload = msg.payload.map(calculate);
} else if (msg.payload && typeof msg.payload === "object") {
    msg.payload = calculate(msg.payload);
}

return msg;

function enrich(item) {
    const { finished_at } = item;

    return {
        ...item,
        isActive: !finished_at
    };
}

if (Array.isArray(msg.payload)) {
    msg.payload = msg.payload.map(enrich);
}
else if (typeof msg.payload === "object" && msg.payload !== null) {
    msg.payload = enrich(msg.payload);
}

return msg;

Серед нових компонентів було використано:

split: виконує перетворення вхідного масиву у потік значень;

join: виконує злиття потоку повідомлень у масив;

sort: сортує елементи потоку.

Компонент потоку повідомлень¶

Подачу нових повідомлень було вирішено додати як нове 'джерело', з приведенням усіх повідомлень до єдиного потоку (розбиття на окремі повідомлення).

Джерело 1: використовується з попередньої практичної для генерації повідомлень (5-15с);
Джерело 2: видає масив повідомлень у кількості 107 одиниць.

Рефактор виводу повідомлень¶

Рефактор цього компонента передбачав винесення перевірки, чи активна тривога, і темплейту з виводом у окремий підпотік:

Також виправлено шаблон для виводу, додаючи період часу і відмітку, що це архівні повідомлення.

Новий компонент статистики¶

Згідно з пунктами завдання, необхідно було створити компонент, що збиратиме статистичні дані на основі тривалості (duration), у форматі загальна тривалість, середня, мінімальна, максимальна.

Кінцевим результатом стає об'єкт після злиття полів, що містить статистичні дані:

{
  totalDuration: "8 hours",
  avgDuration: "2 hours",
  minDuration: "a few seconds",
  maxDuration: "2 hours"
}

Обчислення статистики здійснюється шляхом накопичення буфера на 5 елементів (або надсилання кожні 10с), таким чином статистика накопичується за певні періоди часу.

Головний потік¶

Кінцевим результатом став потік, що фільтрує повідомлення, нормалізує (до певного типу), виводить у кінцеві дебагери.

Вивід помилок

Під час виконання програми у вивід можуть потрапити винятки, спричинені компонентом Заглушка. Це є нормальним, враховуючи, що всі інші області навмисно було відкинуто. Для фільтрації виводу можна скористатися вбудованим selected nodes, функцією для вибору лише певних компонентів для виводу.

Приклад формату повідомлення¶

Приклад формату статистики¶

Висновок¶

Опрацювання цієї роботи дозволило опанувати навички ефективного управління потоками даних у середовищі Node-RED, зокрема освоєння принципів керування потоками, використання вбудованих та користувацьких вузлів для забезпечення оптимального оброблення інформації та забезпечення швидкої реакції на події в системі, з врахуванням подальшого розширення системи.