Spaces:
Running
Running
File size: 10,576 Bytes
aa89cb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | import { context, ROOT_CONTEXT, defaultTextMapSetter } from '@opentelemetry/api';
import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
import { AlwaysOnSampler, BasicTracerProvider, BatchSpanProcessor, SpanProcessor } from '@opentelemetry/sdk-trace-base';
import { LoggerProvider, BatchLogRecordProcessor, LogRecordProcessor } from '@opentelemetry/sdk-logs';
import { ExportResult, W3CTraceContextPropagator } from '@opentelemetry/core';
import { AggregationTemporality, MeterProvider, PeriodicExportingMetricReader, ResourceMetrics } from '@opentelemetry/sdk-metrics';
import { ProtobufLogsSerializer, ProtobufMetricsSerializer, ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { OpenTelemetryRuntimeTelemetry } from './opentelemetry';
import { telemetryNotice } from '../core/telemetry-startup';
import { ExportHealth, lifecycleDeadline } from './export-health';
import { OwnedOTLPHTTP, ownedExporter } from './owned-otlp-http';
import { ResponseHealth } from './response-health';
function reportedExporter<T>(delegate: { export(records: T[], done: (result: ExportResult) => void): void;
shutdown(): Promise<void>; forceFlush?(): Promise<void> }, health: ExportHealth) {
return {
export: (records: T[], done: (result: ExportResult) => void) => {
const ticket = health.beginExport(records as object[]);
if (!ticket) {done({code: 0}); return;}
let completed = false;
const finish = (result: ExportResult) => {
if (completed) return;
completed = true;
ticket.complete(result.code !== 0);
done(result);
};
try { delegate.export(records, finish); }
catch { finish({ code: 1 }); }
},
shutdown: () => delegate.shutdown(),
forceFlush: () => delegate.forceFlush?.() ?? Promise.resolve(),
};
}
/** Isolated providers; never registers or replaces process-global OTel configuration. */
export function createStandardNodePipeline(serviceName: string, endpoint: string) {
const resource = resourceFromAttributes({ 'service.name': serviceName, 'teaql.runtime.language': 'typescript' });
const manager = new AsyncLocalStorageContextManager().enable();
const transport = new OwnedOTLPHTTP();
let lossCounter: ReturnType<import('@opentelemetry/api').Meter['createCounter']> | undefined;
let rejectedCounter: typeof lossCounter, responseCounter: typeof lossCounter;
const responses = new ResponseHealth((signal, reason, rejected) => {
rejectedCounter?.add(rejected, {'teaql.telemetry.signal': signal});
responseCounter?.add(1, {'teaql.telemetry.signal': signal, 'teaql.telemetry.reason': reason});
});
const metricPoints = (data: ResourceMetrics) => data.scopeMetrics.reduce((n, scope) =>
n + scope.metrics.reduce((m, metric) => m + metric.dataPoints.length, 0), 0);
const metricExporter = ownedExporter(transport, endpoint, 'metrics', ProtobufMetricsSerializer,
metricPoints, (signal, kind, rejected) => responses.observe(signal, kind, rejected));
let metricDataObjectsFailed = 0;
let metricFailureAnnounced = false;
const meterProvider = new MeterProvider({ resource, readers: [new PeriodicExportingMetricReader({
exporter: { export: (data, done) => {
let completed = false;
const finish = (result: ExportResult) => {
if (completed) return;
completed = true;
if (result.code !== 0) {
const objects = data.scopeMetrics.reduce((n, scope) => n + scope.metrics.length, 0);
metricDataObjectsFailed += objects;
lossCounter?.add(objects, {'teaql.telemetry.signal': 'metrics', 'teaql.telemetry.reason': 'export_failure'});
if (!metricFailureAnnounced) {
metricFailureAnnounced = true;
telemetryNotice('TeaQL telemetry: records lost; signal=metrics reason=export_failure. Business outcome unchanged.', true);
}
}
done(result);
};
try { metricExporter.export(data, finish); } catch { finish({ code: 1 }); }
}, forceFlush: () => metricExporter.forceFlush(), shutdown: () => metricExporter.shutdown(),
selectAggregationTemporality: () => AggregationTemporality.CUMULATIVE },
exportIntervalMillis: 10000, exportTimeoutMillis: 5000 })] });
const meter = meterProvider.getMeter('io.teaql.runtime');
lossCounter = meter.createCounter('teaql.telemetry.export.lost', {
description: 'Lost span/log records or failed metric-data objects; not metric data points', unit: '{entry}' });
rejectedCounter = meter.createCounter('teaql.telemetry.export.rejected', {
description: 'Receiver-rejected spans, log records or metric data points', unit: '{item}' });
responseCounter = meter.createCounter('teaql.telemetry.export.response', {unit: '{response}'});
for (const signal of ['traces', 'logs', 'metrics']) {
rejectedCounter.add(0, {'teaql.telemetry.signal': signal});
for (const reason of ['partial_success', 'warning', 'invalid_response'])
responseCounter.add(0, {'teaql.telemetry.signal': signal, 'teaql.telemetry.reason': reason});
lossCounter.add(0, {'teaql.telemetry.signal': signal, 'teaql.telemetry.reason': 'export_failure'});
}
const lost = (amount: number, signal: string, reason: string) => lossCounter!.add(amount,
{'teaql.telemetry.signal': signal, 'teaql.telemetry.reason': reason});
const spanHealth = new ExportHealth('traces', 2048, lost);
const logHealth = new ExportHealth('logs', 2048, lost);
const spanBatch = new BatchSpanProcessor(reportedExporter(ownedExporter(transport, endpoint, 'traces',
ProtobufTraceSerializer, spans => spans.length, (signal, kind, rejected) => responses.observe(signal, kind, rejected)), spanHealth), { maxQueueSize: 2048, maxExportBatchSize: 128,
scheduledDelayMillis: 1000, exportTimeoutMillis: 5000 });
const spanProcessor: SpanProcessor = {
onStart: (...args) => spanBatch.onStart(...args),
onEnd: span => spanHealth.handoff(span, () => spanBatch.onEnd(span)),
forceFlush: () => spanBatch.forceFlush(), shutdown: () => spanBatch.shutdown(),
};
const logBatch = new BatchLogRecordProcessor({
exporter: reportedExporter(ownedExporter(transport, endpoint, 'logs', ProtobufLogsSerializer,
logs => logs.length, (signal, kind, rejected) => responses.observe(signal, kind, rejected)), logHealth),
maxQueueSize: 2048, maxExportBatchSize: 128, scheduledDelayMillis: 1000, exportTimeoutMillis: 5000 });
const logProcessor: LogRecordProcessor = {
onEmit: record => logHealth.handoff(record, () => logBatch.onEmit(record)),
forceFlush: () => logBatch.forceFlush(), shutdown: () => logBatch.shutdown(),
};
const tracerProvider = new BasicTracerProvider({ resource, sampler: new AlwaysOnSampler(),
spanProcessors: [spanProcessor] });
const loggerProvider = new LoggerProvider({ resource, processors: [logProcessor] });
let closed = false;
let shutdownPromise: Promise<void> | undefined;
const flush = async () => {
if (closed) return shutdownPromise;
const spans = spanHealth.waitForPrefix(), logs = logHealth.waitForPrefix();
await lifecycleDeadline((async () => {
const records = await Promise.allSettled([tracerProvider.forceFlush(), loggerProvider.forceFlush(), spans, logs]);
if (closed) return;
await meterProvider.forceFlush(); // loss/rejection counters follow record completion
if (records.some(result => result.status === 'rejected')) throw new Error('TeaQL telemetry export incomplete');
})()); };
const shutdown = () => shutdownPromise ??= (async () => {
closed = true;
spanHealth.close(); logHealth.close();
const started = performance.now();
const stopped = Promise.allSettled([tracerProvider.shutdown(), loggerProvider.shutdown()]);
try {
try {await Promise.all([spanHealth.waitForPrefix(5000), logHealth.waitForPrefix(5000)]);}
catch {
spanHealth.abortQueued(); logHealth.abortQueued(); transport.abort();
await Promise.all([spanHealth.waitForIdle(200), logHealth.waitForIdle(200)]);
}
await lifecycleDeadline(stopped, Math.max(1, 5700 - (performance.now() - started)));
await lifecycleDeadline(meterProvider.shutdown(), Math.max(1, 5800 - (performance.now() - started)));
} finally {
spanHealth.abortQueued(); logHealth.abortQueued(); transport.close();
await lifecycleDeadline(Promise.all([transport.waitForIdle(200), metricExporter.forceFlush()]),
Math.max(1, 6000 - (performance.now() - started))).catch(() => {
telemetryNotice('TeaQL telemetry: owned cleanup deadline reached; business outcome unchanged.', true);
});
manager.disable();
}
})();
const executionContext = {
active: () => manager.active() === ROOT_CONTEXT ? context.active() : manager.active(),
with: <T>(active: import('@opentelemetry/api').Context, work: () => T): T => manager.with(active, work),
};
const adapter = new OpenTelemetryRuntimeTelemetry(tracerProvider.getTracer('io.teaql.runtime'),
meterProvider.getMeter('io.teaql.runtime'), {}, loggerProvider.getLogger('io.teaql.runtime'), executionContext);
const telemetry: import('../core/telemetry').RuntimeTelemetry = {
start: operation => {
const delegate = closed ? undefined : adapter.start(operation);
let finished = false;
const finish = (work: () => void) => {
if (finished) return; finished = true;
if (closed) {spanHealth.rejectLate(); logHealth.rejectLate();} else work();
};
return {correlation: () => closed ? undefined : delegate?.correlation?.(),
run: work => !closed && delegate?.run ? delegate.run(work) : work(),
success: completion => finish(() => delegate!.success(completion)),
failure: error => finish(() => delegate!.failure(error))};
},
recordsSQLDiagnostics: () => !closed,
recordDiagnostic: event => {if (!closed) adapter.recordDiagnostic(event);},
};
const propagator = new W3CTraceContextPropagator();
telemetry.inject = carrier => propagator.inject(executionContext.active(), carrier, defaultTextMapSetter);
telemetryNotice('TeaQL telemetry: enabled; traces, metrics, logs, complete TraceChain and masked SQL; recording=all.');
return { telemetry, flush, shutdown, health: () => ({traceSpansFailed: spanHealth.failedRecords,
logRecordsFailed: logHealth.failedRecords, metricDataObjectsFailed, responses: responses.snapshot(),
queues: {traces: spanHealth.snapshot(), logs: logHealth.snapshot()}}) };
}
|