Added creation and initialization helper methods for the existing replication, storage and API metrics

Added the MetricsEngine to the service context

Author: Rentacookie

packages/service-core/src/api/api-index.ts

@@ -1,3 +1,4 @@
 export * from './diagnostics.js';
 export * from './RouteAPI.js';
 export * from './schema.js';
+export * from './api-metrics.js';

packages/service-core/src/api/api-metrics.ts

@@ -0,0 +1,43 @@
+import { MetricsEngine } from '../metrics/MetricsEngine.js';
+
+export enum APIMetricType {
+  // Uncompressed size of synced data from PowerSync to Clients
+  DATA_SYNCED_BYTES = 'powersync_data_synced_bytes_total',
+  // Number of operations synced
+  OPERATIONS_SYNCED_TOTAL = 'powersync_operations_synced_total',
+  // Number of concurrent sync connections
+  CONCURRENT_CONNECTIONS = 'powersync_concurrent_connections'
+}
+
+/**
+ *  Create and register the core API metrics.
+ *  @param engine
+ */
+export function createCoreAPIMetrics(engine: MetricsEngine): void {
+  engine.createCounter({
+    name: APIMetricType.DATA_SYNCED_BYTES,
+    description: 'Uncompressed size of synced data',
+    unit: 'bytes'
+  });
+
+  engine.createCounter({
+    name: APIMetricType.OPERATIONS_SYNCED_TOTAL,
+    description: 'Number of operations synced'
+  });
+
+  engine.createUpDownCounter({
+    name: APIMetricType.CONCURRENT_CONNECTIONS,
+    description: 'Number of concurrent sync connections'
+  });
+}
+
+/**
+ *  Initialise the core API metrics. This should be called after the metrics have been created.
+ *  @param engine
+ */
+export function initializeCoreAPIMetrics(engine: MetricsEngine): void {
+  const concurrent_connections = engine.getUpDownCounter(APIMetricType.CONCURRENT_CONNECTIONS);
+
+  // Initialize the metric, so that it reports a value before connections have been opened.
+  concurrent_connections.add(0);
+}

packages/service-core/src/index.ts

@@ -12,8 +12,8 @@ export * as entry from './entry/entry-index.js';
 // Re-export framework for easy use of Container API
 export * as framework from '@powersync/lib-services-framework';
 
-export * from './metrics/Metrics.js';
-export * as metrics from './metrics/Metrics.js';
+export * from './metrics/metrics-index.js';
+export * as metrics from './metrics/metrics-index.js';
 
 export * from './migrations/migrations-index.js';
 export * as migrations from './migrations/migrations-index.js';

packages/service-core/src/replication/replication-index.ts

@@ -3,3 +3,4 @@ export * from './AbstractReplicator.js';
 export * from './ErrorRateLimiter.js';
 export * from './ReplicationEngine.js';
 export * from './ReplicationModule.js';
+export * from './replication-metrics.js';

packages/service-core/src/replication/replication-metrics.ts

@@ -0,0 +1,55 @@
+import { MetricsEngine } from '../metrics/metrics-index.js';
+
+export enum ReplicationMetricType {
+  // Uncompressed size of replicated data from data source to PowerSync
+  DATA_REPLICATED_BYTES = 'powersync_data_replicated_bytes_total',
+  // Total number of replicated rows. Not used for pricing.
+  ROWS_REPLICATED_TOTAL = 'powersync_rows_replicated_total',
+  // Total number of replicated transactions. Not used for pricing.
+  TRANSACTIONS_REPLICATED_TOTAL = 'powersync_transactions_replicated_total',
+  // Total number of replication chunks. Not used for pricing.
+  CHUNKS_REPLICATED_TOTAL = 'powersync_chunks_replicated_total'
+}
+
+/**
+ *  Create and register the core replication metrics.
+ *  @param engine
+ */
+export function createCoreReplicationMetrics(engine: MetricsEngine): void {
+  engine.createCounter({
+    name: ReplicationMetricType.DATA_REPLICATED_BYTES,
+    description: 'Uncompressed size of replicated data',
+    unit: 'bytes'
+  });
+
+  engine.createCounter({
+    name: ReplicationMetricType.ROWS_REPLICATED_TOTAL,
+    description: 'Total number of replicated rows'
+  });
+
+  engine.createCounter({
+    name: ReplicationMetricType.TRANSACTIONS_REPLICATED_TOTAL,
+    description: 'Total number of replicated transactions'
+  });
+
+  engine.createCounter({
+    name: ReplicationMetricType.CHUNKS_REPLICATED_TOTAL,
+    description: 'Total number of replication chunks'
+  });
+}
+
+/**
+ *  Initialise the core replication metrics. This should be called after the metrics have been created.
+ *  @param engine
+ */
+export function initializeCoreReplicationMetrics(engine: MetricsEngine): void {
+  const data_replicated_bytes = engine.getCounter(ReplicationMetricType.DATA_REPLICATED_BYTES);
+  const rows_replicated_total = engine.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL);
+  const transactions_replicated_total = engine.getCounter(ReplicationMetricType.TRANSACTIONS_REPLICATED_TOTAL);
+  const chunks_replicated_total = engine.getCounter(ReplicationMetricType.CHUNKS_REPLICATED_TOTAL);
+
+  data_replicated_bytes.add(0);
+  rows_replicated_total.add(0);
+  transactions_replicated_total.add(0);
+  chunks_replicated_total.add(0);
+}

packages/service-core/src/storage/storage-index.ts

@@ -6,4 +6,5 @@ export * from './SourceEntity.js';
 export * from './SourceTable.js';
 export * from './StorageEngine.js';
 export * from './StorageProvider.js';
+export * from './storage-metrics.js';
 export * from './WriteCheckpointAPI.js';

packages/service-core/src/storage/storage-metrics.ts

@@ -0,0 +1,75 @@
+import { MetricsEngine } from '../metrics/MetricsEngine.js';
+import { BucketStorageFactory, StorageMetrics } from './BucketStorage.js';
+import { logger } from '@powersync/lib-services-framework';
+
+export enum StorageMetricType {
+  // Size of current replication data stored in PowerSync
+  REPLICATION_SIZE_BYTES = 'powersync_replication_storage_size_bytes',
+  // Size of operations data stored in PowerSync
+  OPERATION_SIZE_BYTES = 'powersync_operation_storage_size_bytes',
+  // Size of parameter data stored in PowerSync
+  PARAMETER_SIZE_BYTES = 'powersync_parameter_storage_size_bytes'
+}
+
+export function createCoreStorageMetrics(engine: MetricsEngine): void {
+  engine.createObservableGauge({
+    name: StorageMetricType.REPLICATION_SIZE_BYTES,
+    description: 'Size of current data stored in PowerSync',
+    unit: 'bytes'
+  });
+
+  engine.createObservableGauge({
+    name: StorageMetricType.OPERATION_SIZE_BYTES,
+    description: 'Size of operations stored in PowerSync',
+    unit: 'bytes'
+  });
+
+  engine.createObservableGauge({
+    name: StorageMetricType.PARAMETER_SIZE_BYTES,
+    description: 'Size of parameter data stored in PowerSync',
+    unit: 'bytes'
+  });
+}
+
+export function initializeCoreStorageMetrics(engine: MetricsEngine, storage: BucketStorageFactory): void {
+  const replication_storage_size_bytes = engine.getObservableGauge(StorageMetricType.REPLICATION_SIZE_BYTES);
+  const operation_storage_size_bytes = engine.getObservableGauge(StorageMetricType.OPERATION_SIZE_BYTES);
+  const parameter_storage_size_bytes = engine.getObservableGauge(StorageMetricType.PARAMETER_SIZE_BYTES);
+
+  const MINIMUM_INTERVAL = 60_000;
+
+  let cachedRequest: Promise<StorageMetrics | null> | undefined = undefined;
+  let cacheTimestamp = 0;
+
+  const getMetrics = () => {
+    if (cachedRequest == null || Date.now() - cacheTimestamp > MINIMUM_INTERVAL) {
+      cachedRequest = storage.getStorageMetrics().catch((e) => {
+        logger.error(`Failed to get storage metrics`, e);
+        return null;
+      });
+      cacheTimestamp = Date.now();
+    }
+    return cachedRequest;
+  };
+
+  replication_storage_size_bytes.setValueProvider(async () => {
+    const metrics = await getMetrics();
+    if (metrics) {
+      return metrics.replication_size_bytes;
+    }
+  });
+
+  operation_storage_size_bytes.setValueProvider(async () => {
+    const metrics = await getMetrics();
+    if (metrics) {
+      return metrics.operations_size_bytes;
+    }
+  });
+
+  parameter_storage_size_bytes.setValueProvider(async () => {
+    const metrics = await getMetrics();
+    if (metrics) {
+      return metrics.parameters_size_bytes;
+    }
+  });
+}

packages/service-core/src/system/ServiceContext.ts

@@ -1,7 +1,7 @@
 import { LifeCycledSystem, MigrationManager, ServiceIdentifier, container } from '@powersync/lib-services-framework';
 
 import { framework } from '../index.js';
-import * as metrics from '../metrics/Metrics.js';
+import * as metrics from '../metrics/MetricsEngine.js';
 import { PowerSyncMigrationManager } from '../migrations/PowerSyncMigrationManager.js';
 import * as replication from '../replication/replication-index.js';
 import * as routes from '../routes/routes-index.js';
@@ -11,7 +11,7 @@ import * as utils from '../util/util-index.js';
 export interface ServiceContext {
   configuration: utils.ResolvedPowerSyncConfig;
   lifeCycleEngine: LifeCycledSystem;
-  metrics: metrics.Metrics | null;
+  metricsEngine: metrics.MetricsEngine;
   replicationEngine: replication.ReplicationEngine | null;
   routerEngine: routes.RouterEngine | null;
   storageEngine: storage.StorageEngine;
@@ -34,6 +34,11 @@ export class ServiceContextContainer implements ServiceContext {
       configuration
     });
 
+    this.lifeCycleEngine.withLifecycle(this.storageEngine, {
+      start: (storageEngine) => storageEngine.start(),
+      stop: (storageEngine) => storageEngine.shutDown()
+    });
+
     const migrationManager = new MigrationManager();
     container.register(framework.ContainerImplementation.MIGRATION_MANAGER, migrationManager);
 
@@ -56,8 +61,8 @@ export class ServiceContextContainer implements ServiceContext {
     return container.getOptional(routes.RouterEngine);
   }
 
-  get metrics(): metrics.Metrics | null {
-    return container.getOptional(metrics.Metrics);
+  get metricsEngine(): metrics.MetricsEngine {
+    return container.getImplementation(metrics.MetricsEngine);
   }
 
   get migrations(): PowerSyncMigrationManager {