src/vs/platform/otel/node/otlp/localOtlpReceiver.ts
227 LOC · 209 covered · 18 uncovered · 40 ranges · 19 concepts · 19 introducers · 11 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
localOtlpReceiver.ts ×6
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import type { AddressInfo } from 'net';
7
>
import type * as http from 'http';
8
>
import { IDisposable, toDisposable } from '../../../../base/common/lifecycle.js';
9
>
import { ILogService } from '../../../log/common/log.js';
10
>
import { decodeExportTraceRequest, IDecodeResult } from './otlpJsonDecode.js';
11
>
import { IOtlpExportTraceServiceRequest, IOtlpExportTraceServiceResponse } from './otlpJsonTypes.js';
12
>
13
>
/** Path the OTLP/HTTP spec mandates for trace export. */
14
>
export const OTLP_TRACES_PATH = '/v1/traces';
15
>
16
>
/** Default request body cap, matching the collector's `confighttp` default. */
17
>
const DEFAULT_MAX_BODY_BYTES = 64 * 1024 * 1024;
18
>
19
>
/** Callbacks the receiver invokes for each accepted request. */
20
>
export interface IOtlpReceiverHandlers {
21
>
/** Invoked with the decoded spans for every successfully-parsed request. */
22
>
onSpans(result: IDecodeResult): void;
23
>
/**
24
>
* Invoked with the raw request body and content-type so the caller can
25
>
* forward the bytes to an upstream collector unchanged. Called before
26
>
* the receiver responds, but failures here MUST NOT affect the response.
27
>
*/
28
>
onForward?(body: Buffer, contentType: string): void;
29
>
}
30
>
31
>
export interface IOtlpReceiverOptions {
32
>
/**
33
>
* Cap on request body size. Anything larger gets HTTP 413. Defaults to
34
>
* 64 MiB, matching the OpenTelemetry Collector default.
35
>
*/
36
>
readonly maxBodyBytes?: number;
37
>
}
38
>
39
>
export interface ILocalOtlpHttpReceiver extends IDisposable {
40
>
/** Loopback URL clients should POST to (without the trailing `/v1/traces`). */
41
>
readonly baseUrl: string;
42
>
/** Ephemeral port chosen by the OS. */
43
>
readonly port: number;
44
>
}
45
>
46
>
/**
47
>
* Loopback OTLP/HTTP receiver. Listens on `127.0.0.1` at an OS-assigned
48
>
* ephemeral port. Accepts `POST /v1/traces` with `Content-Type: application/json`
49
>
* and forwards the parsed result to the caller's handlers.
50
>
*
51
>
* Only `application/json` is supported. `application/x-protobuf` is rejected
52
>
* with HTTP 415; we explicitly require the SDK to use OTLP/HTTP+JSON for the
53
>
* loopback path.
54
>
*
55
>
* Returns `200 OK` with an empty body on full success, or with a
56
>
* `{ partialSuccess: { rejectedSpans, errorMessage } }` body when the
57
>
* decoder dropped some spans. Per OTLP spec, partial success is NOT a retry
58
>
* signal — the SDK will move on.
59
>
*/
60
>
export async function startLocalOtlpHttpReceiver(
61
>
handlers: IOtlpReceiverHandlers,
62
>
logService: ILogService,
63
>
options: IOtlpReceiverOptions = {},
64
>
): Promise<ILocalOtlpHttpReceiver> {
65
>
const maxBodyBytes = options.maxBodyBytes ?? DEFAULT_MAX_BODY_BYTES;
66
>
const httpModule = await import('http');
67
>
const server = httpModule.createServer();
68
>
69
>
server.on('request', (req, res) => {
70
>
handleRequest(req, res, handlers, logService, maxBodyBytes).catch(err => {
localOtlpReceiver.ts ×6
71
logService.error(`[agentHost-otel] receiver: unhandled error: ${err instanceof Error ? err.message : String(err)}`);
72
if (!res.headersSent) {
73
writePlain(res, 500, 'internal error');
74
} else if (!res.writableEnded) {
75
try { res.end(); } catch { /* ignore */ }
76
}
79
>
80
>
await new Promise<void>((resolve, reject) => {
81
>
const onError = (err: Error) => reject(err);
82
>
server.once('error', onError);
83
>
server.listen(0, '127.0.0.1', () => {
84
>
server.removeListener('error', onError);
85
>
resolve();
86
>
});
87
>
});
88
>
89
>
const address = server.address();
90
>
if (!address || typeof address === 'string') {
91
server.close();
92
throw new Error(`local OTLP receiver failed to bind: unexpected address ${String(address)}`);
93
}
95
>
const baseUrl = `http://127.0.0.1:${port}`;
96
>
logService.info(`[agentHost-otel] receiver listening on ${baseUrl}`);
97
>
98
>
const disposable = toDisposable(() => {
99
>
server.closeAllConnections();
100
>
server.close(err => {
101
>
if (err) {
102
logService.warn(`[agentHost-otel] receiver close error: ${err.message}`);
103
}
105
>
});
106
>
107
>
return Object.assign(disposable, { baseUrl, port });
108
>
}
109
>
111
>
req: http.IncomingMessage,
112
>
res: http.ServerResponse,
113
>
handlers: IOtlpReceiverHandlers,
114
>
logService: ILogService,
115
>
maxBodyBytes: number,
116
>
): Promise<void> {
117
>
// Reject anything that isn't the trace export path.
118
>
const url = req.url ?? '';
119
>
const pathname = url.split('?', 1)[0];
120
>
if (pathname !== OTLP_TRACES_PATH) {
122
>
return;
123
>
}
126
>
writePlain(res, 405, 'method not allowed');
127
>
return;
128
>
}
130
>
const contentType = (req.headers['content-type'] ?? '').toString().toLowerCase();
localOtlpReceiver.ts ×6
131
>
if (!contentType.includes('application/json')) {
132
>
writePlain(res, 415, 'unsupported content-type; this receiver only accepts application/json');
localOtlpReceiver.ts ×1
133
>
return;
134
>
}
136
>
const encoding = (req.headers['content-encoding'] ?? '').toString().toLowerCase();
localOtlpReceiver.ts ×6
137
>
if (encoding && encoding !== 'identity') {
138
>
// Compression negotiation is out of scope for v1; let the SDK fall back to identity.
localOtlpReceiver.ts ×1
139
>
writePlain(res, 415, `unsupported content-encoding: ${encoding}`);
140
>
return;
141
>
}
143
>
let body: Buffer;
144
>
try {
145
>
body = await readBody(req, maxBodyBytes);
146
>
} catch (err) {
148
>
writePlain(res, 413, 'payload too large');
149
>
} else {
150
writePlain(res, 400, 'failed to read body');
151
}
153
>
}
155
>
// Best-effort forward of raw bytes BEFORE decoding so the upstream
156
>
// collector sees an identical payload to what the SDK emitted. Failures
157
>
// here are isolated from the local-decode path.
158
>
if (handlers.onForward) {
160
>
handlers.onForward(body, contentType);
161
>
} catch (err) {
162
>
logService.warn(`[agentHost-otel] forward callback threw: ${err instanceof Error ? err.message : String(err)}`);
localOtlpReceiver.ts ×1
163
>
}
166
>
let parsed: IOtlpExportTraceServiceRequest;
167
>
try {
168
>
parsed = JSON.parse(body.toString('utf8')) as IOtlpExportTraceServiceRequest;
169
>
} catch (err) {
170
>
writePlain(res, 400, `invalid json: ${err instanceof Error ? err.message : String(err)}`);
localOtlpReceiver.ts ×1
171
>
return;
172
>
}
174
>
const result = decodeExportTraceRequest(parsed);
175
>
try {
176
>
handlers.onSpans(result);
177
>
} catch (err) {
178
logService.warn(`[agentHost-otel] onSpans handler threw: ${err instanceof Error ? err.message : String(err)}`);
179
}
181
>
const responseBody: IOtlpExportTraceServiceResponse = result.rejected > 0
182
>
? { partialSuccess: { rejectedSpans: result.rejected, errorMessage: result.errors.join('; ').slice(0, 1024) } }
localOtlpReceiver.ts ×1
185
>
res.setHeader('content-type', 'application/json');
186
>
res.end(JSON.stringify(responseBody));
187
>
}
189
>
class PayloadTooLargeError extends Error { }
190
>
191
>
function readBody(req: http.IncomingMessage, maxBytes: number): Promise<Buffer> {
localOtlpReceiver.ts ×5
192
>
return new Promise((resolve, reject) => {
193
>
const chunks: Buffer[] = [];
194
>
let received = 0;
195
>
const onData = (chunk: Buffer) => {
196
>
received += chunk.length;
197
>
if (received > maxBytes) {
199
>
reject(new PayloadTooLargeError(`body exceeds ${maxBytes} bytes`));
200
>
return;
201
>
}
204
>
const onEnd = () => {
206
>
resolve(Buffer.concat(chunks));
207
>
};
209
cleanup();
210
reject(err);
211
};
213
>
req.removeListener('data', onData);
214
>
req.removeListener('end', onEnd);
215
>
req.removeListener('error', onError);
216
>
};
217
>
req.on('data', onData);
218
>
req.on('end', onEnd);
219
>
req.on('error', onError);
220
>
});
221
>
}
223
>
function writePlain(res: http.ServerResponse, status: number, message: string): void {
localOtlpReceiver.ts ×1
224
>
res.statusCode = status;
225
>
res.setHeader('content-type', 'text/plain; charset=utf-8');
226
>
res.end(message);
227
>
}