diff --git a/benchmark/serialize.bench.js b/benchmark/serialize.bench.js new file mode 100644 index 0000000..162f67b --- /dev/null +++ b/benchmark/serialize.bench.js @@ -0,0 +1,130 @@ +'use strict'; + +const { before, bench } = require('node:bench'); +const microtel = require('../index'); +const { resetCaches } = require('../lib/flush'); +const { generateSpanId, generateTraceId } = require('../lib/id'); +const { serializeSpans } = require('../lib/serialize'); +const { + Span, + SPAN_KIND_CLIENT, + SPAN_KIND_SERVER, + STATUS_ERROR, +} = require('../lib/span'); + +// An export sends whatever is buffered when the flush timer fires, or +// maxBufferSize spans once the buffer fills: +// - 2: one request's spans, when traffic is low enough that the timer +// flushes after every request. +// - 20: several requests between timer flushes. +// - 100: the default maxBufferSize, the batch every busy process sends. +// - 512: the batching processor's maxExportBatchSize, for a process that +// raises maxBufferSize to match it. "The default value is `512`." +// https://opentelemetry.io/docs/specs/otel/trace/sdk/#batching-processor +// - 1024 and 2048: larger batches, for reference. +const kBatchSizes = [2, 20, 100, 512, 1024, 2048]; +const kMaxBatchSize = Math.max(...kBatchSizes); + +// Adds 2 spans, a server span, and a client span made while handling it. +// The values that vary per request in a real process vary here too. +function addRequestSpans(spans, i, error) { + const server = new Span('GET', SPAN_KIND_SERVER, { + parent: Span.fromTraceparent( + `00-${generateTraceId()}-${generateSpanId()}-01`, + ), + }); + server.setAttribute('http.request.method', 'GET'); + server.setAttribute('url.path', `/api/users/${1000 + i}/orders`); + server.setAttribute('url.query', `page=${i % 7}&limit=50`); + server.setAttribute('url.scheme', 'http'); + server.setAttribute('network.protocol.version', '1.1'); + server.setAttribute('server.address', 'api.example.com'); + server.setAttribute('server.port', 8080); + server.setAttribute('client.address', `10.0.${i % 16}.${i % 251}`); + server.setAttribute('network.peer.address', `10.0.${i % 16}.${i % 251}`); + server.setAttribute('network.peer.port', 32768 + i); + server.setAttribute('user_agent.original', 'Mozilla/5.0 (X11; Linux x86_64)'); + + const client = new Span('GET', SPAN_KIND_CLIENT, { parent: server }); + client.setAttribute('http.request.method', 'GET'); + client.setAttribute('server.address', 'orders.internal'); + client.setAttribute('server.port', 80); + client.setAttribute( + 'url.full', + `http://orders.internal/orders?user=${1000 + i}`, + ); + client.setAttribute('network.protocol.version', '1.1'); + + if (error == null) { + client.setAttribute('http.response.status_code', 200); + client.end(); + server.setAttribute('http.response.status_code', 200); + } else { + // What the client error handler records. + client.setAttribute('error.type', error.name); + client.setStatus(STATUS_ERROR, error.message); + client.addEvent('exception', { + 'exception.type': error.name, + 'exception.message': error.message, + 'exception.stacktrace': error.stack, + }); + client.end(); + server.setAttribute('http.response.status_code', 502); + server.setStatus(STATUS_ERROR, 'HTTP 502'); + server.setAttribute('error.type', '502'); + } + server.end(); + spans.push(server, client); +} + +const typical = []; +const failed = []; + +before(() => { + // Sets the resource attributes the payload carries. + microtel.init(); + // Exception messages and stack traces are where non-ASCII text, such as + // a localized message or a non-ASCII file path, usually reaches a span. + const error = new Error('connect ECONNREFUSED: 接続が拒否されました'); + // Each call adds a request's 2 spans. + for (let i = 0; i < kMaxBatchSize / 2; i++) { + addRequestSpans(typical, i); + addRequestSpans(failed, i, error); + } + // Drops the spans end() buffered, so they are never exported. + resetCaches(); +}); + +const onFailure = (message) => { + throw new Error(message); +}; + +// Splits the spans into consecutive batches, so every batch size covers +// all the spans. +function toBatches(spans, batchSize) { + const batches = []; + for (let i = 0; i + batchSize <= spans.length; i += batchSize) { + batches.push(spans.slice(i, i + batchSize)); + } + return batches; +} + +// Each sample serializes every batch once, about kMaxBatchSize spans +// whatever the batch size, so the per-span rates are comparable. +function registerSerializeBenchmark(name, spans) { + for (const batchSize of kBatchSizes) { + bench(name, { samples: 30, params: { batchSize } }, (b) => { + const batches = toBatches(spans, batchSize); + let bytes = 0; + b.start(); + for (const batch of batches) { + bytes += serializeSpans(batch, onFailure).length; + } + b.end(batches.length * batchSize); + if (bytes === 0) throw new Error('nothing serialized'); + }); + } +} + +registerSerializeBenchmark('serialize http spans', typical); +registerSerializeBenchmark('serialize failed http spans', failed); diff --git a/lib/flush.js b/lib/flush.js index c4ca388..61ad3f3 100644 --- a/lib/flush.js +++ b/lib/flush.js @@ -3,22 +3,13 @@ const { Buffer } = require('node:buffer'); const http = require('node:http'); const https = require('node:https'); -const { hostname } = require('node:os'); const { addAbortListener } = require('node:events'); const { clearTimeout, setImmediate, setTimeout } = require('node:timers'); const { guard } = require('./guard'); -const { - getEndpoint, - getExportHeaders, - getExportUrl, - getInstanceId, - getResourceAttributes, - getServiceName, - getSdkName, - getSdkVersion, -} = require('./core'); +const { getEndpoint, getExportHeaders, getExportUrl } = require('./core'); +const { resetSerializeCaches, serializeSpans } = require('./serialize'); const { validateAbortSignal, validateObject } = require('./validate'); const kDefaultMaxBufferSize = 100; @@ -26,11 +17,6 @@ const kDefaultFlushIntervalMs = 10_000; const kDefaultExportTimeoutMs = 10_000; const kWarningThrottleMs = 30_000; -// OTLP span flags hold the W3C trace flags in the low 8 bits. The next two -// bits say that whether the parent is remote is known, and that it is. The -// OpenTelemetry SDKs always know, so they always set the first. -const kSpanFlagsHasIsRemote = 0x100; -const kSpanFlagsIsRemote = 0x200; // One keep-alive socket per protocol: batches queue behind it, bounding // in-flight exports to a slow collector. const kExportAgents = { @@ -51,148 +37,6 @@ let lastExportWarningTime = 0; // later flush also waits for the exports an earlier one started. const inFlightExports = new Set(); -let cachedResource = null; -let cachedScope = null; - -// A user value replaces microtel's default for the same key, so each key -// appears once. User keys microtel does not set are appended. -function buildResource() { - const values = { - __proto__: null, - 'service.name': getServiceName(), - 'service.instance.id': getInstanceId(), - 'host.name': hostname(), - 'telemetry.sdk.name': getSdkName(), - 'telemetry.sdk.language': 'nodejs', - 'telemetry.sdk.version': getSdkVersion(), - 'process.runtime.name': 'nodejs', - 'process.runtime.version': process.version, - 'process.pid': process.pid, - }; - const userValues = getResourceAttributes(); - const keys = Object.keys(values); - if (userValues != null) { - for (const key of Object.keys(userValues)) { - values[key] = userValues[key]; - // Keys microtel also sets are already in the list. - if (!keys.includes(key)) keys.push(key); - } - } - const attributes = []; - for (let i = 0; i < keys.length; i++) { - attributes.push({ - key: keys[i], - value: encodeAttributeValue(values[keys[i]]), - }); - } - return { attributes }; -} - -function getResource() { - cachedResource ??= buildResource(); - return cachedResource; -} - -function getScope() { - cachedScope ??= { - name: getSdkName(), - version: getSdkVersion(), - }; - return cachedScope; -} - -function encodeAttributeValue(value) { - switch (typeof value) { - case 'string': - return { stringValue: value }; - // OTLP writes an int64 as a decimal string. Integers past the safe range - // print in exponent form or lose their exact value, so they export as - // doubles. - case 'number': - return Number.isSafeInteger(value) - ? { intValue: String(value) } - : { doubleValue: value }; - case 'boolean': - return { boolValue: value }; - default: - return { stringValue: String(value) }; - } -} - -// Converts epoch milliseconds into an OTLP nanosecond string. Rounding -// to whole milliseconds first keeps the product exactly representable in -// double precision. -const timeToUnixNano = (t) => `${BigInt(Math.round(t)) * 1_000_000n}`; - -function spanToOtlp(span) { - // TODO(bengl): A lot of objects are created in here for all the atributes. - // As a future optimization, we could hand-write the JSON encoding. - const rawAttrs = span.getAttributes(); - const attrKeys = Object.keys(rawAttrs); - const attributes = []; - for (let i = 0; i < attrKeys.length; i++) { - // Unused preallocated keys stay undefined and are skipped. - if (rawAttrs[attrKeys[i]] === undefined) continue; - attributes.push({ - key: attrKeys[i], - value: encodeAttributeValue(rawAttrs[attrKeys[i]]), - }); - } - - const rawEvents = span.getEvents(); - const events = []; - for (let i = 0; i < rawEvents.length; i++) { - const event = rawEvents[i]; - const eventAttrs = []; - const eventAttrKeys = Object.keys(event.attributes); - for (let j = 0; j < eventAttrKeys.length; j++) { - if (event.attributes[eventAttrKeys[j]] === undefined) continue; - eventAttrs.push({ - key: eventAttrKeys[j], - value: encodeAttributeValue(event.attributes[eventAttrKeys[j]]), - }); - } - const otlpEvent = { - name: event.name, - timeUnixNano: timeToUnixNano(event.time), - }; - if (eventAttrs.length > 0) otlpEvent.attributes = eventAttrs; - events.push(otlpEvent); - } - - const otlpSpan = { - traceId: span.traceId, - spanId: span.spanId, - flags: - span.traceFlags | - kSpanFlagsHasIsRemote | - (span.hasRemoteParent ? kSpanFlagsIsRemote : 0), - name: span.name, - kind: span.kind, - startTimeUnixNano: timeToUnixNano(span.startTime), - endTimeUnixNano: timeToUnixNano(span.endTime), - }; - - if (attributes.length > 0) otlpSpan.attributes = attributes; - - // The OTLP field is dropped_attributes_count, spelled - // droppedAttributesCount in ProtoJSON. - const droppedAttributesCount = span.droppedAttributesCount; - if (droppedAttributesCount > 0) - otlpSpan.droppedAttributesCount = droppedAttributesCount; - - const status = span.status; - if (status.code !== 0 || status.message) otlpSpan.status = status; - - if (span.parentSpanId) otlpSpan.parentSpanId = span.parentSpanId; - - if (span.tracestate) otlpSpan.traceState = span.tracestate; - - if (events.length > 0) otlpSpan.events = events; - - return otlpSpan; -} - function addSpan(span) { spanBuffer.push(span); if (spanBuffer.length >= maxBufferSize) { @@ -247,40 +91,8 @@ function exportBuffered() { const spans = spanBuffer; spanBuffer = []; - const otlpSpans = []; - for (let i = 0; i < spans.length; i++) { - try { - otlpSpans.push(spanToOtlp(spans[i])); - } catch (err) { - warnExportFailure( - `Failed to serialize span "${spans[i].name}": ${err.message}`, - ); - } - } - - if (otlpSpans.length === 0) return settleAll(pending); - - let payload; - try { - payload = JSON.stringify({ - resourceSpans: [ - { - resource: getResource(), - scopeSpans: [ - { - scope: getScope(), - spans: otlpSpans, - }, - ], - }, - ], - }); - } catch (err) { - warnExportFailure( - `Failed to serialize ${otlpSpans.length} spans: ${err.message}`, - ); - return settleAll(pending); - } + const payload = serializeSpans(spans, warnExportFailure); + if (payload === undefined) return settleAll(pending); pending.push(sendToCollector(payload)); return settleAll(pending); @@ -474,8 +286,7 @@ function resetCaches() { flushScheduled = false; flushIntervalMs = kDefaultFlushIntervalMs; exportTimeoutMs = kDefaultExportTimeoutMs; - cachedResource = null; - cachedScope = null; + resetSerializeCaches(); exportFailureCount = 0; lastExportWarningTime = 0; maxBufferSize = kDefaultMaxBufferSize; diff --git a/lib/serialize.js b/lib/serialize.js new file mode 100644 index 0000000..3dbc1bd --- /dev/null +++ b/lib/serialize.js @@ -0,0 +1,207 @@ +'use strict'; + +// Encodes spans as an OTLP/JSON export request body. + +const { hostname } = require('node:os'); + +const { + getInstanceId, + getResourceAttributes, + getServiceName, + getSdkName, + getSdkVersion, +} = require('./core'); + +// OTLP span flags hold the W3C trace flags in the low 8 bits. The next two +// bits say that whether the parent is remote is known, and that it is. The +// OpenTelemetry SDKs always know, so they always set the first. +const kSpanFlagsHasIsRemote = 0x100; +const kSpanFlagsIsRemote = 0x200; + +let cachedResource = null; +let cachedScope = null; + +// A user value replaces microtel's default for the same key, so each key +// appears once. User keys microtel does not set are appended. +function buildResource() { + const values = { + __proto__: null, + 'service.name': getServiceName(), + 'service.instance.id': getInstanceId(), + 'host.name': hostname(), + 'telemetry.sdk.name': getSdkName(), + 'telemetry.sdk.language': 'nodejs', + 'telemetry.sdk.version': getSdkVersion(), + 'process.runtime.name': 'nodejs', + 'process.runtime.version': process.version, + 'process.pid': process.pid, + }; + const userValues = getResourceAttributes(); + const keys = Object.keys(values); + if (userValues != null) { + for (const key of Object.keys(userValues)) { + values[key] = userValues[key]; + // Keys microtel also sets are already in the list. + if (!keys.includes(key)) keys.push(key); + } + } + const attributes = []; + for (let i = 0; i < keys.length; i++) { + attributes.push({ + key: keys[i], + value: encodeAttributeValue(values[keys[i]]), + }); + } + return { attributes }; +} + +function getResource() { + cachedResource ??= buildResource(); + return cachedResource; +} + +function getScope() { + cachedScope ??= { + name: getSdkName(), + version: getSdkVersion(), + }; + return cachedScope; +} + +function encodeAttributeValue(value) { + switch (typeof value) { + case 'string': + return { stringValue: value }; + // OTLP writes an int64 as a decimal string. Integers past the safe range + // print in exponent form or lose their exact value, so they export as + // doubles. + case 'number': + return Number.isSafeInteger(value) + ? { intValue: String(value) } + : { doubleValue: value }; + case 'boolean': + return { boolValue: value }; + default: + return { stringValue: String(value) }; + } +} + +// Converts epoch milliseconds into an OTLP nanosecond string. Rounding +// to whole milliseconds first keeps the product exactly representable in +// double precision. +const timeToUnixNano = (t) => `${BigInt(Math.round(t)) * 1_000_000n}`; + +function spanToOtlp(span) { + // TODO(bengl): A lot of objects are created in here for all the atributes. + // As a future optimization, we could hand-write the JSON encoding. + const rawAttrs = span.getAttributes(); + const attrKeys = Object.keys(rawAttrs); + const attributes = []; + for (let i = 0; i < attrKeys.length; i++) { + // Unused preallocated keys stay undefined and are skipped. + if (rawAttrs[attrKeys[i]] === undefined) continue; + attributes.push({ + key: attrKeys[i], + value: encodeAttributeValue(rawAttrs[attrKeys[i]]), + }); + } + + const rawEvents = span.getEvents(); + const events = []; + for (let i = 0; i < rawEvents.length; i++) { + const event = rawEvents[i]; + const eventAttrs = []; + const eventAttrKeys = Object.keys(event.attributes); + for (let j = 0; j < eventAttrKeys.length; j++) { + if (event.attributes[eventAttrKeys[j]] === undefined) continue; + eventAttrs.push({ + key: eventAttrKeys[j], + value: encodeAttributeValue(event.attributes[eventAttrKeys[j]]), + }); + } + const otlpEvent = { + name: event.name, + timeUnixNano: timeToUnixNano(event.time), + }; + if (eventAttrs.length > 0) otlpEvent.attributes = eventAttrs; + events.push(otlpEvent); + } + + const otlpSpan = { + traceId: span.traceId, + spanId: span.spanId, + flags: + span.traceFlags | + kSpanFlagsHasIsRemote | + (span.hasRemoteParent ? kSpanFlagsIsRemote : 0), + name: span.name, + kind: span.kind, + startTimeUnixNano: timeToUnixNano(span.startTime), + endTimeUnixNano: timeToUnixNano(span.endTime), + }; + + if (attributes.length > 0) otlpSpan.attributes = attributes; + + // The OTLP field is dropped_attributes_count, spelled + // droppedAttributesCount in ProtoJSON. + const droppedAttributesCount = span.droppedAttributesCount; + if (droppedAttributesCount > 0) + otlpSpan.droppedAttributesCount = droppedAttributesCount; + + const status = span.status; + if (status.code !== 0 || status.message) otlpSpan.status = status; + + if (span.parentSpanId) otlpSpan.parentSpanId = span.parentSpanId; + + if (span.tracestate) otlpSpan.traceState = span.tracestate; + + if (events.length > 0) otlpSpan.events = events; + + return otlpSpan; +} + +// A span that fails to serialize is skipped and reported through +// onFailure(message), so the rest of the batch still exports. Returns +// undefined when nothing is left to send. +function serializeSpans(spans, onFailure) { + const otlpSpans = []; + for (let i = 0; i < spans.length; i++) { + try { + otlpSpans.push(spanToOtlp(spans[i])); + } catch (err) { + onFailure(`Failed to serialize span "${spans[i].name}": ${err.message}`); + } + } + + if (otlpSpans.length === 0) return undefined; + + try { + return JSON.stringify({ + resourceSpans: [ + { + resource: getResource(), + scopeSpans: [ + { + scope: getScope(), + spans: otlpSpans, + }, + ], + }, + ], + }); + } catch (err) { + onFailure(`Failed to serialize ${otlpSpans.length} spans: ${err.message}`); + return undefined; + } +} + +// Resets per-run state for the test suite. +function resetSerializeCaches() { + cachedResource = null; + cachedScope = null; +} + +module.exports = { + resetSerializeCaches, + serializeSpans, +};