chore: rename fundamentals to base (#9119)

packages/backend/server/src/base/metrics/config.ts

@@ -0,0 +1,33 @@
+import { defineStartupConfig, ModuleConfig } from '../config';
+
+declare module '../config' {
+  interface AppConfig {
+    metrics: ModuleConfig<{
+      /**
+       * Enable metric and tracing collection
+       */
+      enabled: boolean;
+      /**
+       * Enable telemetry
+       */
+      telemetry: {
+        enabled: boolean;
+        token: string;
+      };
+      customerIo: {
+        token: string;
+      };
+    }>;
+  }
+}
+
+defineStartupConfig('metrics', {
+  enabled: false,
+  telemetry: {
+    enabled: false,
+    token: '',
+  },
+  customerIo: {
+    token: '',
+  },
+});

packages/backend/server/src/base/metrics/index.ts

@@ -0,0 +1,55 @@
+import './config';
+
+import {
+  Global,
+  Module,
+  OnModuleDestroy,
+  OnModuleInit,
+  Provider,
+} from '@nestjs/common';
+import { ModuleRef } from '@nestjs/core';
+import { NodeSDK } from '@opentelemetry/sdk-node';
+
+import { Config } from '../config';
+import {
+  LocalOpentelemetryFactory,
+  OpentelemetryFactory,
+  registerCustomMetrics,
+} from './opentelemetry';
+
+const factorProvider: Provider = {
+  provide: OpentelemetryFactory,
+  useFactory: (config: Config) => {
+    return config.metrics.enabled ? new LocalOpentelemetryFactory() : null;
+  },
+  inject: [Config],
+};
+
+@Global()
+@Module({
+  providers: [factorProvider],
+  exports: [factorProvider],
+})
+export class MetricsModule implements OnModuleInit, OnModuleDestroy {
+  private sdk: NodeSDK | null = null;
+  constructor(private readonly ref: ModuleRef) {}
+
+  onModuleInit() {
+    const factor = this.ref.get(OpentelemetryFactory, { strict: false });
+    if (factor) {
+      this.sdk = factor.create();
+      this.sdk.start();
+      registerCustomMetrics();
+    }
+  }
+
+  async onModuleDestroy() {
+    if (this.sdk) {
+      await this.sdk.shutdown();
+    }
+  }
+}
+
+export * from './metrics';
+export * from './utils';
+export { OpentelemetryFactory };

packages/backend/server/src/base/metrics/metrics.ts

@@ -0,0 +1,134 @@
+import {
+  Attributes,
+  Counter,
+  Histogram,
+  Meter,
+  MetricOptions,
+} from '@opentelemetry/api';
+
+import { getMeter } from './opentelemetry';
+
+type MetricType = 'counter' | 'gauge' | 'histogram';
+type Metric<T extends MetricType> = T extends 'counter'
+  ? Counter
+  : T extends 'gauge'
+    ? Histogram
+    : T extends 'histogram'
+      ? Histogram
+      : never;
+
+export type ScopedMetrics = {
+  [T in MetricType]: (name: string, opts?: MetricOptions) => Metric<T>;
+};
+type MetricCreators = {
+  [T in MetricType]: (
+    meter: Meter,
+    name: string,
+    opts?: MetricOptions
+  ) => Metric<T>;
+};
+
+export type KnownMetricScopes =
+  | 'socketio'
+  | 'gql'
+  | 'jwst'
+  | 'auth'
+  | 'controllers'
+  | 'doc'
+  | 'sse'
+  | 'mail'
+  | 'ai';
+
+const metricCreators: MetricCreators = {
+  counter(meter: Meter, name: string, opts?: MetricOptions) {
+    return meter.createCounter(name, opts);
+  },
+  gauge(meter: Meter, name: string, opts?: MetricOptions) {
+    let value: any;
+    let attrs: Attributes | undefined;
+    const ob$ = meter.createObservableGauge(name, opts);
+
+    ob$.addCallback(result => {
+      result.observe(value, attrs);
+    });
+
+    return {
+      record: (newValue, newAttrs) => {
+        value = newValue;
+        attrs = newAttrs;
+      },
+    } satisfies Histogram;
+  },
+  histogram(meter: Meter, name: string, opts?: MetricOptions) {
+    return meter.createHistogram(name, opts);
+  },
+};
+
+const scopes = new Map<string, ScopedMetrics>();
+
+function make(scope: string) {
+  const meter = getMeter();
+  const metrics = new Map<string, { type: MetricType; metric: any }>();
+  const prefix = scope + '/';
+
+  function getOrCreate<T extends MetricType>(
+    type: T,
+    name: string,
+    opts?: MetricOptions
+  ): Metric<T> {
+    name = prefix + name;
+    const metric = metrics.get(name);
+    if (metric) {
+      if (type !== metric.type) {
+        throw new Error(
+          `Metric ${name} has already been registered as ${metric.type} mode, but get as ${type} again.`
+        );
+      }
+
+      return metric.metric;
+    } else {
+      const metric = metricCreators[type](meter, name, opts);
+      metrics.set(name, { type, metric });
+      return metric;
+    }
+  }
+
+  return {
+    counter(name, opts) {
+      return getOrCreate('counter', name, opts);
+    },
+    gauge(name, opts) {
+      return getOrCreate('gauge', name, opts);
+    },
+    histogram(name, opts) {
+      return getOrCreate('histogram', name, opts);
+    },
+  } satisfies ScopedMetrics;
+}
+
+/**
+ * @example
+ *
+ * ```
+ * metrics.scope.counter('example_count').add(1, {
+ *   attr1: 'example-event'
+ * })
+ * ```
+ */
+export const metrics = new Proxy<Record<KnownMetricScopes, ScopedMetrics>>(
+  // @ts-expect-error proxied
+  {},
+  {
+    get(_, scopeName: string) {
+      let scope = scopes.get(scopeName);
+      if (!scope) {
+        scope = make(scopeName);
+        scopes.set(scopeName, scope);
+      }
+
+      return scope;
+    },
+  }
+);
+
+export function stopMetrics() {}

packages/backend/server/src/base/metrics/opentelemetry.ts

@@ -0,0 +1,119 @@
+import { OnModuleDestroy } from '@nestjs/common';
+import { metrics } from '@opentelemetry/api';
+import {
+  CompositePropagator,
+  W3CBaggagePropagator,
+  W3CTraceContextPropagator,
+} from '@opentelemetry/core';
+import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
+import { ZipkinExporter } from '@opentelemetry/exporter-zipkin';
+import { HostMetrics } from '@opentelemetry/host-metrics';
+import { Instrumentation } from '@opentelemetry/instrumentation';
+import { GraphQLInstrumentation } from '@opentelemetry/instrumentation-graphql';
+import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
+import { IORedisInstrumentation } from '@opentelemetry/instrumentation-ioredis';
+import { NestInstrumentation } from '@opentelemetry/instrumentation-nestjs-core';
+import { SocketIoInstrumentation } from '@opentelemetry/instrumentation-socket.io';
+import { Resource } from '@opentelemetry/resources';
+import type { MeterProvider } from '@opentelemetry/sdk-metrics';
+import { MetricProducer, MetricReader } from '@opentelemetry/sdk-metrics';
+import { NodeSDK } from '@opentelemetry/sdk-node';
+import {
+  BatchSpanProcessor,
+  SpanExporter,
+  TraceIdRatioBasedSampler,
+} from '@opentelemetry/sdk-trace-node';
+import {
+  SEMRESATTRS_K8S_NAMESPACE_NAME,
+  SEMRESATTRS_SERVICE_NAME,
+  SEMRESATTRS_SERVICE_VERSION,
+} from '@opentelemetry/semantic-conventions';
+import prismaInstrument from '@prisma/instrumentation';
+
+import { PrismaMetricProducer } from './prisma';
+
+const { PrismaInstrumentation } = prismaInstrument;
+
+export abstract class OpentelemetryFactory {
+  abstract getMetricReader(): MetricReader;
+  abstract getSpanExporter(): SpanExporter;
+
+  getInstractions(): Instrumentation[] {
+    return [
+      new NestInstrumentation(),
+      new IORedisInstrumentation(),
+      new SocketIoInstrumentation({ traceReserved: true }),
+      new GraphQLInstrumentation({ mergeItems: true }),
+      new HttpInstrumentation(),
+      new PrismaInstrumentation(),
+    ];
+  }
+
+  getMetricsProducers(): MetricProducer[] {
+    return [new PrismaMetricProducer()];
+  }
+
+  getResource() {
+    return new Resource({
+      [SEMRESATTRS_K8S_NAMESPACE_NAME]: AFFiNE.AFFINE_ENV,
+      [SEMRESATTRS_SERVICE_NAME]: AFFiNE.flavor.type,
+      [SEMRESATTRS_SERVICE_VERSION]: AFFiNE.version,
+    });
+  }
+
+  create() {
+    const traceExporter = this.getSpanExporter();
+    return new NodeSDK({
+      resource: this.getResource(),
+      sampler: new TraceIdRatioBasedSampler(0.1),
+      traceExporter,
+      metricReader: this.getMetricReader(),
+      spanProcessor: new BatchSpanProcessor(traceExporter),
+      textMapPropagator: new CompositePropagator({
+        propagators: [
+          new W3CBaggagePropagator(),
+          new W3CTraceContextPropagator(),
+        ],
+      }),
+      instrumentations: this.getInstractions(),
+      serviceName: 'affine-cloud',
+    });
+  }
+}
+
+export class LocalOpentelemetryFactory
+  extends OpentelemetryFactory
+  implements OnModuleDestroy
+{
+  private readonly metricsExporter = new PrometheusExporter({
+    metricProducers: this.getMetricsProducers(),
+  });
+
+  async onModuleDestroy() {
+    await this.metricsExporter.shutdown();
+  }
+
+  override getMetricReader(): MetricReader {
+    return this.metricsExporter;
+  }
+
+  override getSpanExporter(): SpanExporter {
+    return new ZipkinExporter();
+  }
+}
+
+function getMeterProvider() {
+  return metrics.getMeterProvider();
+}
+
+export function registerCustomMetrics() {
+  const hostMetricsMonitoring = new HostMetrics({
+    name: 'instance-host-metrics',
+    meterProvider: getMeterProvider() as MeterProvider,
+  });
+  hostMetricsMonitoring.start();
+}
+
+export function getMeter(name = 'business') {
+  return getMeterProvider().getMeter(name);
+}

packages/backend/server/src/base/metrics/prisma.ts

@@ -0,0 +1,132 @@
+import { HrTime, ValueType } from '@opentelemetry/api';
+import { hrTime } from '@opentelemetry/core';
+import { Resource } from '@opentelemetry/resources';
+import {
+  AggregationTemporality,
+  CollectionResult,
+  DataPointType,
+  InstrumentType,
+  MetricProducer,
+  ScopeMetrics,
+} from '@opentelemetry/sdk-metrics';
+
+import { PrismaService } from '../prisma';
+
+function transformPrismaKey(key: string) {
+  // replace first '_' to '/' as a scope prefix
+  // example: prisma_client_query_duration_seconds_sum -> prisma/client_query_duration_seconds_sum
+  return key.replace(/_/, '/');
+}
+
+export class PrismaMetricProducer implements MetricProducer {
+  private readonly startTime: HrTime = hrTime();
+
+  async collect(): Promise<CollectionResult> {
+    const result: CollectionResult = {
+      resourceMetrics: {
+        resource: Resource.EMPTY,
+        scopeMetrics: [],
+      },
+      errors: [],
+    };
+
+    if (!PrismaService.INSTANCE) {
+      return result;
+    }
+
+    const prisma = PrismaService.INSTANCE;
+
+    const endTime = hrTime();
+
+    const metrics = await prisma.$metrics.json();
+    const scopeMetrics: ScopeMetrics = {
+      scope: {
+        name: '',
+      },
+      metrics: [],
+    };
+    for (const counter of metrics.counters) {
+      scopeMetrics.metrics.push({
+        descriptor: {
+          name: transformPrismaKey(counter.key),
+          description: counter.description,
+          unit: '1',
+          type: InstrumentType.COUNTER,
+          valueType: ValueType.INT,
+        },
+        dataPointType: DataPointType.SUM,
+        aggregationTemporality: AggregationTemporality.CUMULATIVE,
+        dataPoints: [
+          {
+            startTime: this.startTime,
+            endTime: endTime,
+            value: counter.value,
+            attributes: counter.labels,
+          },
+        ],
+        isMonotonic: true,
+      });
+    }
+
+    for (const gauge of metrics.gauges) {
+      scopeMetrics.metrics.push({
+        descriptor: {
+          name: transformPrismaKey(gauge.key),
+          description: gauge.description,
+          unit: '1',
+          type: InstrumentType.UP_DOWN_COUNTER,
+          valueType: ValueType.INT,
+        },
+        dataPointType: DataPointType.GAUGE,
+        aggregationTemporality: AggregationTemporality.CUMULATIVE,
+        dataPoints: [
+          {
+            startTime: this.startTime,
+            endTime: endTime,
+            value: gauge.value,
+            attributes: gauge.labels,
+          },
+        ],
+      });
+    }
+
+    for (const histogram of metrics.histograms) {
+      const boundaries = [];
+      const counts = [];
+      for (const [boundary, count] of histogram.value.buckets) {
+        boundaries.push(boundary);
+        counts.push(count);
+      }
+      scopeMetrics.metrics.push({
+        descriptor: {
+          name: transformPrismaKey(histogram.key),
+          description: histogram.description,
+          unit: 'ms',
+          type: InstrumentType.HISTOGRAM,
+          valueType: ValueType.DOUBLE,
+        },
+        dataPointType: DataPointType.HISTOGRAM,
+        aggregationTemporality: AggregationTemporality.CUMULATIVE,
+        dataPoints: [
+          {
+            startTime: this.startTime,
+            endTime: endTime,
+            value: {
+              buckets: {
+                boundaries,
+                counts,
+              },
+              count: histogram.value.count,
+              sum: histogram.value.sum,
+            },
+            attributes: histogram.labels,
+          },
+        ],
+      });
+    }
+
+    result.resourceMetrics.scopeMetrics.push(scopeMetrics);
+
+    return result;
+  }
+}

packages/backend/server/src/base/metrics/utils.ts

@@ -0,0 +1,65 @@
+import type { Attributes } from '@opentelemetry/api';
+
+import { type KnownMetricScopes, metrics } from './metrics';
+
+/**
+ * Decorator for measuring the call time, record call count and if is throw of a function call
+ * @param scope metric scope
+ * @param name metric event name
+ * @param attrs attributes
+ * @returns
+ */
+export const CallMetric = (
+  scope: KnownMetricScopes,
+  name: string,
+  record?: { timer?: boolean; count?: boolean; error?: boolean },
+  attrs?: Attributes
+): MethodDecorator => {
+  // @ts-expect-error allow
+  return (
+    _target,
+    _key,
+    desc: TypedPropertyDescriptor<(...args: any[]) => any>
+  ) => {
+    const originalMethod = desc.value;
+    if (!originalMethod) {
+      return desc;
+    }
+
+    const timer = metrics[scope].histogram('function_timer', {
+      description: 'function call time costs',
+      unit: 'ms',
+    });
+    const count = metrics[scope].counter('function_calls', {
+      description: 'function call counter',
+    });
+
+    desc.value = async function (...args: any[]) {
+      const start = Date.now();
+      let error = false;
+
+      const end = () => {
+        timer?.record(Date.now() - start, { ...attrs, name, error });
+      };
+
+      try {
+        if (!record || !!record.count) {
+          count.add(1, attrs);
+        }
+        return await originalMethod.apply(this, args);
+      } catch (err) {
+        if (!record || !!record.error) {
+          error = true;
+        }
+        throw err;
+      } finally {
+        count.add(1, { ...attrs, name, error });
+        if (!record || !!record.timer) {
+          end();
+        }
+      }
+    };
+
+    return desc;
+  };
+};