Передача контекста фоновой задаче

API завершает ответ до обработки отчёта, поэтому worker выполняется в другом процессе и позже по времени. Активная память контекста сервера там отсутствует. Добавим явную передачу W3C trace context через строку очереди. Producer сохранит ограниченный carrier рядом с заданием, consumer извлечёт его перед созданием span job.process. Тогда один trace сможет показать API, запись в PostgreSQL и фоновую обработку.

Контекст не переносит пользовательские данные. В лаборатории используется только W3C Trace Context propagator: traceparent и при необходимости tracestate. Baggage не включён. Если переносить все заголовки или произвольный JSON с данными заявки, очередь превращается в ещё одно место хранения секретов и персональных сведений. Для диагностики нужен идентификатор и флаги трассировки, а не копия HTTP-запроса.

Inject в producer

После входа в SERVER-span функция currentCarrier() вызывает propagation.inject() с активным контекстом. Получается небольшой объект с полями стандартного формата. Он записывается вместе с UUID в INSERT pulse_jobs. Поскольку в нашем примере задание — сама строка PostgreSQL, данные задания и его контекст фиксируются одной операцией. Нет отдельного сообщения брокеру, опубликованного до COMMIT.

Полный файл server.mjs состояния 04:

import './telemetry.mjs';
import { trace } from '@opentelemetry/api';
import { createServer } from 'node:http';
import { randomUUID } from 'node:crypto';
import { performance } from 'node:perf_hooks';
import { query,pool } from './db.mjs';
import { inSpan,event,currentCarrier,extracted,observeRequest,shutdown } from './telemetry.mjs';
if(!process.env.PULSE_TOKEN || process.env.PULSE_TOKEN.length<32) throw new Error('Set PULSE_TOKEN');
const server=createServer(async(req,res)=>{
  const path=new URL(req.url,'http://localhost').pathname;
  const known=req.method==='POST' && path==='/v1/reports';
  const route=known?'/v1/reports':path==='/health'?'/health':'unmatched';
  const start=performance.now();let status=200;
  await inSpan(req.method+' '+route,{kind:1,attributes:{'http.route':route,
    'http.request.method':req.method}},async()=>{
    try {
      if(path==='/health' && req.method==='GET') return respond(200,{service:'pulse'});
      if(req.headers.authorization!=='Bearer '+process.env.PULSE_TOKEN) return respond(401,{error:'authentication_required'});
      if(!known) return respond(404,{error:'route_not_found'});
      let bytes=0;for await(const chunk of req){bytes+=chunk.length;if(bytes>4096)return respond(413,{error:'body_too_large'});}
      const id=randomUUID();
      await query('INSERT','INSERT INTO pulse_jobs(id,carrier) VALUES($1,$2)',[id,currentCarrier()]);
      event('report.accepted',{route,outcome:'queued'});
      const active=trace.getActiveSpan();
      if(active)res.setHeader('X-Trace-Id',active.spanContext().traceId);
      respond(202,{id,state:'queued'});
    } catch(error) { event('request.failed',{route,error_code:error.code==='57014'?'db_timeout':'operation_failed'});
      respond(503,{error:'operation_failed'});throw error;
    } finally { trace.getActiveSpan()?.setAttribute('http.response.status_code',status);
      observeRequest(route,status,(performance.now()-start)/1000);
      event('request.completed',{route,status_class:Math.floor(status/100)+'xx'}); }
  },extracted({'traceparent':req.headers.traceparent,'tracestate':req.headers.tracestate})).catch(()=>{});
  function respond(code,body){status=code;res.writeHead(code,{'Content-Type':'application/json','Cache-Control':'no-store'});res.end(JSON.stringify(body));}
});
server.listen(Number(process.env.PORT||8090),'127.0.0.1');
process.on('SIGTERM',()=>server.close(async()=>{await pool.end();await shutdown();}));

Trace ID сам по себе недостаточен для создания корректного родительства. traceparent включает версию формата, trace ID, parent span ID и flags. Поэтому мы используем propagator, а не склеиваем строку вручную. SDK проверяет форму извлечённого контекста и создаёт новый trace, если корректного родителя нет. Пользовательский заголовок не становится доказательством авторизации и не влияет на владение заданием.

На границе HTTP внешний посетитель может прислать свои trace headers. В закрытой лаборатории это помогает связать клиентскую операцию; в публичном сервисе нужно установить доверенную границу. Например, внешний proxy может удалить недоверенные заголовки или начинать новый trace с ограниченной связью. Parent-based sampling способен учитывать присланный sampled flag, поэтому слепое доверие внешнему контексту влияет на стоимость наблюдения.

Extract в consumer

Worker получает carrier из выбранной строки и вызывает extracted() перед inSpan('job.process'). Созданный CONSUMER-span использует извлечённый parent, а дочерний report.compute и завершающий SQL UPDATE работают внутри нового активного контекста. События job.started и job.completed получают trace ID и span ID worker. Локальная переменная из API здесь не участвует: связь пришла через данные очереди.

Полный файл worker.mjs состояния 04:

import './telemetry.mjs';
import { randomUUID } from 'node:crypto';
import { setTimeout as pause } from 'node:timers/promises';
import { query,pool } from './db.mjs';
import { inSpan,event,extracted,queueAge,shutdown } from './telemetry.mjs';
let stop=false;process.on('SIGTERM',()=>{stop=true;});
while(!stop) {
  const age=await query('SELECT',"SELECT COALESCE(EXTRACT(EPOCH FROM now()-min(created_at)),0) AS seconds FROM pulse_jobs WHERE state='queued'");
  queueAge(Number(age.rows[0].seconds));
  const token=randomUUID();
  const claim=await query('UPDATE',`UPDATE pulse_jobs SET state='processing',attempts=attempts+1,
    lease_token=$1,lease_until=now()+interval '30 seconds' WHERE id=(SELECT id FROM pulse_jobs WHERE
    (state='queued' AND available_at<=now()) OR (state='processing' AND lease_until<now())
    ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`,[token]);
  if(!claim.rowCount){await pause(1000);continue;}
  const row=claim.rows[0];
  try {
    await inSpan('job.process',{kind:4,attributes:{'job.type':'report'}},async()=>{
      event('job.started',{attempt:row.attempts});
      await inSpan('report.compute',{attributes:{'report.format':'summary'}},async()=>{
        await pause(100); // Учебная операция; это не измерение фактической задержки сервиса.
      });
      const finished=await query('UPDATE',"UPDATE pulse_jobs SET state='done',completed_at=now(),lease_token=NULL WHERE id=$1 AND lease_token=$2",[row.id,token]);
      event('job.completed',{outcome:finished.rowCount?'done':'lease_lost'});
    },extracted(row.carrier));
  } catch {
    await query('UPDATE',`UPDATE pulse_jobs SET state=CASE WHEN attempts>=3 THEN 'failed' ELSE 'queued' END,
      available_at=now()+interval '5 seconds',lease_token=NULL WHERE id=$1 AND lease_token=$2`,[row.id,token]);
  }
}
await pool.end();await shutdown();

Время до начала consumer-span означает ожидание между принятием и выбором задачи. Оно не является длительностью report.compute. Trace помогает визуально различить эти интервалы, но для общего возраста очереди нужна метрика следующего урока. Искать тысячи отдельных traces, чтобы оценить задолженность всех задач, неудобно и дорого. Точный случай и совокупное состояние очереди требуют разных сигналов.

SQL claim worker выполняется до извлечения конкретного carrier, потому что задача ещё неизвестна. Поэтому его span может оказаться отдельным trace. Это ожидаемая граница ручной инструментации, а не потеря связи уже выбранной работы. После claim можно добавить span link к потреблению, если требуется подробная модель broker/queue; в лаборатории яснее держать выбор и обработку раздельно.

Повторы и длительные задания

Carrier сохраняется при повторе. Если задача отказала и снова выбрана, новый consumer-span имеет тот же trace ID, но другой span ID. Это позволяет увидеть несколько попыток одной операции. Номер попытки допускается в логах как число, но не добавляется в метки общих HTTP-метрик. Отдельный UUID задания также не становится label. Корреляция не должна увеличивать число временных серий без границ.

После долгого ожидания trace может быть разделён по блокам хранения и появиться в поиске с задержкой. Backend должен хранить данные достаточно долго, чтобы producer и consumer имели шанс встретиться в одном интервале поиска. Если задача выполняется через неделю, сутки retention лаборатории недостаточны. Нельзя считать передачу контекста полной гарантией доступности всей истории при любом сроке выполнения.

Для очень длительных или независимых повторов иногда полезнее новый trace с span links на исходный контекст. Родительская связь означает непрерывную причинную операцию; link допускает связь нескольких независимых операций. Наш одноступенчатый отчёт использует parent, чтобы проще видеть путь. При переходе к пакетам из многих сообщений модель надо выбрать осознанно, иначе один случайно огромный trace станет неудобным и дорогим.

Очередь и право завершения

Lease и token управляют задачей независимо от trace context. Trace ID помогает наблюдать попытку, но не даёт worker права фиксировать результат. UPDATE требует собственный lease token. Если старый consumer завершился после нового claim, его span может показать выполненную работу, а строка базы не изменится. Поэтому событие «вычисление закончено» нужно отличать от факта принятого результата в PostgreSQL.

В текущем worker количество изменённых строк завершающего UPDATE определяет outcome. Нулевой результат приводит к безопасному событию lease_lost, а успешная фиксация — к done. Это сохраняет границу между вычислением и принятым результатом. Наблюдаемость сама по себе не исправляет протокол очереди: она должна честно отражать факт, который установила база, а не подменять его возвращением локальной функции.

Ожидаемая структура четвёртого состояния — SERVER-span API, SQL INSERT, позже job.process, внутри него report.compute и UPDATE. Специально заданная учебная пауза обозначает участок вычисления и не представлена как измерение работы отчётного движка. Контейнеры и процессы не запускались, поэтому сопоставление этой структуры выполняется по исходникам; реальный поиск trace предстоит проверить в подготовленной лаборатории.

Теперь распределённая связь имеет конкретный переносимый формат. Следующий урок введёт метрики, которые остаются полезными даже при выборке traces: число запросов, длительность HTTP и возраст очереди. Перед чтением полезно объяснить, почему carrier содержит только контекст и почему его сохранение в строке задания не заменяет lease и проверку успешной фиксации результата.

Предыдущий урок: Spans запросов API и PostgreSQL · Следующий урок: Метрики нагрузки и задержки

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

Документация: Context propagation, W3C Trace Context.