Add Graceful Shutdown Support

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

@@ -184,15 +184,20 @@ describe('PrometheusExporter', () => {
 
  describe('export', () => {
  let exporter: PrometheusExporter;
+ let meterProvider: MeterProvider;
  let meter: Meter;
 
  beforeEach(done => {
  exporter = new PrometheusExporter();
- meter = new MeterProvider().getMeter('test-prometheus');
+ meterProvider = new MeterProvider();
+ meter = meterProvider.getMeter('test-prometheus', '1', {
+ exporter: exporter,
+ });
  exporter.startServer(done);
  });
 
  afterEach(done => {
+ process.removeAllListeners('SIGTERM');
  exporter.shutdown(done);
  });
 
@@ -319,6 +324,69 @@ describe('PrometheusExporter', () => {
  });
  });
 
+ it('should export multiple labels on graceful shutdown', done => {
+ const counter = meter.createCounter('counter', {
+ description: 'a test description',
+ }) as CounterMetric;
+
+ counter.bind({ counterKey1: 'labelValue1' }).add(10);
+ counter.bind({ counterKey1: 'labelValue2' }).add(20);
+ counter.bind({ counterKey1: 'labelValue3' }).add(30);
+ process.once('SIGTERM', () => {
+ http
+ .get('http://localhost:9464/metrics', res => {
+ res.on('data', chunk => {
+ const body = chunk.toString();
+ const lines = body.split('\n');
+
+ assert.deepStrictEqual(lines, [
+ '# HELP counter a test description',
+ '# TYPE counter counter',
+ `counter{counterKey1="labelValue1"} 10 ${mockedTimeMS}`,
+ `counter{counterKey1="labelValue2"} 20 ${mockedTimeMS}`,
+ `counter{counterKey1="labelValue3"} 30 ${mockedTimeMS}`,
+ '',
+ ]);
+
+ done();
+ });
+ })
+ .on('error', errorHandler(done));
+ });
+ process.kill(process.pid, 'SIGTERM');
+ });
+
+ it('should export multiple labels on manual shutdown', done => {
+ const counter = meter.createCounter('counter', {
+ description: 'a test description',
+ }) as CounterMetric;
+
+ counter.bind({ counterKey1: 'labelValue1' }).add(10);
+ counter.bind({ counterKey1: 'labelValue2' }).add(20);
+ counter.bind({ counterKey1: 'labelValue3' }).add(30);
+ meterProvider.shutdown(() => {
+ http
+ .get('http://localhost:9464/metrics', res => {
+ res.on('data', chunk => {
+ const body = chunk.toString();
+ const lines = body.split('\n');
+
+ assert.deepStrictEqual(lines, [
+ '# HELP counter a test description',
+ '# TYPE counter counter',
+ `counter{counterKey1="labelValue1"} 10 ${mockedTimeMS}`,
+ `counter{counterKey1="labelValue2"} 20 ${mockedTimeMS}`,
+ `counter{counterKey1="labelValue3"} 30 ${mockedTimeMS}`,
+ '',
+ ]);
+
+ done();
+ });
+ })
+ .on('error', errorHandler(done));
+ });
+ });
+
  it('should export a comment if no metrics are registered', done => {
  exporter.export([], () => {
  http
@@ -425,6 +493,7 @@ describe('PrometheusExporter', () => {
 
  beforeEach(() => {
  meter = new MeterProvider().getMeter('test-prometheus');
+ process.removeAllListeners('SIGTERM');
  counter = meter.createCounter('counter') as CounterMetric;
  counter.bind({ key1: 'labelValue1' }).add(10);
  });

packages/opentelemetry-metrics/src/Meter.ts

@@ -40,6 +40,7 @@ export class Meter implements api.Meter {
  private readonly _batcher: Batcher;
  private readonly _resource: Resource;
  private readonly _instrumentationLibrary: InstrumentationLibrary;
+ private readonly _controller: PushController;
 
  /**
  * Constructs a new Meter instance.
@@ -55,7 +56,7 @@ export class Meter implements api.Meter {
  // start the push controller
  const exporter = config.exporter || new NoopExporter();
  const interval = config.interval;
- new PushController(this, exporter, interval);
+ this._controller = new PushController(this, exporter, interval);
  }
 
  /**
@@ -309,6 +310,10 @@ export class Meter implements api.Meter {
  return this._batcher;
  }
 
+ async shutdown(): Promise<void> {
+ await this._controller.shutdown();
+ }
+
  /**
  * Registers metric to register.
  * @param name The name of the metric.

packages/opentelemetry-metrics/src/MeterProvider.ts

@@ -36,6 +36,9 @@ export class MeterProvider implements api.MeterProvider {
  logger: this.logger,
  resource: this.resource,
  });
+ if (this._config.gracefulShutdown) {
+ process.once('SIGTERM', this.onShutdown.bind(this));
+ }
  }
 
  /**
@@ -54,4 +57,21 @@ export class MeterProvider implements api.MeterProvider {
 
  return this._meters.get(key)!;
  }
+
+ shutdown(cb: () => void = () => {}) {
+ this.onShutdown().then(() => {
+ setTimeout(cb, 0);
+ });
+ }
+
+ private async onShutdown() {
+ if (this._config.exporter) {
+ this._config.exporter.shutdown();
+ }
+ const shutdownPromises: Promise<void>[] = [];
+ this._meters.forEach((meter, _) => {
+ shutdownPromises.push(meter.shutdown());
+ });
+ await Promise.all(shutdownPromises);
+ }
 }

packages/opentelemetry-metrics/src/export/Controller.ts

@@ -38,12 +38,20 @@ export class PushController extends Controller {
  unrefTimer(this._timer);
  }
 
- private _collect() {
- this._meter.collect();
- this._exporter.export(this._meter.getBatcher().checkPointSet(), result => {
- if (result !== ExportResult.SUCCESS) {
- // @todo: log error
- }
+ async shutdown() {
+ await this._collect();
+ }
+
+ private async _collect() {
+ await this._meter.collect().then(() => {
+ this._exporter.export(
+ this._meter.getBatcher().checkPointSet(),
+ result => {
+ if (result !== ExportResult.SUCCESS) {
+ // @todo: log error
+ }
+ }
+ );
  });
  }
 }

packages/opentelemetry-metrics/src/types.ts

@@ -39,11 +39,15 @@ export interface MeterConfig {
 
  /** Metric batcher. */
  batcher?: Batcher;
+
+ /** Bool for whether or not graceful shutdown is enabled */
+ gracefulShutdown?: boolean;
 }
 
 /** Default Meter configuration. */
 export const DEFAULT_CONFIG = {
  logLevel: LogLevel.INFO,
+ gracefulShutdown: true,
 };
 
 /** The default metric creation options value. */

packages/opentelemetry-metrics/test/MeterProvider.test.ts

@@ -15,10 +15,15 @@
  */
 
 import * as assert from 'assert';
+import * as sinon from 'sinon';
 import { MeterProvider, Meter, CounterMetric } from '../src';
 import { NoopLogger } from '@opentelemetry/core';
 
 describe('MeterProvider', () => {
+ afterEach(() => {
+ process.removeAllListeners('SIGTERM');
+ });
+
  describe('constructor', () => {
  it('should construct an instance without any options', () => {
  const provider = new MeterProvider();
@@ -73,4 +78,41 @@ describe('MeterProvider', () => {
  assert.notEqual(meter3, meter4);
  });
  });
+
+ describe('shutdown()', () => {
+ it('should call shutdown when SIGTERM is received', () => {
+ const meterProvider = new MeterProvider();
+ const sandbox = sinon.createSandbox();
+ const shutdownStub1 = sandbox.stub(
+ meterProvider.getMeter('meter1'),
+ 'shutdown'
+ );
+ const shutdownStub2 = sandbox.stub(
+ meterProvider.getMeter('meter2'),
+ 'shutdown'
+ );
+ process.once('SIGTERM', () => {
+ sinon.assert.calledOnce(shutdownStub1);
+ sinon.assert.calledOnce(shutdownStub2);
+ sandbox.restore();
+ });
+ process.kill(process.pid, 'SIGTERM');
+ });
+
+ it('should not trigger shutdown if graceful shutdown is turned off', () => {
+ const meterProvider = new MeterProvider({
+ gracefulShutdown: false,
+ });
+ const sandbox = sinon.createSandbox();
+ const shutdownStub = sandbox.stub(
+ meterProvider.getMeter('meter1'),
+ 'shutdown'
+ );
+ process.once('SIGTERM', () => {
+ sinon.assert.notCalled(shutdownStub);
+ sandbox.restore();
+ });
+ process.kill(process.pid, 'SIGTERM');
+ });
+ });
 });

packages/opentelemetry-tracing/src/BasicTracerProvider.ts

@@ -47,6 +47,9 @@ export class BasicTracerProvider implements api.TracerProvider {
  logger: this.logger,
  resource: this.resource,
  });
+ if (this._config.gracefulShutdown) {
+ process.once('SIGTERM', this.onShutdown.bind(this));
+ }
  }
 
  getTracer(name: string, version = '*', config?: TracerConfig): Tracer {
@@ -99,4 +102,12 @@ export class BasicTracerProvider implements api.TracerProvider {
  api.propagation.setGlobalPropagator(config.propagator);
  }
  }
+
+ shutdown(cb: () => void = () => {}) {
+ this.activeSpanProcessor.shutdown(cb);
+ }
+
+ private onShutdown() {
+ this.activeSpanProcessor.shutdown();
+ }
 }

packages/opentelemetry-tracing/src/config.ts

@@ -38,4 +38,5 @@ export const DEFAULT_CONFIG = {
  numberOfLinksPerSpan: DEFAULT_MAX_LINKS_PER_SPAN,
  numberOfEventsPerSpan: DEFAULT_MAX_EVENTS_PER_SPAN,
  },
+ gracefulShutdown: true,
 };

packages/opentelemetry-tracing/src/types.ts

@@ -52,6 +52,9 @@ export interface TracerConfig {
 
  /** Resource associated with trace telemetry */
  resource?: Resource;
+
+ /** Bool for whether or not graceful shutdown is enabled */
+ gracefulShutdown?: boolean;
 }
 
 /**

packages/opentelemetry-tracing/test/BasicTracerProvider.test.ts

@@ -27,13 +27,18 @@ import {
 } from '@opentelemetry/core';
 import { Resource } from '@opentelemetry/resources';
 import * as assert from 'assert';
+import * as sinon from 'sinon';
 import { BasicTracerProvider, Span } from '../src';
 
 describe('BasicTracerProvider', () => {
  beforeEach(() => {
  context.disable();
  });
 
+ afterEach(() => {
+ process.removeAllListeners('SIGTERM');
+ });
+
  describe('constructor', () => {
  it('should construct an instance without any options', () => {
  const provider = new BasicTracerProvider();
@@ -381,4 +386,36 @@ describe('BasicTracerProvider', () => {
  assert.ok(tracerProvider.resource instanceof Resource);
  });
  });
+
+ describe('.shutdown()', () => {
+ it('should trigger shutdown when SIGTERM is recieved', () => {
+ const tracerProvider = new BasicTracerProvider();
+ const sandbox = sinon.createSandbox();
+ const shutdownStub = sandbox.stub(
+ tracerProvider.getActiveSpanProcessor(),
+ 'shutdown'
+ );
+ process.once('SIGTERM', () => {
+ sinon.assert.calledOnce(shutdownStub);
+ sandbox.restore();
+ });
+ process.kill(process.pid, 'SIGTERM');
+ });
+
+ it('should not trigger shutdown if graceful shutdown is turned off', () => {
+ const tracerProvider = new BasicTracerProvider({
+ gracefulShutdown: false,
+ });
+ const sandbox = sinon.createSandbox();
+ const shutdownStub = sandbox.stub(
+ tracerProvider.getActiveSpanProcessor(),
+ 'shutdown'
+ );
+ process.once('SIGTERM', () => {
+ sinon.assert.notCalled(shutdownStub);
+ sandbox.restore();
+ });
+ process.kill(process.pid, 'SIGTERM');
+ });
+ });
 });