feat(otlp-exporter-base): support CONNECT proxies

open-telemetry · Oct 9, 2024 · ac5a4b8 · ac5a4b8
1 parent 6ccd4df
commit ac5a4b8
Show file tree

Hide file tree

Showing 7 changed files with 109 additions and 11 deletions.
diff --git a/experimental/packages/otlp-exporter-base/src/configuration/otlp-http-configuration.ts b/experimental/packages/otlp-exporter-base/src/configuration/otlp-http-configuration.ts
@@ -50,7 +50,9 @@ function mergeHeaders(
   return Object.assign(headers, requiredHeaders);
 }
 
-function validateUserProvidedUrl(url: string | undefined): string | undefined {
+export function validateUserProvidedUrl(
+  url: string | undefined
+): string | undefined {
   if (url == null) {
     return undefined;
   }

diff --git a/experimental/packages/otlp-exporter-base/src/platform/node/OTLPExporterNodeBase.ts b/experimental/packages/otlp-exporter-base/src/platform/node/OTLPExporterNodeBase.ts
@@ -26,6 +26,7 @@ import { convertLegacyAgentOptions } from './convert-legacy-agent-options';
 import {
   getHttpConfigurationDefaults,
   mergeOtlpHttpConfigurationWithDefaults,
+  validateUserProvidedUrl,
 } from '../../configuration/otlp-http-configuration';
 import { getHttpConfigurationFromEnvironment } from '../../configuration/otlp-http-env-configuration';
 
@@ -75,6 +76,7 @@ export abstract class OTLPExporterNodeBase<
         compression: actualConfig.compression,
         headers: actualConfig.headers,
         url: actualConfig.url,
+        proxy: validateUserProvidedUrl(config.proxy),
       }),
     });
   }

diff --git a/experimental/packages/otlp-exporter-base/src/platform/node/http-exporter-transport.ts b/experimental/packages/otlp-exporter-base/src/platform/node/http-exporter-transport.ts
@@ -40,10 +40,7 @@ class HttpExporterTransport implements IExporterTransport {
         createHttpAgent,
         // eslint-disable-next-line @typescript-eslint/no-var-requires
       } = require('./http-transport-utils');
-      this._agent = createHttpAgent(
-        this._parameters.url,
-        this._parameters.agentOptions
-      );
+      this._agent = createHttpAgent(this._parameters);
       this._send = sendWithHttp;
     }
 

diff --git a/experimental/packages/otlp-exporter-base/src/platform/node/http-transport-types.ts b/experimental/packages/otlp-exporter-base/src/platform/node/http-transport-types.ts
@@ -31,4 +31,5 @@ export interface HttpRequestParameters {
   headers: Record<string, string>;
   compression: 'gzip' | 'none';
   agentOptions: http.AgentOptions | https.AgentOptions;
+  proxy?: string;
 }
diff --git a/experimental/packages/otlp-exporter-base/src/platform/node/http-transport-utils.ts b/experimental/packages/otlp-exporter-base/src/platform/node/http-transport-utils.ts
@@ -16,6 +16,7 @@
 import * as http from 'http';
 import * as https from 'https';
 import * as zlib from 'zlib';
+import { TcpNetConnectOpts, Socket } from 'net';
 import { Readable } from 'stream';
 import { HttpRequestParameters } from './http-transport-types';
 import { ExportResponse } from '../../export-response';
@@ -138,11 +139,58 @@ function readableFromUint8Array(buff: string | Uint8Array): Readable {
   return readable;
 }
 
-export function createHttpAgent(
-  rawUrl: string,
-  agentOptions: http.AgentOptions | https.AgentOptions
-) {
-  const parsedUrl = new URL(rawUrl);
+export function createHttpAgent(params: HttpRequestParameters) {
+  const parsedUrl = new URL(params.url);
   const Agent = parsedUrl.protocol === 'http:' ? http.Agent : https.Agent;
-  return new Agent(agentOptions);
+
+  if (!params.proxy) {
+    return new Agent(params.agentOptions);
+  }
+
+  const parsedProxy = new URL(params.proxy);
+
+  const headers: http.OutgoingHttpHeaders = {};
+  if (parsedProxy.username) {
+    const basic = Buffer.from(
+      `${parsedProxy.username}:${parsedProxy.password}`
+    ).toString('base64');
+    headers['Proxy-Authorization'] = `Basic ${basic}`;
+  }
+
+  const request =
+    parsedProxy.protocol === 'http:' ? http.request : https.request;
+
+  class ProxyAgent extends Agent {
+    constructor() {
+      super({ keepAlive: true, ...params.agentOptions });
+    }
+
+    createConnection(
+      options: TcpNetConnectOpts,
+      callback: (err: Error | null, conn?: Socket | null) => void
+    ) {
+      const req = request({
+        method: 'CONNECT',
+        hostname: parsedProxy.hostname,
+        port: parsedProxy.port,
+        headers,
+        path: `${options.host || parsedUrl.hostname}:${options.port}`,
+        timeout: options.timeout,
+      })
+        .on('connect', (res, conn) => {
+          if (res.statusCode === 200) {
+            callback(null, conn);
+          } else {
+            callback(new OTLPExporterError(res.statusMessage, res.statusCode));
+          }
+        })
+        .on('error', callback)
+        .end();
+
+      req.on('timeout', () =>
+        req.socket!.destroy(new Error('Request Timeout'))
+      );
+    }
+  }
+  return new ProxyAgent();
 }
diff --git a/experimental/packages/otlp-exporter-base/src/platform/node/types.ts b/experimental/packages/otlp-exporter-base/src/platform/node/types.ts
@@ -24,6 +24,7 @@ import { OTLPExporterConfigBase } from '../../types';
 export interface OTLPExporterNodeConfigBase extends OTLPExporterConfigBase {
   keepAlive?: boolean;
   compression?: CompressionAlgorithm;
+  proxy?: string
   httpAgentOptions?: http.AgentOptions | https.AgentOptions;
 }
 

diff --git a/experimental/packages/otlp-exporter-base/test/node/http-exporter-transport.test.ts b/experimental/packages/otlp-exporter-base/test/node/http-exporter-transport.test.ts
@@ -16,6 +16,7 @@
 
 import { createHttpExporterTransport } from '../../src/platform/node/http-exporter-transport';
 import * as http from 'http';
+import * as net from 'net';
 import * as assert from 'assert';
 import sinon = require('sinon');
 import {
@@ -70,6 +71,52 @@ describe('HttpExporterTransport', function () {
       );
     });
 
+    it('returns success on proxied success status', async function () {
+      // arrange
+      const expectedResponseData = Buffer.from([4, 5, 6]);
+      server = http
+        .createServer((_, res) => {
+          res.statusCode = 200;
+          res.write(expectedResponseData);
+          res.end();
+        })
+        .on('connect', (req, socket, head) => {
+          const authorization = req.headers['proxy-authorization'];
+          const credentials = Buffer.from('open:telemetry').toString('base64');
+          if (authorization?.slice('Basic '.length) !== credentials) {
+            socket.write('HTTP/1.1 407');
+            socket.end();
+            return;
+          }
+
+          const [hostname, port] = req.url!.split(':');
+          const proxy = net.connect(Number(port), hostname, () => {
+            socket.write('HTTP/1.1 200\r\n\r\n');
+            proxy.write(head);
+            socket.pipe(proxy).pipe(socket);
+          });
+        });
+      server.listen(8080);
+
+      const transport = createHttpExporterTransport({
+        url: 'http://localhost:8080',
+        proxy: 'http://open:telemetry@localhost:8080',
+        headers: {},
+        compression: 'none',
+        agentOptions: {},
+      });
+
+      // act
+      const result = await transport.send(sampleRequestData, 1000);
+
+      // assert
+      assert.strictEqual(result.status, 'success');
+      assert.deepEqual(
+        (result as ExportResponseSuccess).data,
+        expectedResponseData
+      );
+    });
+
     it('returns retryable on retryable status', async function () {
       //arrange
       server = http.createServer((_, res) => {