ydb/library/cpp/unified_agent_client, branch CLI_2.30.0

Drop backward compat with grpc-prev

2026-03-20T10:18:33Z

commit_hash:e5545cade7cc946c943e85c680db7c276edc48b5

Fix TAsyncJoiner "already joined" crash

2026-03-14T02:16:45Z

## 1. Где и когда это может происходить Гонка проявляется при **одновременном** завершении задачи (Unregister/Join) и выполнении finish-действия `CommitTimer()` в воркере. Типичные сценарии: - **Остановка агента** — при shutdown закрываются сессии, по задачам вызывается Unregister(); в это же время воркер может доходить до finish-действий уже «уходящей» задачи. - **Закрытие сессии/канала** — например, отключение клиента, ошибка — владелец вызывает Unregister() по задаче, в которой использовался таймер. - **Плагины и задачи, подверженные багу:** любые, что используют `TLocalTimersQueue` (отложенные таймеры через `DelayedExecutor`). В коде UA это, в частности: - **http_output** (`plugins/lib/http/http_sender.cpp`) — таймеры для отложенных повторов запросов и flush; - **file_input** (`plugins/file_input/file_input.cpp`) — таймеры при работе с файлами; - **logbroker_output_new** (`plugins/logbroker_output_new/logbroker_output_impl.h`) — таймер обновления метрик. Во всех этих случаях задача владеет `TLocalTimersQueue`, при установке таймера в finish-действие попадает `CommitTimer()`, и при быстром Unregister() возможна гонка с Join(). --- {% cut "Зачем потоку воркера вызывать Ref()" %} Таймер планируется **асинхронно**: `DelayedExecutor.SetTimer(Timer, triggerTime)` регистрирует callback в отдельном потоке (sleeper). Когда время наступит, callback вызовется **уже после** того, как `Run()` задачи завершился. Задача и её контекст (сессия, `TLocalTimersQueue`, `ExecutionJoiner`) должны оставаться валидными до срабатывания или отмены таймера — иначе callback приведёт к use-after-free. **Ref()** — это «удержание» задачи: пока есть лишний Ref, `Join()` не завершится (Refs не станет 0). То есть: «задача не считается полностью завершённой, пока таймер не сработал или не был сброшен». Когда таймер сбрасывают или очередь финализируют, вызывается **UnRef()** — тогда задача может перейти в joined. **Где именно вызывается Ref/UnRef:** Файл `logbroker/unified_agent/common/delayed_executor.cpp`: ```cpp void TLocalTimersQueue::CommitTimer() { if (Queue.GetCount() > 0) { const auto triggerTime = Top().Value(); if (!TimerTriggerTime.Defined() || ...) { if (!TimerTriggerTime.Defined()) { // ← ЗДЕСЬ: перед первой установкой таймера в DelayedExecutor // держим задачу «живой» до срабатывания/сброса таймера if (!TTaskExecutor::CurrentTaskOrDie().ExecutionJoiner().TryRef()) { CommitTimerScheduled = false; return; } } DelayedExecutor.SetTimer(Timer, triggerTime); // асинхронный таймер TimerTriggerTime = triggerTime; } } else if (TimerTriggerTime.Defined()) { DelayedExecutor.ResetTimer(Timer); TTaskExecutor::CurrentTaskOrDie().ExecutionJoiner().UnRef(); // таймер снят — отдаём Ref TimerTriggerTime.Clear(); } ... } ``` `CommitTimer()` вызывается из **finish-действия** задачи (добавляется в `EnsureCommitTimerScheduled()` → `AddFinishAction([this]() { CommitTimer(); })`), т.е. выполняется в потоке воркера после выхода из `Run()`. Пример использования таймера из кода плагина — `plugins/lib/http/http_sender.cpp`: там `LocalTimerQueue.SetTimer(request->Timer, triggerTime)` планирует отложенный повтор запроса; callback при срабатывании постит событие в сессию. {% endcut %} --- {% cut "Контекст: участники и суть гонки" %} **Участники:** - **ExecutionJoiner** (`TAsyncJoiner`) — объект с атомарным счётчиком `Refs` (начальное значение 1). Пока `Refs >= 1`, задачу считают «активной». `Join()` вызывает `UnRef()`; когда `Refs` становится 0, вызывается `Promise.SetValue()` («joined»). - **Поток задачи (Task/Unregister)** — владелец задачи; вызывает `Unregister()` → `ExecutionJoiner_.Join()` → внутри один раз `UnRef()`. - **Поток таймера (Worker/Timer)** — воркер пула задач; выполняет finish-действия задачи. Одно из них — `CommitTimer()`, которое при первой установке таймера вызывало `ExecutionJoiner().Ref()`. **Гонка:** между моментом, когда поток задачи делает `Join()` (и доводит `Refs` до 0), и моментом, когда поток воркера выполняет `CommitTimer()` и вызывает `Ref()`. Если `Ref()` вызывается уже после перехода в «joined», `fetch_add(1)` возвращает 0 и срабатывает `Y_ABORT_UNLESS(result >= 1, "already joined")`. {% endcut %} --- {% cut "До фикса: креш при гонке" %} Поток задачи вызывает `Unregister()` и ждёт `Join()`. Поток воркера после завершения `Run()` выполняет finish-действие `CommitTimer()`. Если к этому моменту `Join()` уже выполнил `UnRef()` и `Refs == 0`, вызов `Ref()` в `CommitTimer()` приводит к падению. ```mermaid sequenceDiagram participant TaskThread as Поток задачи participant Joiner as ExecutionJoiner participant WorkerThread as Поток воркера Note over TaskThread,WorkerThread: Задача с таймером: Run() вызвал SetTimer(), в finish-действия добавлен CommitTimer() TaskThread->>TaskThread: Unregister() TaskThread->>Joiner: Join() Joiner->>Joiner: UnRef() → Refs = 0 Joiner->>Joiner: Promise.SetValue() — joined WorkerThread->>WorkerThread: Выполняет finish-действия WorkerThread->>WorkerThread: CommitTimer() WorkerThread->>Joiner: Ref() Joiner->>Joiner: fetch_add(1) → result = 0 Joiner->>WorkerThread: Y_ABORT_UNLESS(result >= 1) — CRASH ``` **Итог:** в момент вызова `Ref()` в `CommitTimer()` объект уже в состоянии «joined» (`Refs == 0`), проверка в `Ref()` не выполняется → **SIGABRT**. {% endcut %} --- {% cut "После фикса: корректный выход без креша" %} В `CommitTimer()` вместо `Ref()` вызывается `TryRef()`: атомарно проверяется `Refs >= 1` (через CAS); если уже 0, `TryRef()` возвращает `false` и `CommitTimer()` сразу выходит, не вызывая `Ref()` и не трогая таймер. ```mermaid sequenceDiagram participant TaskThread as Поток задачи participant Joiner as ExecutionJoiner participant WorkerThread as Поток воркера Note over TaskThread,WorkerThread: Та же гонка: Join() и CommitTimer() выполняются почти одновременно TaskThread->>TaskThread: Unregister() TaskThread->>Joiner: Join() Joiner->>Joiner: UnRef() → Refs = 0 Joiner->>Joiner: Promise.SetValue() — joined WorkerThread->>WorkerThread: Выполняет finish-действия WorkerThread->>WorkerThread: CommitTimer() WorkerThread->>Joiner: TryRef() Joiner->>Joiner: load(Refs) = 0 → current < 1 Joiner->>WorkerThread: return false WorkerThread->>WorkerThread: CommitTimerScheduled = false return Note over WorkerThread: Таймер не ставится, креша нет ``` **Итог:** при уже «joined» состоянии `TryRef()` возвращает `false`, `CommitTimer()` завершается без вызова `Ref()` и без падения. {% endcut %} --- {% cut "Сводка изменений" %} | Место | До фикса | После фикса | |-------|----------|-------------| | `CommitTimer()` при первой установке таймера | `Ref()` → при Refs=0 креш | `TryRef()` → при `false` ранний выход | | `TAsyncJoiner` | Только `Ref()` / `UnRef()` | Добавлен `TryRef()` (CAS, при refs<1 возврат false) | | `Finalize()` | Не сбрасывал активный таймер | При `TimerTriggerTime.Defined()` — `ResetTimer`, `UnRef()`, `Clear()` до `Finalized = true` | Тест `TestTimerQueueUnregisterNoCrash` (500 итераций: задача с таймером → Pulse → Unregister) без фикса периодически воспроизводит креш; с фиксом — стабильно зелёный. {% endcut %} commit_hash:5f57d88fc53f44db31e87deaeca57a7e9ef262ca

/0: Implement TExecutorSet and TExecutorSetManager

2026-02-25T12:06:35Z

## Абстракция управления executor’ами (TExecutorSet, TExecutorSetManager) ### Что сделано Вводится единый слой для работы с пулами потоков agent’а: main, background и IO. **TExecutorSet** — представление набора executor’ов: - `Main()`, `Background()`, `IO()` — одинаковый интерфейс для всех трёх пулов - IO возвращает тот же объект, что и Main, когда `separate_io_thread_pool == 0`, иначе — отдельный executor **TExecutorSetManager** — создание и владение executor’ами: - `GetExecutors(EExecutorPool pool)` — возвращает нужный набор; пока используются только `Global` и `Metrics` (оба отдают один и тот же набор) - `Start()`, `Stop()` — жизненный цикл - Main создаётся сразу, Background и IO — лениво при первом обращении **Рефакторинг TAgent**: - Вместо отдельных полей `TaskExecutor`, `BackgroundTaskExecutor`, `IOTaskExecutor` и `DelayedExecutor` используется `ExecutorSetManager` - Обращения к executor’ам идут через `ExecutorSetManager->GetExecutors(EExecutorPool::Global)` - FSNotifier, MemoryQuotaManager, AgentContext, CountersUpdater работают с глобальными executor’ами **RegisterCountersWith**: - Регистрирует счётчики main, background, IO (если включён отдельный IO-пул) и delayed - Раньше IO-счётчики не учитывались — это исправлено **Тесты**: - Добавлен gtest-модуль `executor_set_tests` в `tests/gtests/`: - Проверяются GetExecutors, доступность Main, ленивая инициализация Background - Поведение IO при `separate_io_thread_pool == 0` и `> 0` - Start/Stop и параметр pool в GetExecutors ### Цель Подготовка к PR1: выделение отдельных пулов для pipelines. Сейчас меняется только абстракция, без изменения конфигурации и без добавления dedicated pools. commit_hash:0f1b2205c474db1b601a72f063916333a74882d6

/3: Get rid of YLOG_F macros in favor of just YLOG

2026-02-20T18:52:08Z

_Все вызовы вида YLOG\*\_F(...) заменены на YLOG\*_(...) по всему дереву unified\_agent (например, YLOG\_DEBUG\_F → YLOG\_DEBUG, YLOG\_ERR\_F → YLOG\_ERROR). commit_hash:483c6f5cb6db2d44e9d71b427a697dc2850cba1f

/2: Use our internal logger header, add logging metrics

2026-02-19T20:07:43Z

# Улучшения библиотеки логирования и переход на троттлинг логов ## Описание Этот PR содержит улучшения системы логирования unified\_agent с акцентом на предотвращение флуда логов и добавление метрик логирования. ## Основные изменения ### 1. Улучшения библиотеки логирования (`library/cpp/unified_agent_client/logger`) - **Встроенный троттлинг логов**: Добавлена поддержка ограничения частоты логирования на уровне библиотеки - Новые макросы `YLOG_*_T` с автоматическим троттлингом (20 логов на 10 секунд по умолчанию) - Независимый троттлинг для каждой точки логирования (по `__FILE__:__LINE__`) - Автоматический подсчет подавленных сообщений с выводом `[+N suppressed]` - **Метрики логирования**: Добавлены счетчики для мониторинга активности логирования - `RecordsReceived` - общее количество попыток логирования - `RecordsDropped` - количество подавленных сообщений из-за троттлинга - Счетчики передаются через `TLogger::TCounters` при создании логгера - **Оптимизация производительности**: - Использование `GetCycleCount()` для быстрого получения времени (вместо системных вызовов) - Relaxed memory ordering для атомарных операций (достаточно для троттлинга) - Минимальные накладные расходы при отключенном логировании ### 2. Переход всех логов агента на троттлинг - **Унификация макросов**: Все макросы `YLOG_*` в `logbroker/unified_agent/common/util/logger.h` теперь используют троттлинг - `YLOG_DEBUG`, `YLOG_INFO`, `YLOG_WARNING`, `YLOG_ERROR` и т.д. теперь автоматически применяют троттлинг - Старые макросы `YLOG_*_F` теперь алиасы для новых троттлированных версий - Обратная совместимость полностью сохранена - **Обновление документации**: Файл `for_ai_cpp.md` обновлен с новыми рекомендациями по логированию ### 3. Интеграция метрик логирования в телеметрию - **Новые счетчики в `TAgentLogCounters`**: - `RecordsReceived` - rate метрика `agent.log.records_received` - `RecordsDropped` - rate метрика `agent.log.records_dropped` - **Экспорт метрик**: Метрики логирования автоматически собираются и отправляются в телеметрию агента - **Рефакторинг конструктора `TAgent`**: - Упрощена передача счетчиков через структуру `TAgent::TCounters` - Счетчики логирования передаются в `TLogger` при инициализации ### 4. Тесты - **Перенос тестов**: Тесты логирования перемещены из `logbroker/unified_agent/tests/gtests/logger_tests` в `library/cpp/unified_agent_client/ut` - **Расширенное покрытие**: - Тесты базового троттлинга - Тесты счетчика подавленных сообщений - Тесты независимого троттлинга для разных точек логирования - Тесты форматирования сообщений - Тесты счетчиков метрик ### 5. Исправления и улучшения - **Удаление дублирования кода**: Логика троттлинга теперь находится только в `library/cpp/unified_agent_client/logger.cpp` - **Упрощение API**: Удален отдельный файл `logbroker/unified_agent/common/util/logger.cpp` - **Обновление импортов**: Все файлы обновлены для использования правильных заголовочных файлов ## Преимущества 1. **Защита от флуда логов**: Автоматическое ограничение частоты логирования предотвращает переполнение логов 2. **Наблюдаемость**: Метрики логирования позволяют отслеживать активность и проблемы с логированием 3. **Производительность**: Минимальные накладные расходы благодаря оптимизированной реализации 4. **Простота использования**: Троттлинг работает автоматически, не требует изменений в коде 5. **Обратная совместимость**: Все существующие макросы продолжают работать ## Тестирование - ✅ Все unit-тесты логирования проходят - ✅ Integration тесты обновлены (исключение нестабильной метрики `RecordsReceived` из сравнений) - ✅ Проверена работа троттлинга в реальных условиях commit_hash:75fc97a8576114446bfb9ec11efbb80df322e443

/0: Implement logging with throttling, replace YLOG macro with a new one

2026-02-15T17:01:58Z

### Добавлен троттлинг логов для предотвращения флуда **Основная цель:** Защита от переполнения логов повторяющимися сообщениями. #### Что добавлено: 1. **Механизм троттлинга логов** ([`common/util/logger.h`](common/util/logger.h), [`common/util/logger.cpp`](common/util/logger.cpp)) - Ограничение: по умолчанию максимум 20 сообщений за 10 секунд для каждой точки логирования - Независимый троттлинг для каждого места вызова `YLOG` (через `static` переменную состояния) - Подсчёт пропущенных сообщений с выводом в формате `[+N suppressed]` - Минимальные накладные расходы: использует `GetCycleCount()` вместо системных вызовов времени - Атомарные операции для потокобезопасности 2. **Настраиваемый троттлинг** - Макрос `YLOG_THROTTLED(logger, priority, maxLogs, intervalSec, ...)` для кастомных лимитов - Макросы `YLOG_THROTTLE_MAX_LOGS` и `YLOG_THROTTLE_INTERVAL_SECS` для глобальной настройки (например, в тестах) 3. **Поддержка C++20 `std::format`** - Макросы `YLOG_*` теперь поддерживают форматирование в стиле C++20: `YLOG_ERR("value: {}", x)` - Обратная совместимость с `YLOG_*_F` макросами сохранена - Специальная обработка `TStringBuilder` для избежания проблем с временными объектами 4. **Тесты** ([`tests/gtests/logger_tests/`](tests/gtests/logger_tests/)) - Проверка базового троттлинга - Проверка счётчика пропущенных сообщений - Проверка независимости троттлинга для разных точек логирования - Проверка форматирования сообщений #### Технические изменения: **Изменён порядок аргументов в макросе `YLOG`:** - Было: `YLOG(priority, message, logger)` - Стало: `YLOG(logger, priority, message, ...)` Это изменение потребовало обновления всех вызовов `YLOG` в кодовой базе (~30 файлов), но обеспечивает: - Единообразие с другими логирующими библиотеками - Возможность использования variadic arguments для форматирования - Правильную работу с `const auto&` для продления времени жизни временных объектов #### Примеры использования: ```cpp // Базовое использование (троттлинг 20 логов/10 сек) YLOG_ERR("Error occurred"); YLOG_INFO("Processing item: {}", itemId); ``` #### Влияние на производительность: - Минимальные накладные расходы: ~10-20 наносекунд на вызов (использование CPU cycles вместо системного времени) - Атомарные операции только при необходимости логирования - Отсутствие дополнительных (по сравнению с текущим кодом) аллокаций памяти в hot path #### Обратная совместимость: - Все существующие `YLOG_*` и `YLOG_*_F` макросы работают как раньше - Добавлен троттлинг по умолчанию для всех логов (может потребовать внимания при отладке) commit_hash:d6e98b82486b99b945a1ba264fb2da969db4fcfc

Add throttling for max inflight error logs

2026-01-14T22:51:38Z

## Problem Log spam with repeated ``'max inflight of [***] bytes reached, [***] bytes dropped'`` errors appearing multiple times within the same second, causing log flooding. ## Solution Implemented one-second throttling mechanism for this specific error message to prevent log spam while maintaining accurate counter tracking. ## Testing - Added `TestMaxInflightBytesThrottling` integration test - Confirms throttling limits log entries to ≤3 for 50 rapid message drops (vs 50 without throttling) - Ensures all dropped message counters remain accurate regardless of throttling commit_hash:58f44ca8ce2b8b416586f8ca7a3d3ca971f1e9cb

fix: Migrate to YLOG_*_F and eliminate dangerous c_str() patterns

2025-11-29T08:46:09Z

## Summary Eliminated all dangerous c_str() patterns from unified_agent by migrating to modern YLOG_*_F formatting. ## Changes - Remove .c_str() on temporaries in YLOG_*_F macros (1 fix) - Migrate YLOG_*(Sprintf(...)) to modern YLOG_*_F(...) (23 fixes) ## Benefits - Modern C++20 type-safe formatting with std::format - Cleaner, more maintainable code (40% less in some cases) - Better performance (no Sprintf overhead) commit_hash:ee3f665ccc823d9e2165a90c036e5bd887cd179e

- Global Storage Limit Feature

2025-11-28T01:24:15Z

See full description in separate comment. The open source part is to add implementation for formatter specialization for TString. commit_hash:9501df4a7287050b72162b80823d5cbbd1d7464f

Create new plugin agent_logs_input

2025-11-25T06:07:44Z

**Summary** Реализован новый входной плагин `agent_logs` для отправки логов самого агента в пайплайн unified\_agent. Это решает проблему доступа к логам агента в production окружении, где требовались права System developer для доступа к файлам на дисках. ## **What's Changed** {#whats-changed} ### **New Plugin: `agent_logs_input`** {#new-plugin-agent_logs_input} Добавлен новый входной плагин, который: * Перехватывает логи агента через кастомный `TLogBackend` * Отправляет логи в пайплайн стриммингово (не пулингом) * Поддерживает фильтрацию по уровню логов (настройка **level**, по умолчанию `NOTICE`) ### **Implementation Details** {#implementation-details} **Основные компоненты:** 1. **TAgentLogsBackend** - кастомный log backend: * Наследуется от `TLogBackend` * Перехватывает все логи агента через метод `WriteData()` * Фильтрует логи по приоритету * Отправляет отфильтрованные логи в `IMessageConsumer` * Поддерживает graceful stop 2. **TPlugin** - основной класс плагина: * Управляет lifecycle плагина * Создает сессию с ID `agent_logs` * Передает конфигурацию в session handler 3. **TSessionHandler** - обработчик сессии: * Создает и регистрирует `TAgentLogsBackend` * Добавляет backend в глобальный logger через `AddLog()` * Управляет остановкой backend'а **Пример конфигурации:** ``` routes: - input: plugin: agent_logs config: level: INFO channel: pipe: - filter: plugin: format_otel_logs output: # https://nda.ya.ru/t/YexxO8Qf7NZMzy plugin: otel_logs config: url: "collector.logs.yandex-team.ru:443" project: hobbit_test_project service: logs # https://nda.ya.ru/t/o4MQT5rI7NZMzz oauth: secret: env: TEST_LOGS_OAUTH agent_log: priority: INFO rate_limit_bytes: 100000 ``` commit_hash:c6f115ab98c30a47845ce9f9f8a58ef89973d422