OpenTelemetry Collector и OTLP

До этого сигналы «Пульса» выводились локально. Добавим OpenTelemetry Collector как отдельный процесс приёма и маршрутизации. Приложение отправит traces, metrics и logs по OTLP HTTP на один endpoint, а Collector разделит их по pipelines. На этом шаге exporter будет диагностическим debug; настоящие хранилища появятся в следующей главе. Такой промежуточный слой позволяет рассмотреть доставку отдельно от поиска и долговременного хранения.

Collector не является ещё одной базой телеметрии. Он принимает данные, применяет processors и передаёт exporters. Буферы и очереди помогают перенести кратковременный отказ, но не превращают процесс в историческое хранилище автоматически. Если backend отсутствует или данные только печатаются, Grafana не получает готовую историю запросов. Это различие важно для объяснения, почему «Collector отвечает» не равнозначно «traces сохранены».

Receiver, processor и exporter

Receiver OTLP HTTP слушает порт 4318 внутри контейнера. Явно заданный 0.0.0.0 нужен, чтобы приложение с хоста лаборатории могло обратиться к опубликованному loopback-порту. Публикация в Compose ограничена 127.0.0.1, поэтому лабораторный endpoint не объявлен публичным. Для реальной сети требуется TLS и аутентификация, поскольку OTLP payload содержит операционные сведения и способен нагружать Collector.

Полный файл collector.yaml состояния 07:

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 192
    spike_limit_mib: 48
  batch:
    timeout: 1s
exporters:
  debug:
    verbosity: basic
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [debug]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [debug]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [debug]

Pipeline объединяет receiver, processors и exporters для конкретного сигнала. Одинаковое имя receiver можно использовать в трёх pipelines, но каждый маршрут перечисляется отдельно. Если настроить только traces, metrics и logs сами не получат путь доставки. Поддержка протокола в binary ещё не означает, что его компонент активирован в service configuration; именно pipelines делают выбранные связи действующими.

memory_limiter стоит первым. Он помогает ограничить потребление памяти и создаёт backpressure при превышении заданного бюджета. batch объединяет данные перед отправкой, уменьшая число небольших сетевых запросов. Порядок processors имеет значение: если сначала накопить большие batch, а ограничение применить позже, система может расходовать память до защитного действия. При этом лимиты выбирают под реальные ресурсы процесса, а не считают любую цифру универсальной.

OTLP exporter приложения

В telemetry.mjs trace exporter получает /v1/traces, metric exporter — /v1/metrics, logs exporter — /v1/logs. Общий endpoint задаётся в окружении, но пути добавляются явно. Это предотвращает неоднозначность между базовым адресом и адресом конкретного сигнала. Для другого протокола, например OTLP gRPC, нужны другие exporter и настройка порта; простая замена числа 4318 на 4317 не переключает формат HTTP автоматически.

Логи используют отдельный LoggerProvider. Для закреплённого sdk-logs 0.200.0 процессор добавляется через addLogRecordProcessor(new BatchLogRecordProcessor(exporter)). Актуальные примеры более новых выпусков могут показывать массив processors в конструкторе. Мы сверили определение выбранной версии: копирование нового API в старый pin могло бы дать код, который выглядит правдоподобно, но не настроит экспорт правильно.

Полный файл telemetry.mjs состояния 07:

import { context,trace,propagation,SpanStatusCode } from '@opentelemetry/api';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { W3CTraceContextPropagator } from '@opentelemetry/core';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { ParentBasedSampler,TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
const endpoint=process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4318';
const traceExporter=new OTLPTraceExporter({url:endpoint+'/v1/traces'});
import { metrics } from '@opentelemetry/api';
import { PeriodicExportingMetricReader,ConsoleMetricExporter,AggregationType } from '@opentelemetry/sdk-metrics';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { LoggerProvider,BatchLogRecordProcessor } from '@opentelemetry/sdk-logs';
import { SeverityNumber } from '@opentelemetry/api-logs';
import { OTLPLogExporter } from '@opentelemetry/exporter-logs-otlp-http';
const service=process.env.OTEL_SERVICE_NAME || 'pulse-api';
const resource=resourceFromAttributes({'service.name':service,'service.version':'1.0.0',
  'deployment.environment.name':'lab'});
const ratio=Number(process.env.TRACE_RATIO || 1);
if(!Number.isFinite(ratio) || ratio<0 || ratio>1) throw new Error('Invalid TRACE_RATIO');
const sdk=new NodeSDK({resource,autoDetectResources:false,traceExporter,metricReader:new PeriodicExportingMetricReader({exporter:new OTLPMetricExporter({url:endpoint+'/v1/metrics'}),exportIntervalMillis:5000}),
  views:[{instrumentName:'pulse_request_duration',aggregation:{type:AggregationType.EXPLICIT_BUCKET_HISTOGRAM,options:{boundaries:[0.005,0.01,0.025,0.05,0.1,0.25,0.5,1,2.5,5]}}}],
  sampler:new ParentBasedSampler({root:new TraceIdRatioBasedSampler(ratio)}),
  textMapPropagator:new W3CTraceContextPropagator()});
sdk.start();
const tracer=trace.getTracer('pulse','1.0.0');
export async function inSpan(name,options,work,parent=context.active()) {
  return tracer.startActiveSpan(name,options,parent,async span=>{
    try { return await work(); }
    catch(error) { span.setStatus({code:SpanStatusCode.ERROR});
      span.addEvent('failure',{'error_code':error.code==='57014'?'db_timeout':'operation_failed'});
      throw error;
    } finally { span.end(); }
  });
}
export function currentCarrier(){const carrier={};propagation.inject(context.active(),carrier);return carrier;}
export function extracted(carrier){return propagation.extract(context.active(),carrier || {});}
const logs=new LoggerProvider({resource});
logs.addLogRecordProcessor(new BatchLogRecordProcessor(new OTLPLogExporter({url:endpoint+'/v1/logs'})));
const logger=logs.getLogger('pulse','1.0.0');
import { safeAttributes } from './attributes.mjs';
const allowed=new Set(['route','status_class','error_code','outcome','attempt']);
export function event(name,attributes={}) {
  const safe=safeAttributes(attributes);
  const spanContext=trace.getSpan(context.active())?.spanContext();
  console.log(JSON.stringify({time:new Date().toISOString(),event:name,...safe,
    trace_id:spanContext?.traceId,span_id:spanContext?.spanId}));
  logger.emit({context:context.active(),severityNumber:SeverityNumber.INFO,body:name,attributes:safe});
}
const meter=metrics.getMeter('pulse','1.0.0');
const requests=meter.createCounter('pulse_requests');
const duration=meter.createHistogram('pulse_request_duration',{unit:'s'});
meter.createObservableGauge('pulse_process_up').addCallback(result=>result.observe(1,{service}));
let oldest=0;
meter.createObservableGauge('pulse_queue_oldest',{unit:'s'}).addCallback(result=>result.observe(oldest,{service}));
export function observeRequest(route,status,seconds) {
  const labels={service,route,status_class:Math.floor(status/100)+'xx'};
  requests.add(1,labels);duration.record(seconds,labels);
}
export function queueAge(seconds){oldest=seconds;}
export async function shutdown(){await sdk.shutdown();await logs.shutdown();}

Событие передаётся logger с активным context. Благодаря этому OTLP LogRecord связан с trace ID и span ID. JSON stdout также печатает эти идентификаторы для локального чтения. Мы не устанавливаем одновременно сбор stdout в тот же backend, чтобы не получить два одинаковых события. Если инфраструктура позже добавит такой путь, нужно выбрать один канал либо специально обеспечить дедупликацию.

Буферы, повторы и пределы

Batch exporter не отправляет каждое событие синхронно в HTTP-запросе пользователя. Это снижает зависимость прикладной задержки от Collector, но создаёт окно потери при остановке процесса. Shutdown SDK и LoggerProvider пытается отправить оставшиеся данные. Принудительное завершение может оборвать отправку. Следовательно, телеметрию нельзя использовать как единственный журнал принятия задания; PostgreSQL сохраняет прикладной факт независимо.

Collector также может отказать при переполнении памяти или недоступном downstream. Сигнал отказа нужно наблюдать отдельно от HTTP 5xx «Пульса». Пользовательский API может продолжать обслуживать, пока exporter теряет данные. Дашборд в такой ситуации не должен утверждать отсутствие ошибок просто потому, что ошибок не видно. Отсутствие сигнала и нулевое значение — разные состояния, которые далее разделим в правилах оповещения.

Для промышленной передачи можно настроить persistent sending queue, retries и disk storage extension. В текущем конфиге они не объявлены как готовая долговечная доставка. Поддержка этих компонентов зависит от сборки Collector и правил конкретного exporter. Перед добавлением нужно проверить лимиты диска, время повторов и поведение при отказе, а не обещать «без потерь» по наличию одного процесса между приложением и backend.

Что покажет debug

Exporter debug с verbosity: basic сообщает о принятых и экспортируемых данных без полного подробного payload. Это полезно для первичного подтверждения маршрута, но не даёт поиска по вчерашнему trace ID. Более подробный режим может показать чувствительные данные, если приложение их соберёт, поэтому его включают в контролируемой среде и на ограниченное время. Фильтрация в приложении остаётся главной границей.

Ожидаемый результат седьмого состояния — три pipelines принимают OTLP от API и worker, а debug сообщает о соответствующих сигналах. Мы не приводим реальный вывод Collector: контейнер не запускался. Для будущей лаборатории проверяют адреса из двух сетевых пространств: Node на хосте использует localhost, а контейнеры следующего урока используют service names внутренней сети Compose.

Закреплён otel/opentelemetry-collector-contrib:0.123.0. Выбрана contrib-сборка с нужными processors и exporters; название версии не является рекомендацией текущего безопасного образа. Перед настоящим выпуском требуется согласованное обновление и проверка конфигурации. Теги удобны для урока, а строгая воспроизводимость инфраструктуры дополнительно фиксирует digest образа после собственной подготовки среды.

Теперь у сигналов есть явный маршрут доставки, а ошибка пути может быть локализована до хранилища. Следующая глава заменит debug на Tempo, Loki и Prometheus и подключит Grafana к этим отдельным источникам. Сохраните различие обязанностей: приложение создаёт факт, Collector переносит его, backend хранит, интерфейс выполняет запрос. Тогда пустая панель станет вопросом к цепочке, а не загадкой одного инструмента.

Предыдущий урок: Кардинальность и безопасные атрибуты · Следующий урок: Tempo, Loki и Prometheus

Оглавление серии · Исходники всех состояний.

Документация: Collector configuration, Collector 0.123.0.