feat(otlp-exporter-http): add retries

packages/exporter-trace-otlp-http/src/OTLPExporterBase.ts

@@ -64,8 +64,12 @@ export abstract class OTLPExporterBase<
    * Export items.
    * @param items
    * @param resultCallback
+   * @param exportTimeoutMillis - optional
+   * @param onError - optional
    */
+
   export(items: ExportItem[], resultCallback: (result: ExportResult) => void): void {
+
     if (this._shutdownOnce.isCalled) {
       resultCallback({
         code: ExportResultCode.FAILED,
@@ -83,12 +87,12 @@ export abstract class OTLPExporterBase<
     }
 
     this._export(items)
-      .then(() => {
-        resultCallback({ code: ExportResultCode.SUCCESS });
-      })
-      .catch((error: ExportServiceError) => {
-        resultCallback({ code: ExportResultCode.FAILED, error });
-      });
+    .then(() => {
+      resultCallback({ code: ExportResultCode.SUCCESS });
+    })
+    .catch((error: ExportServiceError) => {
+      resultCallback({ code: ExportResultCode.FAILED, error });
+    });
   }
 
   private _export(items: ExportItem[]): Promise<unknown> {

packages/exporter-trace-otlp-http/src/platform/node/util.ts

@@ -25,6 +25,9 @@ import { diag } from '@opentelemetry/api';
 import { CompressionAlgorithm } from './types';
 
 let gzip: zlib.Gzip | undefined;
+const DEFAULT_MAX_ATTEMPTS = 4;
+const DEFAULT_INITIAL_BACKOFF = 1000;
+const DEFAULT_BACKOFF_MULTIPLIER = 1.5;
 
 /**
  * Sends data using http
@@ -43,6 +46,21 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
 ): void {
   const parsedUrl = new url.URL(collector.url);
 
+  // temp code - this will be merged from timeout pr
+  const exporterTimeout = collector._timeoutMillis;
+  let reqIsDestroyed: boolean;
+
+  let req: http.ClientRequest;
+  let retryTimer: ReturnType<typeof setTimeout>;
+
+  const exporterTimer = setTimeout(() => {
+    clearTimeout(retryTimer);
+    reqIsDestroyed = true;
+    req.destroy();
+    // create error here?
+    // onError(new Error('Request Timeout'));
+  }, exporterTimeout);
+
   const options: http.RequestOptions | https.RequestOptions = {
     hostname: parsedUrl.hostname,
     port: parsedUrl.port,
@@ -58,48 +76,75 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
 
   const request = parsedUrl.protocol === 'http:' ? http.request : https.request;
 
-  const req = request(options, (res: http.IncomingMessage) => {
-    let responseData = '';
-    res.on('data', chunk => (responseData += chunk));
-    res.on('end', () => {
-      if (res.statusCode && res.statusCode < 299) {
-        diag.debug(`statusCode: ${res.statusCode}`, responseData);
-        onSuccess();
-      } else {
-        const error = new otlpTypes.OTLPExporterError(
-          res.statusMessage,
-          res.statusCode,
-          responseData
+  const sendWithRetry = (retries = DEFAULT_MAX_ATTEMPTS, backoffMillis = DEFAULT_INITIAL_BACKOFF) => {
+    req = request(options, (res: http.IncomingMessage) => {
+      let responseData = '';
+      res.on('data', chunk => (responseData += chunk));
+      res.on('end', () => {
+        if (res.statusCode && res.statusCode < 299) {
+          diag.debug(`statusCode: ${res.statusCode}`, responseData);
+            // clear all timers since request was successful and resolve promise
+          clearTimeout(exporterTimer);
+          clearTimeout(retryTimer);
+          onSuccess();
+        } else if (res.statusCode && isRetryable(res.statusCode) && retries > 0) {
+            retryTimer = setTimeout(() => {
+              return sendWithRetry(retries - 1, backoffMillis * DEFAULT_BACKOFF_MULTIPLIER);
+            }, backoffMillis);
+        } else {
+          const error = new otlpTypes.OTLPExporterError(
+            res.statusMessage,
+            res.statusCode,
+            responseData
+          );
+          // clear all timers since request failed and there are no more retries left
+          // then reject promise
+          clearTimeout(exporterTimer);
+          clearTimeout(retryTimer);
+          onError(error);
+        }
+      });
+    });
+
+    // temp code - this will be merged from timeout pr
+    req.on('error', (error: Error | any) => {
+      if (reqIsDestroyed) {
+        const err = new otlpTypes.OTLPExporterError(
+          'Request Timeout', error.code
         );
+        onError(err);
+      } else {
         onError(error);
       }
     });
-  });
-
 
-  req.on('error', (error: Error) => {
-    onError(error);
-  });
+    switch (collector.compression) {
+      case CompressionAlgorithm.GZIP: {
+        if (!gzip) {
+          gzip = zlib.createGzip();
+        }
+        req.setHeader('Content-Encoding', 'gzip');
+        const dataStream = readableFromBuffer(data);
+        dataStream.on('error', onError)
+          .pipe(gzip).on('error', onError)
+          .pipe(req);
 
-  switch (collector.compression) {
-    case CompressionAlgorithm.GZIP: {
-      if (!gzip) {
-        gzip = zlib.createGzip();
+        break;
       }
-      req.setHeader('Content-Encoding', 'gzip');
-      const dataStream = readableFromBuffer(data);
-      dataStream.on('error', onError)
-        .pipe(gzip).on('error', onError)
-        .pipe(req);
+      default:
+        req.write(data);
+        req.end();
 
-      break;
+        break;
     }
-    default:
-      req.write(data);
-      req.end();
+  };
+  sendWithRetry();
+}
 
-      break;
-  }
+function isRetryable(statusCode: number) {
+  const retryCodes = [429, 502, 503, 504];
+
+  return retryCodes.includes(statusCode);
 }
 
 function readableFromBuffer(buff: string | Buffer): Readable {

packages/opentelemetry-sdk-trace-base/src/export/SpanExporter.ts

@@ -28,10 +28,15 @@ export interface SpanExporter {
   /**
    * Called to export sampled {@link ReadableSpan}s.
    * @param spans the list of sampled Spans to be exported.
+   * @param resultCallback
+   * @param exportTimeoutMillis optional - currently only used by BatchSpanProcessorBase
+   * @param onError optional - currently only used by BatchSpanProcessorBase
    */
   export(
     spans: ReadableSpan[],
-    resultCallback: (result: ExportResult) => void
+    resultCallback: (result: ExportResult) => void,
+    exportTimeoutMillis?: number,
+    onError?: (error: object) => void,
   ): void;
 
   /** Stops the exporter. */