unifying shutdown across code base

packages/opentelemetry-exporter-collector/src/CollectorExporterBase.ts

@@ -36,6 +36,8 @@ export abstract class CollectorExporterBase<
   public readonly hostname: string | undefined;
   public readonly attributes?: Attributes;
   protected _isShutdown: boolean = false;
+  private _shuttingDownPromise: Promise<void> = Promise.resolve();
+  protected _sendingPromises: Promise<unknown>[] = [];
 
   /**
    * @param config
@@ -98,16 +100,29 @@ export abstract class CollectorExporterBase<
   /**
    * Shutdown the exporter.
    */
-  shutdown(): void {
+  shutdown(): Promise<void> {
     if (this._isShutdown) {
       this.logger.debug('shutdown already started');
-      return;
+      return this._shuttingDownPromise;
     }
     this._isShutdown = true;
     this.logger.debug('shutdown started');
-
-    // platform dependent
-    this.onShutdown();
+    this._shuttingDownPromise = new Promise((resolve, reject) => {
+      Promise.resolve()
+        .then(() => {
+          return this.onShutdown();
+        })
+        .then(() => {
+          return Promise.all(this._sendingPromises);
+        })
+        .then(() => {
+          resolve();
+        })
+        .catch(e => {
+          reject(e);
+        });
+    });
+    return this._shuttingDownPromise;
   }
 
   abstract onShutdown(): void;

packages/opentelemetry-exporter-collector/src/platform/browser/CollectorExporterBrowserBase.ts

@@ -61,20 +61,41 @@ export abstract class CollectorExporterBrowserBase<
     onSuccess: () => void,
     onError: (error: collectorTypes.CollectorExporterError) => void
   ) {
+    if (this._isShutdown) {
+      this.logger.debug('Shutdown already started. Cannot send objects');
+      return;
+    }
     const serviceRequest = this.convert(items);
     const body = JSON.stringify(serviceRequest);
 
-    if (this._useXHR) {
-      sendWithXhr(
-        body,
-        this.url,
-        this._headers,
-        this.logger,
-        onSuccess,
-        onError
-      );
-    } else {
-      sendWithBeacon(body, this.url, this.logger, onSuccess, onError);
-    }
+    const promise = new Promise(resolve => {
+      const _onSuccess = (): void => {
+        onSuccess();
+        _onFinish();
+      };
+      const _onError = (error: collectorTypes.CollectorExporterError): void => {
+        onError(error);
+        _onFinish();
+      };
+      const _onFinish = () => {
+        const index = this._sendingPromises.indexOf(promise);
+        this._sendingPromises.splice(index, 1);
+        resolve();
+      };
+
+      if (this._useXHR) {
+        sendWithXhr(
+          body,
+          this.url,
+          this._headers,
+          this.logger,
+          _onSuccess,
+          _onError
+        );
+      } else {
+        sendWithBeacon(body, this.url, this.logger, _onSuccess, _onError);
+      }
+    });
+    this._sendingPromises.push(promise);
   }
 }

packages/opentelemetry-exporter-collector/src/platform/node/CollectorExporterNodeBase.ts

@@ -95,13 +95,32 @@ export abstract class CollectorExporterNodeBase<
       this.logger.debug('Shutdown already started. Cannot send objects');
       return;
     }
-    if (this._protocol === CollectorProtocolNode.HTTP_JSON) {
-      sendWithJson(this, objects, onSuccess, onError);
-    } else if (this._protocol === CollectorProtocolNode.HTTP_PROTO) {
-      sendWithJsonProto(this, objects, onSuccess, onError);
-    } else {
-      sendWithGrpc(this, objects, onSuccess, onError);
-    }
+
+    const promise = new Promise(resolve => {
+      const _onSuccess = (): void => {
+        onSuccess();
+        _onFinish();
+      };
+      const _onError = (error: collectorTypes.CollectorExporterError): void => {
+        onError(error);
+        _onFinish();
+      };
+      const _onFinish = () => {
+        const index = this._sendingPromises.indexOf(promise);
+        this._sendingPromises.splice(index, 1);
+        resolve();
+      };
+
+      if (this._protocol === CollectorProtocolNode.HTTP_JSON) {
+        sendWithJson(this, objects, _onSuccess, _onError);
+      } else if (this._protocol === CollectorProtocolNode.HTTP_PROTO) {
+        sendWithJsonProto(this, objects, _onSuccess, _onError);
+      } else {
+        sendWithGrpc(this, objects, _onSuccess, _onError);
+      }
+    });
+
+    this._sendingPromises.push(promise);
   }
 
   onShutdown(): void {
@@ -116,5 +135,6 @@ export abstract class CollectorExporterNodeBase<
   }
 
   abstract getServiceProtoPath(): string;
+
   abstract getServiceClientType(): ServiceClientType;
 }

packages/opentelemetry-exporter-jaeger/src/jaeger.ts

@@ -30,6 +30,9 @@ export class JaegerExporter implements SpanExporter {
   private readonly _process: jaegerTypes.ThriftProcess;
   private readonly _sender: typeof jaegerTypes.UDPSender;
   private readonly _onShutdownFlushTimeout: number;
+  private _isShutdown = false;
+  private _shutdownFlushTimeout: NodeJS.Timeout | undefined;
+  private _shuttingDownPromise: Promise<void> = Promise.resolve();
 
   constructor(config: jaegerTypes.ExporterConfig) {
     const localConfig = Object.assign({}, config);
@@ -85,14 +88,40 @@ export class JaegerExporter implements SpanExporter {
   }
 
   /** Shutdown exporter. */
-  shutdown(): void {
-    // Make an optimistic flush.
-    this._flush();
-    // Sleeping x seconds before closing the sender's connection to ensure
-    // all spans are flushed.
-    setTimeout(() => {
-      this._sender.close();
-    }, this._onShutdownFlushTimeout);
+  shutdown(): Promise<void> {
+    if (this._isShutdown) {
+      return this._shuttingDownPromise;
+    }
+    this._isShutdown = true;
+
+    this._shuttingDownPromise = new Promise((resolve, reject) => {
+      let rejected = false;
+      this._shutdownFlushTimeout = setTimeout(() => {
+        rejected = true;
+        reject('timeout');
+        this._sender.close();
+      }, this._onShutdownFlushTimeout);
+
+      Promise.resolve()
+        .then(() => {
+          // Make an optimistic flush.
+          return this._flush();
+        })
+        .then(() => {
+          if (rejected) {
+            return;
+          } else {
+            this._shutdownFlushTimeout &&
+              clearTimeout(this._shutdownFlushTimeout);
+            resolve();
+            this._sender.close();
+          }
+        })
+        .catch(e => {
+          reject(e);
+        });
+    });
+    return this._shuttingDownPromise;
   }
 
   /** Transform spans and sends to Jaeger service. */

packages/opentelemetry-exporter-prometheus/src/prometheus.ts

@@ -70,7 +70,7 @@ export class PrometheusExporter implements MetricExporter {
     ).replace(/^([^/])/, '/$1');
 
     if (config.startServer || PrometheusExporter.DEFAULT_OPTIONS.startServer) {
-      this.startServer(callback);
+      this.startServer().then(callback);
     } else if (callback) {
       callback();
     }
@@ -108,12 +108,10 @@ export class PrometheusExporter implements MetricExporter {
 
   /**
    * Shuts down the export server and clears the registry
-   *
-   * @param cb called when server is stopped
    */
-  shutdown(cb?: () => void) {
+  shutdown(): Promise<void> {
     this._registry.clear();
-    this.stopServer(cb);
+    return this.stopServer();
   }
 
   /**
@@ -252,39 +250,45 @@ export class PrometheusExporter implements MetricExporter {
 
   /**
    * Stops the Prometheus export server
-   * @param callback A callback that will be executed once the server is stopped
    */
-  stopServer(callback?: () => void) {
+  stopServer(): Promise<void> {
     if (!this._server) {
       this._logger.debug(
         'Prometheus stopServer() was called but server was never started.'
       );
-      if (callback) {
-        callback();
-      }
+      return Promise.resolve();
     } else {
-      this._server.close(() => {
-        this._logger.debug('Prometheus exporter was stopped');
-        if (callback) {
-          callback();
-        }
+      return new Promise(resolve => {
+        this._server.close(err => {
+          if (!err) {
+            this._logger.debug('Prometheus exporter was stopped');
+          } else {
+            if (
+              ((err as unknown) as { code: string }).code !==
+              'ERR_SERVER_NOT_RUNNING'
+            ) {
+              this._logger.error(
+                `Error during stopping the Prometheus Exporter "${err.message}"`
+              );
+            }
+          }
+          resolve();
+        });
       });
     }
   }
 
   /**
    * Starts the Prometheus export server
-   *
-   * @param callback called once the server is ready
    */
-  startServer(callback?: () => void) {
-    this._server.listen(this._port, () => {
-      this._logger.debug(
-        `Prometheus exporter started on port ${this._port} at endpoint ${this._endpoint}`
-      );
-      if (callback) {
-        callback();
-      }
+  startServer(): Promise<void> {
+    return new Promise(resolve => {
+      this._server.listen(this._port, () => {
+        this._logger.debug(
+          `Prometheus exporter started on port ${this._port} at endpoint ${this._endpoint}`
+        );
+        resolve();
+      });
     });
   }