example-otel-metrics.ts

import {
  service,
  Operation, sleep,
  internal as restateInternal, date
} from "@restatedev/restate-sdk-gen";
import * as restate from "@restatedev/restate-sdk";
import {Attributes, Context, Counter, Histogram, metrics} from '@opentelemetry/api';
import {
  ConsoleMetricExporter,
  MeterProvider,
  PeriodicExportingMetricReader,
} from '@opentelemetry/sdk-metrics';
import {
  defaultResource,
  resourceFromAttributes,
} from '@opentelemetry/resources';
import {
  ATTR_SERVICE_NAME,
  ATTR_SERVICE_VERSION,
} from '@opentelemetry/semantic-conventions';

// --- Slop to mock the console exporter, for demo purposes ---
const resource = defaultResource().merge(
    resourceFromAttributes({
      [ATTR_SERVICE_NAME]: 'sample',
      [ATTR_SERVICE_VERSION]: '0.1.0',
    }),
);
const metricReader = new PeriodicExportingMetricReader({
  exporter: new ConsoleMetricExporter(),
  // Default is 60000ms (60 seconds). Set to 10 seconds for demonstrative purposes only.
  exportIntervalMillis: 10000,
});
const myServiceMeterProvider = new MeterProvider({
  resource: resource,
  readers: [metricReader],
});
metrics.setGlobalMeterProvider(myServiceMeterProvider);

// --- Define metrics here ---

// Otel APIs to create counters
const meter = metrics.getMeter('sample', '0.1.0');
const otelCounter = meter.createCounter('sample.my-counter');

// Wrap in restate wrapper that is aware of replaying or not
const counter = restateAwareCounter(otelCounter);

// Wrapper that emits counter only when processing
function restateAwareCounter<A extends Attributes>(counter: Counter<A>) {
  return {
    add: (value: number, attributes?: A) => {
      const isProcessing = restateInternal.isProcessing()
      if (isProcessing === undefined || isProcessing) {
        // Record the metric either if i'm not in restate context, or if restate is processing.
        counter.add(value, attributes)
      }
    },
  };
}

// Wrapper that emits histogram only when processing
function restateAwareHistogram<A extends Attributes>(histogram: Histogram<A>): Histogram<A> {
  return {
    record(value: number, attributes?: A, context?: Context): void {
      const isProcessing = restateInternal.isProcessing()
      if (isProcessing === undefined || isProcessing) {
        // Record the metric either if i'm not in restate context, or if restate is processing.
        histogram.record(value, attributes, context)
      }
    }
  }
}

// --- Idea for timer histogram API ---

interface TimerHistogram {
  start(): Operation<TimerHistogramRecording>;
}
interface TimerHistogramRecording {
  end(): void;
}

function restateTimerHistogram(histogram: Histogram<any>): TimerHistogram {
  // Wrap the histogram to make sure we record only on replay
  const wrappedHistogram = restateAwareHistogram(histogram)

  return {
    *start(): Operation<TimerHistogramRecording> {
      // Record the date using Restate
      const start = yield * date().now()
      return {
        end: function (): void {
          wrappedHistogram.record(Date.now() - start)
        }
      }
    }
  }
}

const timerHistogram = restateTimerHistogram(meter.createHistogram("sample.timer"))

// --- Sample restate service ---

export const greeter = service({
  name: "greeter",
  handlers: {
    *hello(name: string): Operation<string> {

      const timer = yield* timerHistogram.start()

      counter.add(1);
      yield *sleep({seconds: 10})
      counter.add(1);

      timer.end();

      return `Hi ${name}!`;
    },
  },
  options: {
    inactivityTimeout: {seconds: 1}
  }
});

restate.serve({
  services: [
    greeter,
  ],
});