События и логи приложения (потоки)

Потоки — это хранение логов и событий приложения, по которым потом нужно искать и разбираться: журнал запросов к API, журнал аудита входов и прав, платежи, доставка вебхуков, ответы внешних API, бизнес-события. Приложение отправляет агенту стека JSON-записи с любыми полями, а Наблюдатель собирает записи одного вида в поток — «Запросы», «Платежи», «Аудит».

Схему объявлять не нужно: поля, фильтры и подсказки выводятся из данных сами. Поток появляется на вкладке «Потоки» стека, как только придёт первая запись. Записи хранятся 30 дней.

#Чем поток отличается от ошибок и метрик

• Поток хранит каждое событие со всеми полями — для поиска и разбора.
• Свои метрики — числа во времени: длина очереди, число заказов.
• Ошибки приложения агент собирает из stdout/stderr сам, отправлять их не нужно.

#Что стоит отправлять в потоки

#Как отправлять записи агенту

Агент стека принимает записи по HTTP на порту 4319 без авторизации: он доступен только внутри вашей сети и сам добавляет токен при пересылке.

curl -X POST http://nabl-agent:4319/v1/log -H 'Content-Type: application/json' -d '{
  "stream": "Запросы",
  "records": [
    {"method": "POST", "path": "/api/login", "status": 401, "duration_ms": 84,
     "user": "ivan@mail.ru", "at": "2026-10-30T12:00:00Z"}
  ]
}'

Формат:

• Имя потока — поле "stream", можно по-русски, до 128 символов. Без него записи попадут в поток default.
• Пачка — "records": [...], одна запись — "record": {...} или просто объект записи с полем "stream" рядом с данными (агент 0.8.3+).
• NDJSON: заголовок Content-Type: application/x-ndjson, по записи на строку, в каждой своё "stream".
• Время записи — поле observed_at, at, timestamp или time в формате RFC3339. Без него — время приёма.
• В одном запросе — до 1000 записей и до 16 МиБ.
• Ответ 202 — записи приняты. 503 pii key not ready — агент ещё стартует, повторите позже.

Нужен агент стека 0.5.0 или новее.

#Отправка пачками в фоне: примеры кода

Отправляйте записи асинхронно и пачками: кладите запись в ограниченную очередь в памяти и раз в 1–5 секунд отправляйте пачку в фоне. Запрос пользователя не должен ждать агента, а если агент недоступен, пачка отбрасывается, а не копится.

#Node.js

const queue = [];
function logEvent(stream, fields) {
  queue.push({ stream, at: new Date().toISOString(), ...fields });
  if (queue.length > 5000) queue.splice(0, queue.length - 5000); // агент недоступен — старое отбрасываем
}
setInterval(async () => {
  const batch = queue.splice(0, 1000);
  if (!batch.length) return;
  try {
    await fetch('http://nabl-agent:4319/v1/log', {
      method: 'POST',
      headers: { 'Content-Type': 'application/x-ndjson' },
      body: batch.map((r) => JSON.stringify(r)).join('\n'),
      signal: AbortSignal.timeout(2000),
    });
  } catch { /* пачка теряется, приложение не страдает */ }
}, 2000).unref();

// logEvent('Платежи', { order_id: 'A-1042', amount: 3400, provider: 'bank', status: 'failed', error_code: 'timeout' });

#Python (только стандартная библиотека)

import json, queue, threading, time, urllib.request
from datetime import datetime, timezone

_events = queue.Queue(maxsize=5000)

def log_event(stream, **fields):
    try:
        _events.put_nowait({"stream": stream, "at": datetime.now(timezone.utc).isoformat(), **fields})
    except queue.Full:
        pass  # агент недоступен — не копим

def _sender():
    while True:
        time.sleep(2)
        batch = []
        while len(batch) < 1000:
            try:
                batch.append(_events.get_nowait())
            except queue.Empty:
                break
        if not batch:
            continue
        body = "\n".join(json.dumps(r, ensure_ascii=False) for r in batch).encode()
        req = urllib.request.Request("http://nabl-agent:4319/v1/log", body,
                                     {"Content-Type": "application/x-ndjson"})
        try:
            urllib.request.urlopen(req, timeout=2).read()
        except Exception:
            pass

threading.Thread(target=_sender, daemon=True).start()

Go, .NET, Java — тот же шаблон: буферизованный канал или очередь (chan, Channel<T>, ArrayBlockingQueue) с неблокирующей записью и фоновая отправка пачки по таймеру с таймаутом 2 секунды.

#Как писать хорошие записи

• Плоские или неглубокие объекты. Вложенность до 4 уровней превращается в поля вида response.status. Массивы и null сохраняются и находятся текстовым поиском, но фильтровать и считать по ним нельзя — нужное значение вынесите отдельным полем.
• Параметры и тела запросов — строкой. То, у чего в каждой записи свои ключи, передавайте строкой ("query": "{\"limit\":\"10\"}"), а не объектом. Иначе у каждого эндпоинта появятся свои поля, и список полей быстро упрётся в лимит. Строка находится текстовым поиском (агент 0.11.0+).
• В path — шаблон маршрута (/api/orders/:id), а не адрес с id: иначе разбивка по пути бесполезна.
• Числа — числами ("duration_ms": 84, а не "84ms"), булевы — true/false.
• Добавляйте trace_id, если в приложении есть трассировка (см. ниже).

#Связь записей с трассами

Если приложение подключено к трассировке, кладите в каждую запись trace_id текущего запроса:

Что это даёт:

• trace_id в таблице записей становится ссылкой и открывает трассу запроса;
• у трассы появляется блок «Записи потоков с этим trace_id» — что пришло в запросе и что ответили;
• Наблюдатель сам определяет, какой сервис пишет поток: на вкладке видно «Пишет сервис … (по trace_id записей)», а поток показан у сервиса на карте и странице сервиса;
• ответы 5xx из журналов запросов учитываются при поиске причины сбоя сервиса.

Если в записях потока нет trace_id, вкладка подскажет об этом. Если записи ведут в несколько сервисов, точный сервис можно указать через ИИ-ассистента.

#Персональные данные в потоках

Агент обезличивает записи до отправки в облако:

• пароли, токены, Authorization, cookie, CVV удаляются;
• email, телефон, ИНН, СНИЛС и IP заменяются маской для показа и псевдонимом для точного поиска;
• от номера карты остаются первые 6 и последние 4 цифры.

Это работает и для вложенного JSON, и для текста внутри значений: в "order created for ivan@mail.ru" маскируется только email. Поэтому слать «сырые» записи безопасно — маскировать заранее не нужно. Обезличенные поля в таблице помечены значком «ПДн». Подробнее — в статье как работает агент.

#Вкладка «Потоки»

Вкладка появляется на странице стека с подключённым агентом. Наверху — переключатель потоков с числом записей за срок хранения, режим «Таблица» или «Дашборд», кнопка обновления и меню «⋯» с действиями над потоком.

#Поиск и фильтры

Строка «Поиск за период: текст, поле:значение…» и выбор периода — «За 24 часа», «За 7 дней», «За 30 дней». При вводе имени поля появляются подсказки, в том числе вложенных ключей (user. → user.id). Условия пишутся через пробел, выполняться должны все:

По исходному значению ищутся email, телефон, ИНН, СНИЛС и IP; номер карты — нет. «ИЛИ» и отрицания для текста нет — сделайте несколько запросов.

Под строкой поиска — автоматические фильтры по полям с небольшим числом значений: status, method, result. Выбор значения дописывает условие в поиск. Если фильтров много, остальные спрятаны под «Ещё фильтры».

#Таблица и колонки

Записи показаны таблицей, новые сверху; дальше — по кнопке «Показать ещё». Любую запись можно развернуть и посмотреть все её поля.

Без настройки колонки подбираются сами: сначала статус, метод, путь, длительность и ошибки, без повтора времени записи — оно и так в колонке «Когда». Кнопка «Колонки» позволяет выбрать до 8 полей и их порядок, например status, method, path, duration_ms, user_id. Колонки видят все, кто смотрит поток. «Подобрать автоматически» возвращает выбор Наблюдателя.

#Дашборд и разбивка

Режим «Дашборд» строит графики без настройки: число событий во времени, разбивку по ключевым полям (например, сколько записей с каждым status и какая доля приходится на прочие значения) и тренды числовых полей — среднее, p95, минимум и максимум длительности или суммы. Период — 24 часа, 7 или 30 дней.

Посчитать произвольную разбивку — «откуда 500-е по эндпоинтам», «p95 длительности входа», «что было с платежами вчера с 14:00 до 16:00» — удобно через ИИ-ассистента: он считает по всем записям потока за период или точный интервал.

#Поля потока и лимиты

Поля сверх лимита не попадают в фильтры, подсказки и колонки, но значения остаются в записи и находятся текстовым поиском. В этом случае на вкладке появится предупреждение: какие объекты занимают больше всего полей (обычно параметры или тела запросов, переданные объектом) и что исправить. Если в стеке агент старше 0.11.0, оно предложит его обновить: старые агенты превращают JSON внутри строк в поля.

Убрать лишние поля. Меню «⋯» → «Убрать лишние поля…». Возле каждого поля написано, приходит ли оно: «приходит» или «не приходит 3 дн.». Кнопка «Отметить все, что не приходят» выбирает поля, от которых приложение уже отказалось. Отмеченные поля уходят из подсказок, фильтров и колонок; записи не меняются. Если отметить поле, которое ещё приходит, оно вернётся со следующей записью — главное исправить отправку.

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

#Уведомления об аномалиях в потоке

Когда в стеке появляется новый поток, Наблюдатель сам заводит для него включённое правило «аномалия по количеству событий». Порог задавать не нужно: правило сравнивает последний завершённый 10-минутный интервал с историей потока за 6 часов и срабатывает на резкий рост или падение — поток ошибок вырос в разы или журнал запросов вдруг замолчал. Первые ~2 часа правило копит историю и не срабатывает.

• Аномалию по числовому полю — например, среднему duration_ms или amount — можно добавить отдельным правилом в разделе «Правила» проекта: выберите поток, поле и чувствительность.
• Чувствительность: низкая — только явные всплески (для шумных потоков), обычная, высокая — реагировать на лёгкие отклонения.
• Если одна и та же аномалия повторяется 5 раз за сутки, Наблюдатель считает это обычным поведением: пятое уведомление предупредит об этом, а следующие не придут, пока аномалия не станет реже.
• Правил с фиксированным порогом вроде «больше 10 ошибок за 5 минут» для потоков нет — только аномалии.

Уведомления приходят в каналы проекта. Подробнее — в статьях аномалии и каналы уведомлений.

#Как удалить поток

Имя потока задаёт приложение — поле "stream" в записях. Если имя в коде сменили, со следующей записи появится новый поток, а старый перестанет пополняться и останется в списке.

Удалить поток — меню «⋯» → «Удалить поток…». Поток удаляется вместе со всеми записями и правилом аномалии для него, без возможности восстановления. Если приложение продолжит слать записи со старым именем, поток появится снова — сначала уберите или переименуйте отправку в коде.

Если записей в потоке не было весь срок хранения, на вкладке появится пометка «Записей нет 30 дней» с предложением удалить поток.

#Как проверить

После первой отправки откройте стек → вкладка «Потоки»: появится поток с числом записей. Найдите запись поиском, переключитесь на «Дашборд» и настройте колонки под то, как будете разбирать поток.

#Что дальше

• Аномалии — уведомления, когда поток ведёт себя необычно.
• Трассировка OpenTelemetry — чтобы записи открывали свою трассу.
• Свои метрики — когда нужно число во времени, а не каждое событие.
• ИИ-ассистент — поиск и подсчёт по записям за точные даты.