Processor API Guide

The processor has two responsibilities: choosing which aggregator to choose for a metric instrument and store the last record for each metric ready to be exported.

有两个职责：为指标度量工具选择哪一个聚合器，和存储每一个导出的指标的最后一条记录。

为指标选择特定的聚合器

有时，您可能希望对您的指标之一使用特定的聚合器，导出最后 X 值的平均值，而不是仅导出最后一个值。

下面是一个聚合器的使用样例：

  1. import { Aggregator } from '@opentelemetry/sdk-metrics-base';
  2. import { hrTime } from '@opentelemetry/core';
  4. export class AverageAggregator implements Aggregator {
  6.   private _values: number[] = [];
  7.   private _limit: number;
  9.   constructor (limit?: number) {
  10.     this._limit = limit ?? 10;
  11.   }
  13.   update (value: number) {
  14.     this._values.push(value);
  15.     if (this._values.length >= this._limit) {
  16.       this._values = this._values.slice(0, 10);
  17.     }
  18.   }
  20.   toPoint() {
  21.     const sum =this._values.reduce((agg, value) => {
  22.       agg += value;
  23.       return agg;
  24.     }, 0);
  25.     return {
  26.       value: this._values.length > 0 ? sum / this._values.length : 0,
  27.       timestamp: hrTime(),
  28.     }
  29.   }
  30. }

现在我们需要实现我们自己的处理器来配置 sdk 以使用我们的新聚合器。 为了进一步简化，我们将只扩展 UngroupedProcessor（这是默认设置）以避免重新实现整个 Aggregator 接口。

结果如下：

  1. import {
  2.   UngroupedProcessor,
  3.   MetricDescriptor,
  4.   CounterSumAggregator,
  5.   ObserverAggregator,
  6.   MeasureExactAggregator,
  7. } from '@opentelemetry/sdk-metrics-base';
  9. export class CustomProcessor extends UngroupedProcessor {
  10.   aggregatorFor (metricDescriptor: MetricDescriptor) {
  11.     if (metricDescriptor.name === 'requests') {
  12.       return new AverageAggregator(10);
  13.     }
  14.     // this is exactly what the "UngroupedProcessor" does, we will re-use it
  15.     // to fallback on the default behavior
  16.     switch (metricDescriptor.metricKind) {
  17.       case MetricKind.COUNTER:
  18.         return new CounterSumAggregator();
  19.       case MetricKind.OBSERVER:
  20.         return new ObserverAggregator();
  21.       default:
  22.         return new MeasureExactAggregator();
  23.     }
  24.   }
  25. }

最后，我们需要指定MeterProvider在创建新仪表时使用我们的CustomProcessor：

  1. import {
  2.   UngroupedProcessor,
  3.   MetricDescriptor,
  4.   CounterSumAggregator,
  5.   ObserverAggregator,
  6.   MeasureExactAggregator,
  7.   MeterProvider,
  8.   Aggregator,
  9. } from '@opentelemetry/sdk-metrics-base';
  10. import { hrTime } from '@opentelemetry/core';
  12. export class AverageAggregator implements Aggregator {
  14.   private _values: number[] = [];
  15.   private _limit: number;
  17.   constructor (limit?: number) {
  18.     this._limit = limit ?? 10;
  19.   }
  21.   update (value: number) {
  22.     this._values.push(value);
  23.     if (this._values.length >= this._limit) {
  24.       this._values = this._values.slice(0, 10);
  25.     }
  26.   }
  28.   toPoint() {
  29.     const sum =this._values.reduce((agg, value) => {
  30.       agg += value;
  31.       return agg;
  32.     }, 0);
  33.     return {
  34.       value: this._values.length > 0 ? sum / this._values.length : 0,
  35.       timestamp: hrTime(),
  36.     }
  37.   }
  38. }
  40. export class CustomProcessor extends UngroupedProcessor {
  41.   aggregatorFor (metricDescriptor: MetricDescriptor) {
  42.     if (metricDescriptor.name === 'requests') {
  43.       return new AverageAggregator(10);
  44.     }
  45.     // this is exactly what the "UngroupedProcessor" does, we will re-use it
  46.     // to fallback on the default behavior
  47.     switch (metricDescriptor.metricKind) {
  48.       case MetricKind.COUNTER:
  49.         return new CounterSumAggregator();
  50.       case MetricKind.OBSERVER:
  51.         return new ObserverAggregator();
  52.       default:
  53.         return new MeasureExactAggregator();
  54.     }
  55.   }
  56. }
  58. const meter = new MeterProvider({
  59.   processor: new CustomProcessor(),
  60.   interval: 1000,
  61. }).getMeter('example-custom-processor');
  63. const requestsLatency = meter.createValueRecorder('requests', {
  64.   monotonic: true,
  65.   description: 'Average latency'
  66. });