Дедупликация строк на доставщике логов: сравниваем OpenTelemetry, Vector, vlagent, Fluent Bit Grafana Alloy и Filebeat

Один и тот же стек трейс, напечатанный 50 тысяч раз, не сделает расследование инцидентов в 50 тысяч раз полезнее. Зато он честно увеличит сетевой трафик, объем хранения и время на извлечение данных.

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

Меня зовут Антон Касимов, я работаю в Gals Software, веду авторский телеграм по наблюдаемости Мониторим ИТ и я сравнил пять известных лог-шипперов (мог бы назвать их популярными, но это субъективно):

• OpenTelemetry Collector

• Vector

• vlagent

• Fluent Bit

• Grafana Alloy

• Filebeat

А затем прогнал через них одинаковый поток данных. Агентов, которые действительно умеют в дедупликацию оказалось два: OpenTelemetry Collector и Vector. У Fluent Bit задачу можно решить скриптом, а у Filebeat, vlagent и Alloy встроенного механизма дедупликации вообще не оказалось.

Сначала определимся, что именно считаем дублем

Возьмём две записи:

{"service":"billing","level":"error","message":"database timeout","request_id":"a-17","timestamp":"2026-09-18T09:00:00Z"}
{"service":"billing","level":"error","message":"database timeout","request_id":"b-42","timestamp":"2026-09-18T09:00:01Z"}

Побайтово они разные. По смыслу это может быть один повторяющийся сбой. Чтобы объединить записи, агенту нужен ключ дедупликации, например:

(service, level, message)

Если включить в ключ timestamp и request_id, каждый лог станет уникальным, и дедупликация ничего не даст. Если оставить только message, можно случайно склеить события разных сервисов или тенантов.

Поэтому, разделим наших подопытных на 4 категории:

1. Дедупликация по отпечатку — агент запоминает выбранные поля и удаляет или агрегирует повтор в пределах временного окна.

2. Лимит пропускной способности — агент не пропускает больше N записей в секунду. Он не проверяет, одинаковые они или нет.

3. Защита от повторного чтения — в специальном файле хранится текущая позиция чтения.

4. Дедупликация при запросе/отображении — бэкэнд хранит все события, а UI скрывает похожие строки от пользователя.

Самая честная дедупликация происходит в первом случае. Остальные тоже полезны, но больше имеют косметический эффект.

Короткий ответ

Пояснение: LRU (Least Recently Used) — алгоритм вытеснения из кэша: когда кэш заполнен, удаляется элемент, которым дольше всего не пользовались.

1. OpenTelemetry Collector: агрегировать, чтобы помнить

В Contrib-дистрибутиве есть процессор log_dedup. На текущий момент его релиз имеет статус альфа.

Механика такая:

1. Процессор собирает одинаковые записи в течение заданного интервала, по умолчанию 10 секунд.

2. В конце интервала выпускает одну запись.

3. Добавляет к ней log_count, first_observed_timestamp и last_observed_timestamp.

Это 100% честная дедупликация. Вместо ста одинаковых ошибок мы получаем одну ошибку с признаком «встретилась сто раз».

Как формируется ключ

Без дополнительных настроек уникальность определяется по body, resource attributes, severity и log attributes. На реальных структурированных логах это почти всегда слишком строгий ключ: достаточно меняющегося request_id, чтобы повторы перестали считаться повторами.

Лучше выбрать небольшой белый список через include_fields:

processors:
  log_dedup:
    interval: 30s
    include_fields:
      - attributes.service
      - attributes.level
      - attributes.message
    log_count_attribute: repeat_count

Альтернатива — exclude_fields, но одновременно использовать оба параметра нельзя. Есть ещё неприятная деталь: исключённые поля удаляются и из итоговой агрегированной записи. Поэтому allowlist обычно понятнее и безопаснее.

Также через conditions можно отправлять в дедупликацию только выбранные логи, например только ошибки.

Полный минимальный пример

После json_parser поля из тела в этом примере оказываются в attributes, поэтому путь начинается с attributes.*:

receivers:
  tcp_log:
    listen_address: 0.0.0.0:9000
    operators:
      - type: json_parser
        parse_from: body

processors:
  log_dedup:
    interval: 10s
    include_fields:
      - attributes.service
      - attributes.level
      - attributes.message
    log_count_attribute: duplicate_count

exporters:
  otlphttp/logs:
    endpoint: https://logs.example.com

service:
  pipelines:
    logs:
      receivers: [tcp_log]
      processors: [log_dedup]
      exporters: [otlphttp/logs]

Что важно учесть

• Запись задерживается до закрытия interval. Это плата за агрегацию.

• Метка времени итогового лога — момент его выпуска из дедупликатора, а не метка времени первой исходной записи. Исходный диапазон можно обозревать в first_observed_timestamp и last_observed_timestamp.

• Высокая кардинальность способна съесть память раньше, чем вы увидите экономию.

• После рестарта коллектора активное окно начинается заново. Очередь экспортера не делает состояние log_dedup постоянным.

• Старое имя logdedup пока принимается, но уже помечено как устаревшее; актуальное имя — log_dedup.

Где преимущество

OpenTelemetry лучше всего подходит там, где важно сократить объём передаваемых данных. По repeat_count можно строить алерты и статистику. Обратная сторона — альфа-статус релиза, задержка на интервал и необходимость внимательно контролировать кардинальность (чтобы потребление памяти не выросло чересчур драматически).

2. Vector: первый раз бывает только раз

У Vector есть стабильный stateful transform dedupe. Он хранит последние отпечатки в LRU-cache и удаляет повторные события.

Минимальная рабочая конфигурация для структурированных логов:

transforms:
  dedupe_errors:
    type: dedupe
    inputs: [input]
    fields:
      match:
        - service
        - level
        - message
    cache:
      num_events: 2000
    time_settings:
      max_age_ms: 60000
      refresh_on_drop: false

Почему конфигурацию по умолчанию стоит менять

Если fields не заданы, Vector сравнивает timestamp, host и message из глобальной схемы. Для двух одинаковых сообщений с разными timestamp это разные события. Формально dedupe включён, фактически ничего не дедуплицируется.

Есть два способа собрать ключ:

• fields.match — учитывать только перечисленные поля;

• fields.ignore — учитывать всё, кроме перечисленных полей.

Советую выбирать match. Он лучше документирует смысл события и экономнее по памяти: Vector хранит в кэше только значения полей, участвующих в сравнении. При ignore приходится хранить почти всю запись за исключением игнорируемых полей.

Окно Vector — не интервал OTel

cache.num_events по умолчанию равен 5000. Это число событий в кэше. При высокой кардинальности полезные строки логов быстро вытесняются, и старые дубли снова начинают проходить. Аtime_settings.max_age_ms добавляет временную границу.

time_settings.max_age_ms и cache.num_events работают в логике И. Т.е., чтобы дубли отбрасывались, не должно истечь время time_settings.max_age_ms и сообщение должно находиться в cache.num_events.

Дополнительный параметр refresh_on_drop меняет смысл окна:

• false — возраст считается от первого принятого события; через минуту следующая копия снова пройдёт;

• true — каждый удалённый повтор продлевает жизнь ключа; непрерывный шторм сообщений может подавляться бесконечно, пока пауза не превысит max_age_ms.

Это небольшая настройка с большим эффектом. В первом варианте вы периодически видите, что проблема сохраняется, во втором — после первого сообщения остальные отбрасываются.

Что теряется

Vector оставляет первое сообщение и удаляет остальные. Встроенного аналога repeat_count у этого transform нет. Сколько именно повторов произошло, можно увидеть в технической метрике component_discarded_events_total, но это число относится к компоненту, а не к конкретной сигнатуре лога.

LRU находится в памяти. После рестарта тот же лог снова будет считаться новым. То же самое происходит при горизонтальном масштабировании: два Vector не делятся друг с другом кэшем, поэтому глобальной дедупликации между репликами нет.

Где преимущество

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

3. Fluent Bit: сделай своими руками

В актуальном списке фильтров Fluent Bit нет универсального stateful фильтра дедупликации. throttle тоже не замена: он ограничивает количество записей, не сравнивая их содержание.

Практический вариант — lua filter. Callback возвращает -1, когда запись нужно отбросить.

Пример конфигурацияи:

service:
  flush: 1
  log_level: info

pipeline:
  inputs:
    - name: tail
      path: /var/log/containers/*.log
      tag: kubernetes.*

  filters:
    - name: lua
      match: kubernetes.*
      script: /fluent-bit/scripts/dedupe.lua
      call: dedupe
      protected_mode: true

  outputs:
    - name: forward
      match: '*'
      host: logs.example.com
      port: 24224

Упрощённый Lua-кэш с TTL и ограничением размера:

local seen = {}
local fifo = {}
local head = 1
local tail = 0

local max_entries = 2000
local ttl_seconds = 60

local function evict(now)
  while head <= tail do
    local item = fifo[head]
    local over_capacity = (tail - head + 1) > max_entries
    local expired = (now - item.seen_at) >= ttl_seconds

    if not over_capacity and not expired then
      break
    end

    if seen[item.key] == item.seen_at then
      seen[item.key] = nil
    end
    fifo[head] = nil
    head = head + 1
  end
end

function dedupe(tag, timestamp, record)
  local now = os.time()
  local key = table.concat({
    tostring(record["service"]),
    tostring(record["level"]),
    tostring(record["message"]),
  }, "\31")

  evict(now)

  if seen[key] ~= nil and (now - seen[key]) < ttl_seconds then
    return -1, 0, 0
  end

  seen[key] = now
  tail = tail + 1
  fifo[tail] = { key = key, seen_at = now }
  return 0, timestamp, record
end

Что придётся решить самостоятельно

• состав ключа и канонизацию типов

• TTL и политику eviction

• ограничение памяти

• коллизии, если вместо строки используется хэш

• счётчики удалённых записей

• поведение при рестарте

• выявление идентичностей результата на нескольких репликах

• тесты при изменении схемы лога

Этот пример хранит состояние только в памяти и теряет его после рестарта. При большом размере Lua-таблицы тоже не бесплатны по части утилизации RAM. Подключить внешнее хранилище теоретически можно, но тогда конвейер обработки получает зависимость и перестаёт быть простым.

Где преимущество

Fluent Bit выигрывает там, где он уже развёрнут повсеместно, а правило узкое и легко тестируется: например, подавить конкретный известные штормы на определенных узлах. Для реализации общего подхода к дедупликации лучше присмотреться к Vector или OTel: меньше собственного кода и больше универсальности.

4. vlagent: дедупка не выросла

vlagent умеет читать файлы и Kubernetes logs, добавлять и удалять поля, реплицировать поток и буферизовать его на диске при недоступности VictoriaLogs. Универсального кэша отпечатков логов у него нет.

Здесь легко перепутать два механизма:

• checkpoint запоминает позицию чтения файла и помогает не перечитать уже обработанные строки после рестарта;

• -remoteWrite.tmpDataPath хранит недоставленные данные и отправляет их после восстановления backend.

Ни один из них не определяет, что две разные строки с одинаковыми service, level и _msg являются дублями. Если приложение записало одно сообщение дважды, vlagent отправит оба. Если одна запись пришла через два независимых источника, checkpoint тоже не поможет.

Пример настроек надёжной доставки:

./vlagent-prod \
  -fileCollector.glob='/var/log/apps/*.log' \
  -fileCollector.checkpointsPath=/var/lib/vlagent/file-checkpoints.json \
  -tmpDataPath=/var/lib/vlagent \
  -remoteWrite.tmpDataPath=/var/lib/vlagent/remote-write \
  -remoteWrite.maxDiskUsagePerURL=20GiB \
  -remoteWrite.url=http://victoria-logs:9428/insert/native

Эта конфигурация полезна, но к content deduplication не относится.

На стороне VictoriaLogs можно свернуть результат запроса через LogsQL:

_time:15m
| uniq by (service, level, _msg) with hits

uniq ... with hits покажет уникальные сочетания и количество совпадений. Но к этому моменту все записи уже приняты, переданы и сохранены. Экономии при передаче и хранении нет.

Кстати, в VictoriaMetrics (именно в бэкэнде) есть параметр -dedup.minScrapeInterval , который как раз-таки дедуплицирует метрики одной временной серии после их приёма. В VictoriaLogs пока такое не завезли.

Где преимущество

Преимущество vlagent — не дедупликация, а экономичный с точки зрения потребления ресурсов путь доставки в VictoriaLogs.

5. Grafana Alloy: фильтр и лимит есть, дедупликации нет

На текущий момент в каталоге компонентов Alloy нет аналога log_dedup(хоть и там под капотом OTEL-экспортер), а в loki.process нет механизма сверки отпечатков записей логов.

Ближайшие по смыслу компоненты решают другие задачи:

• stage.drop удаляет запись по условию или регулярному выражению, но ничего не знает о предыдущем событии

• stage.limit ограничитель скорости логов внутри loki.process

• otelcol.processor.filter проверяет каждую запись лога независимо и не хранит историю.

Например, этот пример конфигурации защищает бэкэнд от потока больше 100 строк в секунду, но одинаковые и разные логи для него равнозначны:

loki.process "rate_limit" {
  stage.limit {
    rate  = 100
    burst = 200
    drop  = true
  }

  forward_to = [loki.write.backend.receiver]
}

Ещё один источник путаницы — переключатель Deduplication в Grafana Explore. Режимы Exact, Numbers и Signature действительно скрывают похожие строки, но это настройка отображения результата. Она не уменьшает объём, который Alloy отправил и Loki сохранил.

Что делать пользователю Alloy

Если нужна настоящая дедупликация до backend, варианты такие:

1. поставить перед Loki pipeline Vector с transform dedupe;

2. использовать отдельный OpenTelemetry Collector Contrib с log_dedup, а Alloy оставить для остальных pipeline;

3. подавить шум через stage.drop или stage.limit, честно называя это фильтрацией или ограничением скорости передачи

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

Модуль Alloy может красиво переиспользовать комбинацию встроенных компонентов, но сам по себе не добавляет механизм дедупликации.

Где преимущество

Alloy удобен как единый агент для Grafana-стека и OpenTelemetry-компонентов. Но выбирать его ради дедупликации логов не стоит. Его сильная сторона — маршрутизация, парсинг, метки, фильтрация и доставка, а не кэширование повторов.

6. Filebeat: не кэш дублей, а идемпотентная индексация

У Filebeat нет встроенного stateful-процессора, который держит кэш отпечатков и отбрасывает одинаковые сообщения в течение заданного времени. Зато есть другой полезный механизм: Filebeat может передать Elasticsearch стабильный идентификатор документа через поле @metadata._id.

Это ответ на особенность доставки типа at least once. Представим, что Elasticsearch принял bulk-запрос, но подтверждение потерялось из-за сетевого сбоя. Filebeat не знает, сохранился документ или нет, поэтому отправляет событие повторно. Если id генерирует Elasticsearch, в индексе появляются два документа. Если Filebeat повторно присылает тот же стабильный id, новая операция перезаписывает документ с таким ID вместо создания ещё одного.

Схема получается такой:

одинаковый fingerprint → одинаковый @metadata._id → одинаковый Elasticsearch _id

Конфигурация через fingerprint

Процессор fingerprint вычисляет ID из выбранных полей. Например:

filebeat.inputs:
  - type: filestream
    id: app-json
    paths:
      - /var/log/app/*.json
    parsers:
      - ndjson:
          target: ""
          add_error_key: true

processors:
  - fingerprint:
      fields:
        - service.name
        - log.level
        - message
      target_field: "@metadata._id"
      method: sha256
      encoding: hex

output.elasticsearch:
  hosts:
    - https://elasticsearch.example:9200

Filebeat поддерживает md5, sha1, семейство sha2 и xxhash; по умолчанию используется sha256. Для дедупликации важнее не выбор криптографического алгоритма, а правильный состав полей для генерации отпечатка.

Если приложение уже пишет уникальный event_id, лучше использовать его, а не собирать ключ из текста сообщения. Для JSON, упакованного в поле message, ID можно извлечь сразу при декодировании:

processors:
  - decode_json_fields:
      fields: [message]
      target: ""
      document_id: event_id

document_id забирает значение указанного поля и кладёт его в @metadata._id. Если между Filebeat и Elasticsearch стоит Logstash, metadata нужно явно перенести в настройку output:

output {
  elasticsearch {
    hosts => ["https://elasticsearch.example:9200"]
    document_id => "%{[@metadata][_id]}"
  }
}

Иначе стабильный ID потеряется на последнем участке pipeline, и Elasticsearch снова начнёт создавать документы с автоматически сгенерированными _id.

Что именно будет считаться дублем

У Filebeat нет временного окна. Если два события получают одинаковый _id и попадают в один и тот же конкретный индекс, второе заменяет первое независимо от того, прошло между ними пять секунд или пять месяцев.

Отсюда два разных сценария:

• ключ из настоящего event_id защищает от повторной доставки одного события;

• ключ из (service.name, log.level, message) начинает подавлять повторения сообщения, даже если это два реальных сбоя.

Второй вариант опаснее, чем кажется. Он не сохраняет repeat_count, не даёт временного окна и способен перезаписать полезное повторное событие. Кроме того, _id уникален только внутри конкретного индекса. При переходе на новый ежедневный одинаковый отпечаток снова создаст отдельный документ.

Ещё есть процессор add_id, который создаёт уникальный ID для события. Он помогает, когда Filebeat повторно отправляет то же событие из своей очереди: ID остаётся с ним. Но если строка будет заново прочитана после потери registry-файла (это хэш с сохраненной позицией чтения) или повторно записана приложением, add_id сгенерирует уже другое значение. Для сохранения идемпотентности нужен отпечатоклибо естественный ID из самого события.

Не путать с file identity fingerprint

У filestream также есть настройка file_identity.fingerprint. Она узнаёт файл по отпечатку участка его содержимого и защищает от проблем с ротацией, повторным использованием inode и переименованием файлов. Это идентичность файла, а не лог-записи. Она снижает риск повторного чтения, но не сравнивает сообщения между собой.

Что происходит с ресурсами

Такой подход почти не требует хранения чего бы то ни было в самом агенте: fingerprint вычисляется на лету, LRU-cache не растёт, а после рестарта получается тот же ID. Но это не бесплатная дедупликация до backend. Повтор всё равно проходит через Filebeat, сеть и ingest-пайплайн Elasticsearch; экономия возникает на уровне числа итоговых документов. Elasticsearch также платит за повторную операцию индексации.

Где преимущество

Filebeat особенно хорош, когда крайне важно не получить второй документ из-за повторной отправки, а конечное хранилище — Elasticsearch.

Проверяем на одинаковом потоке

Чтобы подтвердить все на практике, я собрал небольшое демо в Docker-контейнерах. Генератор отправил 100 000 NDJSON-записей. В них было 1000 логических сигнатур, каждая повторялась 100 раз. Поля timestamp, request_id и seq менялись в каждой записи, а ключ был одинаковым для всех реализаций:

(service, level, message)

Ожидаемый результат корректной дедупликации — 1000 записей, то есть сокращение количества событий на 99%.

Результаты

Самая полезная ошибка теста

В первом прогоне OTel я указал:

include_fields:
  - body.service
  - body.level
  - body.message

Но json_parser раскидал значения в attributes, а оригинальный JSON остался в body. В итоге меняющиеся request_id, timestamp и seq сделали уникальной каждую запись. На выходе остались все те же 100 000 записей, а память контейнера выросла до примерно 600 МБ. Получилось печально.

После замены путей на attributes.service, attributes.level и attributes.message результат стал ожидаемым: 1000 записей, сумма пролетевших записей 100 000, а сам контейнер около 84 МБ после прогона.

Неверный отпечаток способен одновременно убрать всю экономию и превратить дедупликацию в генератор новых записей.

Как вы можете протестить это у себя

Одного потока из одинаковых сообщений недостаточно, желательно прогнать тесты по сценариям ниже:

1. Повторяющееся количество отброшенных событий

Проверьте, сколько записей остаётся и сохраняется ли кратность. Для OTel сумма log_count должна совпасть с входом. Для Vector и Fluent Bit считайте отброшенные события.

2. Почти уникальный поток

Подайте логи с высокой кардинальностью ключа. Дедупликация не принесёт экономии, зато покажет верхнюю границу RAM и CPU. Именно этот сценарий обычно опаснее идеального шторма сообщений в логах.

3. Граница окна

Отправьте одинаковую запись до и после interval или max_age_ms. Убедитесь, что семантика сообщений соответствует ожиданиям. Особенно проверьте refresh_on_drop у Vector.

4. Рестарт агента

Отправьте событие, перезапустите агент и отправьте его снова. Кэш забудет прошлые данные. Это нормально, если вы используете дедупликацию как защиту от шторма событий, но недостаточно для гарантированной однократной доставки.

5. Несколько реплик

Распределите одинаковые сообщения между двумя агентами. Локальные кэши не видят друг друга, поэтому каждый пропустит свою первую копию. Если нужна глобальная дедупликация, её место — общий stateful-слой или backend с idempotency key.

И обязательно проверяйте ложные срабатывания. Потерять полезное событие из-за слишком грубого ключа хуже, чем сохранить лишние 100 МБ логов.

Что выбрать

Выбирайте OpenTelemetry Collector, когда важно считать количество повторов. Он единственный в этом сравнении даёт честную дедупликацию. Альфа-статус меня тоже смущает.

Vector — наиболее удобный готовый вариант для отбрасывания повторов. Конфигурация LRU прозрачна, transform стабилен, но учет кратности теряется.

Fluent Bit + Lua оправдан для решения локальной задачи (известного паттерна) при уже существующем парке агентов Fluent Bit. Рассматривать этот подход как универсальный я б не стал, но если вы тщательно все протестируете и убедитесь в стабильности, отчего ж нет.

vlagent можно выбирать если у вас VictoriaLogs. Для дедупликации на уровне сбора не обойтись без дополнительно уровня обработки. Ну или при запросе можно использовать uniq ... with hitsс полным пониманием, что в бэкэнде дубли все равно лежат.

В Grafana Alloy механизм дедупликации логов отсутствует. drop, limit и Explore Dedup полезны, но это фильтрация, ограничения при отправке и отображение соответственно.

Полезные ссылки

• OpenTelemetry Collector Contrib: Log DeDuplication Processor

• Vector: Dedupe transform

• Fluent Bit: filters

• Fluent Bit: Lua filter

• VictoriaLogs: vlagent

• VictoriaLogs LogsQL: uniq pipe

• Grafana Alloy: component reference

• Grafana Alloy: loki.process

• Grafana: logs in Explore

• Filebeat: deduplicate data

• Filebeat: fingerprint processor

• Filebeat: filestream input and file identity

Спасибо за внимание, надеюсь, статья была полезна. Если вам нужна консультация или помощь с проектированием observability-платформы, я всегда открыт к предложениям —пишите в телеграм (контакты в профиле).