unifying shutdown across code base

packages/opentelemetry-exporter-collector-grpc/src/CollectorExporterNodeBase.ts

@@ -49,6 +49,31 @@ export abstract class CollectorExporterNodeBase<
     }
     this.metadata = config.metadata;
   }
+  private _sendPromise(
+    objects: ExportItem[],
+    onSuccess: () => void,
+    onError: (error: collectorTypes.CollectorExporterError) => void
+  ): void {
+    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();
+      };
+
+      this._send(this, objects, _onSuccess, _onError);
+    });
+
+    this._sendingPromises.push(promise);
+  }
 
   onInit(config: CollectorExporterConfigNode): void {
     this._isShutdown = false;
@@ -77,10 +102,11 @@ export abstract class CollectorExporterNodeBase<
         // eslint-disable-next-line @typescript-eslint/no-var-requires
         const { send } = require('./util');
         this._send = send;
-        this._send(this, objects, onSuccess, onError);
+
+        this._sendPromise(objects, onSuccess, onError);
       });
     } else {
-      this._send(this, objects, onSuccess, onError);
+      this._sendPromise(objects, onSuccess, onError);
     }
   }
 

packages/opentelemetry-exporter-collector-proto/src/CollectorExporterNodeBase.ts

@@ -28,6 +28,33 @@ export abstract class CollectorExporterNodeBase<
   ServiceRequest
 > extends CollectorExporterBaseMain<ExportItem, ServiceRequest> {
   private _send!: Function;
+
+  private _sendPromise(
+    objects: ExportItem[],
+    onSuccess: () => void,
+    onError: (error: collectorTypes.CollectorExporterError) => void
+  ): void {
+    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();
+      };
+
+      this._send(this, objects, _onSuccess, _onError);
+    });
+
+    this._sendingPromises.push(promise);
+  }
+
   onInit(config: collectorTypes.CollectorExporterConfigBase): void {
     this._isShutdown = false;
     // defer to next tick and lazy load to avoid loading protobufjs too early
@@ -55,10 +82,10 @@ export abstract class CollectorExporterNodeBase<
         // eslint-disable-next-line @typescript-eslint/no-var-requires
         const { send } = require('./util');
         this._send = send;
-        this._send(this, objects, onSuccess, onError);
+        this._sendPromise(objects, onSuccess, onError);
       });
     } else {
-      this._send(this, objects, onSuccess, onError);
+      this._sendPromise(objects, onSuccess, onError);
     }
   }
 

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

@@ -59,13 +59,30 @@ export abstract class CollectorExporterNodeBase<
     }
     const serviceRequest = this.convert(objects);
 
-    sendWithHttp(
-      this,
-      JSON.stringify(serviceRequest),
-      'application/json',
-      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();
+      };
+      sendWithHttp(
+        this,
+        JSON.stringify(serviceRequest),
+        'application/json',
+        _onSuccess,
+        _onError
+      );
+    });
+
+    this._sendingPromises.push(promise);
   }
 
   onShutdown(): void {}

packages/opentelemetry-exporter-collector/test/common/CollectorMetricExporter.test.ts

@@ -138,20 +138,24 @@ describe('CollectorMetricExporter - common', () => {
     });
 
     describe('when exporter is shutdown', () => {
-      it('should not export anything but return callback with code "FailedNotRetryable"', () => {
-        collectorExporter.shutdown();
-        spySend.resetHistory();
-
-        const callbackSpy = sinon.spy();
-        collectorExporter.export(metrics, callbackSpy);
-        const returnCode = callbackSpy.args[0][0];
-        assert.strictEqual(
-          returnCode,
-          ExportResult.FAILED_NOT_RETRYABLE,
-          'return value is wrong'
-        );
-        assert.strictEqual(spySend.callCount, 0, 'should not call send');
-      });
+      it(
+        'should not export anything but return callback with code' +
+          ' "FailedNotRetryable"',
+        async () => {
+          await collectorExporter.shutdown();
+          spySend.resetHistory();
+
+          const callbackSpy = sinon.spy();
+          collectorExporter.export(metrics, callbackSpy);
+          const returnCode = callbackSpy.args[0][0];
+          assert.strictEqual(
+            returnCode,
+            ExportResult.FAILED_NOT_RETRYABLE,
+            'return value is wrong'
+          );
+          assert.strictEqual(spySend.callCount, 0, 'should not call send');
+        }
+      );
     });
     describe('when an error occurs', () => {
       it('should return a Not Retryable Error', done => {
@@ -173,7 +177,7 @@ describe('CollectorMetricExporter - common', () => {
           );
           assert.strictEqual(spySend.callCount, 1, 'should call send');
           done();
-        }, 500);
+        });
       });
 
       it('should return a Retryable Error', done => {
@@ -195,7 +199,7 @@ describe('CollectorMetricExporter - common', () => {
           );
           assert.strictEqual(spySend.callCount, 1, 'should call send');
           done();
-        }, 500);
+        });
       });
     });
   });
@@ -220,12 +224,9 @@ describe('CollectorMetricExporter - common', () => {
       onShutdownSpy.restore();
     });
 
-    it('should call onShutdown', done => {
-      collectorExporter.shutdown();
-      setTimeout(() => {
-        assert.equal(onShutdownSpy.callCount, 1);
-        done();
-      });
+    it('should call onShutdown', async () => {
+      await collectorExporter.shutdown();
+      assert.strictEqual(onShutdownSpy.callCount, 1);
     });
   });
 });

packages/opentelemetry-exporter-collector/test/common/CollectorTraceExporter.test.ts

@@ -134,23 +134,27 @@ describe('CollectorTraceExporter - common', () => {
     });
 
     describe('when exporter is shutdown', () => {
-      it('should not export anything but return callback with code "FailedNotRetryable"', () => {
-        const spans: ReadableSpan[] = [];
-        spans.push(Object.assign({}, mockedReadableSpan));
-        collectorExporter.shutdown();
-        spySend.resetHistory();
+      it(
+        'should not export anything but return callback with code' +
+          ' "FailedNotRetryable"',
+        async () => {
+          const spans: ReadableSpan[] = [];
+          spans.push(Object.assign({}, mockedReadableSpan));
+          await collectorExporter.shutdown();
+          spySend.resetHistory();
+
+          const callbackSpy = sinon.spy();
+          collectorExporter.export(spans, callbackSpy);
+          const returnCode = callbackSpy.args[0][0];
 
-        const callbackSpy = sinon.spy();
-        collectorExporter.export(spans, callbackSpy);
-        const returnCode = callbackSpy.args[0][0];
-
-        assert.strictEqual(
-          returnCode,
-          ExportResult.FAILED_NOT_RETRYABLE,
-          'return value is wrong'
-        );
-        assert.strictEqual(spySend.callCount, 0, 'should not call send');
-      });
+          assert.strictEqual(
+            returnCode,
+            ExportResult.FAILED_NOT_RETRYABLE,
+            'return value is wrong'
+          );
+          assert.strictEqual(spySend.callCount, 0, 'should not call send');
+        }
+      );
     });
     describe('when an error occurs', () => {
       it('should return a Not Retryable Error', done => {
@@ -174,7 +178,7 @@ describe('CollectorTraceExporter - common', () => {
           );
           assert.strictEqual(spySend.callCount, 1, 'should call send');
           done();
-        }, 500);
+        });
       });
 
       it('should return a Retryable Error', done => {
@@ -198,7 +202,7 @@ describe('CollectorTraceExporter - common', () => {
           );
           assert.strictEqual(spySend.callCount, 1, 'should call send');
           done();
-        }, 500);
+        });
       });
     });
   });
@@ -223,12 +227,9 @@ describe('CollectorTraceExporter - common', () => {
       onShutdownSpy.restore();
     });
 
-    it('should call onShutdown', done => {
-      collectorExporter.shutdown();
-      setTimeout(() => {
-        assert.equal(onShutdownSpy.callCount, 1);
-        done();
-      });
+    it('should call onShutdown', async () => {
+      await collectorExporter.shutdown();
+      assert.strictEqual(onShutdownSpy.callCount, 1);
     });
   });
 });