Updated replication modules to use the MetricsEngine instead of the old singleton metrics instance

Author: Rentacookie

modules/module-mongodb/src/module/MongoModule.ts

@@ -37,11 +37,14 @@ export class MongoModule extends replication.ReplicationModule<types.MongoConnec
       id: this.getDefaultId(normalisedConfig.database ?? ''),
       syncRuleProvider: syncRuleProvider,
       storageEngine: context.storageEngine,
+      metricsEngine: context.metricsEngine,
       connectionFactory: connectionFactory,
       rateLimiter: new MongoErrorRateLimiter()
     });
   }
 
+  async onInitialized(context: system.ServiceContextContainer): Promise<void> {}
+
   /**
    * Combines base config with normalized connection settings
    */

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -8,7 +8,14 @@ import {
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
-import { Metrics, SaveOperationTag, SourceEntityDescriptor, SourceTable, storage } from '@powersync/service-core';
+import {
+  MetricsEngine,
+  ReplicationMetricType,
+  SaveOperationTag,
+  SourceEntityDescriptor,
+  SourceTable,
+  storage
+} from '@powersync/service-core';
 import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
 import { MongoLSN } from '../common/MongoLSN.js';
 import { PostImagesOption } from '../types/types.js';
@@ -20,6 +27,7 @@ import { CHECKPOINTS_COLLECTION } from './replication-utils.js';
 export interface ChangeStreamOptions {
   connections: MongoManager;
   storage: storage.SyncRulesBucketStorage;
+  metrics: MetricsEngine;
   abort_signal: AbortSignal;
 }
 
@@ -52,13 +60,15 @@ export class ChangeStream {
   private connections: MongoManager;
   private readonly client: mongo.MongoClient;
   private readonly defaultDb: mongo.Db;
+  private readonly metrics: MetricsEngine;
 
   private abort_signal: AbortSignal;
 
   private relation_cache = new Map<string | number, storage.SourceTable>();
 
   constructor(options: ChangeStreamOptions) {
     this.storage = options.storage;
+    this.metrics = options.metrics;
     this.group_id = options.storage.group_id;
     this.connections = options.connections;
     this.client = this.connections.client;
@@ -322,7 +332,7 @@ export class ChangeStream {
         logger.info(`[${this.group_id}] Replicating ${table.qualifiedName} ${at}/${estimatedCount}`);
         lastLogIndex = at;
       }
-      Metrics.getInstance().rows_replicated_total.add(1);
+      this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
 
       await touch();
     }
@@ -438,7 +448,7 @@ export class ChangeStream {
       return null;
     }
 
-    Metrics.getInstance().rows_replicated_total.add(1);
+    this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
     if (change.operationType == 'insert') {
       const baseRecord = constructAfterRecord(change.fullDocument);
       return await batch.save({

modules/module-mongodb/src/replication/ChangeStreamReplicationJob.ts

@@ -71,6 +71,7 @@ export class ChangeStreamReplicationJob extends replication.AbstractReplicationJ
       const stream = new ChangeStream({
         abort_signal: this.abortController.signal,
         storage: this.options.storage,
+        metrics: this.options.metrics,
         connections: connectionManager
       });
       await stream.replicate();

modules/module-mongodb/src/replication/ChangeStreamReplicator.ts

@@ -20,6 +20,7 @@ export class ChangeStreamReplicator extends replication.AbstractReplicator<Chang
     return new ChangeStreamReplicationJob({
       id: this.createJobId(options.storage.group_id),
       storage: options.storage,
+      metrics: this.metrics,
       connectionFactory: this.connectionFactory,
       lock: options.lock,
       rateLimiter: new MongoErrorRateLimiter()

modules/module-mysql/src/module/MySQLModule.ts

@@ -25,9 +25,7 @@ export class MySQLModule extends replication.ReplicationModule<types.MySQLConnec
     });
   }
 
-  async initialize(context: system.ServiceContextContainer): Promise<void> {
-    await super.initialize(context);
-  }
+  async onInitialized(context: system.ServiceContextContainer): Promise<void> {}
 
   protected createRouteAPIAdapter(): api.RouteAPI {
     return new MySQLRouteAPIAdapter(this.resolveConfig(this.decodedConfig!));
@@ -42,6 +40,7 @@ export class MySQLModule extends replication.ReplicationModule<types.MySQLConnec
       id: this.getDefaultId(normalisedConfig.database),
       syncRuleProvider: syncRuleProvider,
       storageEngine: context.storageEngine,
+      metricsEngine: context.metricsEngine,
       connectionFactory: connectionFactory,
       rateLimiter: new MySQLErrorRateLimiter()
     });

modules/module-mysql/src/replication/BinLogReplicationJob.ts

@@ -63,6 +63,7 @@ export class BinLogReplicationJob extends replication.AbstractReplicationJob {
       const stream = new BinLogStream({
         abortSignal: this.abortController.signal,
         storage: this.options.storage,
+        metrics: this.options.metrics,
         connections: connectionManager
       });
       await stream.replicate();

modules/module-mysql/src/replication/BinLogReplicator.ts

@@ -19,6 +19,7 @@ export class BinLogReplicator extends replication.AbstractReplicator<BinLogRepli
     return new BinLogReplicationJob({
       id: this.createJobId(options.storage.group_id),
       storage: options.storage,
+      metrics: this.metrics,
       lock: options.lock,
       connectionFactory: this.connectionFactory,
       rateLimiter: this.rateLimiter

modules/module-mysql/src/replication/BinLogStream.ts

@@ -2,7 +2,14 @@ import { logger, ReplicationAbortedError, ReplicationAssertionError } from '@pow
 import * as sync_rules from '@powersync/service-sync-rules';
 import async from 'async';
 
-import { ColumnDescriptor, framework, getUuidReplicaIdentityBson, Metrics, storage } from '@powersync/service-core';
+import {
+  ColumnDescriptor,
+  framework,
+  getUuidReplicaIdentityBson,
+  MetricsEngine,
+  ReplicationMetricType,
+  storage
+} from '@powersync/service-core';
 import mysql, { FieldPacket } from 'mysql2';
 
 import { BinLogEvent, StartOptions, TableMapEntry } from '@powersync/mysql-zongji';
@@ -16,6 +23,7 @@ import * as zongji_utils from './zongji/zongji-utils.js';
 export interface BinLogStreamOptions {
   connections: MySQLConnectionManager;
   storage: storage.SyncRulesBucketStorage;
+  metrics: MetricsEngine;
   abortSignal: AbortSignal;
 }
 
@@ -62,7 +70,7 @@ export class BinLogStream {
 
   private tableCache = new Map<string | number, storage.SourceTable>();
 
-  constructor(protected options: BinLogStreamOptions) {
+  constructor(private options: BinLogStreamOptions) {
     this.storage = options.storage;
     this.connections = options.connections;
     this.syncRules = options.storage.getParsedSyncRules({ defaultSchema: this.defaultSchema });
@@ -74,6 +82,10 @@ export class BinLogStream {
     return this.connections.connectionTag;
   }
 
+  private get metrics() {
+    return this.options.metrics;
+  }
+
   get connectionId() {
     const { connectionId } = this.connections;
     // Default to 1 if not set
@@ -334,6 +346,8 @@ AND table_type = 'BASE TABLE';`,
         after: record,
         afterReplicaId: getUuidReplicaIdentityBson(record, table.replicaIdColumns)
       });
+
+      this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
     }
     await batch.flush();
   }
@@ -460,7 +474,7 @@ AND table_type = 'BASE TABLE';`,
                 });
                 break;
               case zongji_utils.eventIsXid(evt):
-                Metrics.getInstance().transactions_replicated_total.add(1);
+                this.metrics.getCounter(ReplicationMetricType.TRANSACTIONS_REPLICATED_TOTAL).add(1);
                 // Need to commit with a replicated GTID with updated next position
                 await batch.commit(
                   new common.ReplicatedGTID({
@@ -602,7 +616,7 @@ AND table_type = 'BASE TABLE';`,
   ): Promise<storage.FlushedResult | null> {
     switch (payload.type) {
       case storage.SaveOperationTag.INSERT:
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         const record = common.toSQLiteRow(payload.data, payload.columns);
         return await batch.save({
           tag: storage.SaveOperationTag.INSERT,
@@ -613,7 +627,7 @@ AND table_type = 'BASE TABLE';`,
           afterReplicaId: getUuidReplicaIdentityBson(record, payload.sourceTable.replicaIdColumns)
         });
       case storage.SaveOperationTag.UPDATE:
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         // "before" may be null if the replica id columns are unchanged
         // It's fine to treat that the same as an insert.
         const beforeUpdated = payload.previous_data
@@ -633,7 +647,7 @@ AND table_type = 'BASE TABLE';`,
         });
 
       case storage.SaveOperationTag.DELETE:
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         const beforeDeleted = common.toSQLiteRow(payload.data, payload.columns);
 
         return await batch.save({

modules/module-postgres/src/module/PostgresModule.ts

@@ -5,6 +5,7 @@ import {
   ConnectionTestResult,
   modules,
   replication,
+  ReplicationMetricType,
   system
 } from '@powersync/service-core';
 import * as jpgwire from '@powersync/service-jpgwire';
@@ -29,9 +30,7 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     });
   }
 
-  async initialize(context: system.ServiceContextContainer): Promise<void> {
-    await super.initialize(context);
-
+  async onInitialized(context: system.ServiceContextContainer): Promise<void> {
     const client_auth = context.configuration.base_config.client_auth;
 
     if (client_auth?.supabase && client_auth?.supabase_jwt_secret == null) {
@@ -46,13 +45,11 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     }
 
     // Record replicated bytes using global jpgwire metrics.
-    if (context.metrics) {
-      jpgwire.setMetricsRecorder({
-        addBytesRead(bytes) {
-          context.metrics!.data_replicated_bytes.add(bytes);
-        }
-      });
-    }
+    jpgwire.setMetricsRecorder({
+      addBytesRead(bytes) {
+        context.metricsEngine.getCounter(ReplicationMetricType.DATA_REPLICATED_BYTES).add(bytes);
+      }
+    });
   }
 
   protected createRouteAPIAdapter(): api.RouteAPI {
@@ -68,6 +65,7 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
       id: this.getDefaultId(normalisedConfig.database),
       syncRuleProvider: syncRuleProvider,
       storageEngine: context.storageEngine,
+      metricsEngine: context.metricsEngine,
       connectionFactory: connectionFactory,
       rateLimiter: new PostgresErrorRateLimiter()
     });

modules/module-postgres/src/replication/WalStream.ts

@@ -8,7 +8,13 @@ import {
   ReplicationAbortedError,
   ReplicationAssertionError
 } from '@powersync/lib-services-framework';
-import { getUuidReplicaIdentityBson, Metrics, SourceEntityDescriptor, storage } from '@powersync/service-core';
+import {
+  getUuidReplicaIdentityBson,
+  MetricsEngine,
+  ReplicationMetricType,
+  SourceEntityDescriptor,
+  storage
+} from '@powersync/service-core';
 import * as pgwire from '@powersync/service-jpgwire';
 import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern, toSyncRulesRow } from '@powersync/service-sync-rules';
 import * as pg_utils from '../utils/pgwire_utils.js';
@@ -20,6 +26,7 @@ import { checkSourceConfiguration, getReplicationIdentityColumns } from './repli
 export interface WalStreamOptions {
   connections: PgManager;
   storage: storage.SyncRulesBucketStorage;
+  metrics: MetricsEngine;
   abort_signal: AbortSignal;
 }
 
@@ -72,7 +79,7 @@ export class WalStream {
   connection_id = 1;
 
   private readonly storage: storage.SyncRulesBucketStorage;
-
+  private readonly metrics: MetricsEngine;
   private readonly slot_name: string;
 
   private connections: PgManager;
@@ -85,6 +92,7 @@ export class WalStream {
 
   constructor(options: WalStreamOptions) {
     this.storage = options.storage;
+    this.metrics = options.metrics;
     this.sync_rules = options.storage.getParsedSyncRules({ defaultSchema: POSTGRES_DEFAULT_SCHEMA });
     this.group_id = options.storage.group_id;
     this.slot_name = options.storage.slot_name;
@@ -471,7 +479,7 @@ WHERE  oid = $1::regclass`,
       }
 
       at += rows.length;
-      Metrics.getInstance().rows_replicated_total.add(rows.length);
+      this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(rows.length);
 
       await touch();
     }
@@ -562,7 +570,7 @@ WHERE  oid = $1::regclass`,
       }
 
       if (msg.tag == 'insert') {
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         const baseRecord = pg_utils.constructAfterRecord(msg);
         return await batch.save({
           tag: storage.SaveOperationTag.INSERT,
@@ -573,7 +581,7 @@ WHERE  oid = $1::regclass`,
           afterReplicaId: getUuidReplicaIdentityBson(baseRecord, table.replicaIdColumns)
         });
       } else if (msg.tag == 'update') {
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         // "before" may be null if the replica id columns are unchanged
         // It's fine to treat that the same as an insert.
         const before = pg_utils.constructBeforeRecord(msg);
@@ -587,7 +595,7 @@ WHERE  oid = $1::regclass`,
           afterReplicaId: getUuidReplicaIdentityBson(after, table.replicaIdColumns)
         });
       } else if (msg.tag == 'delete') {
-        Metrics.getInstance().rows_replicated_total.add(1);
+        this.metrics.getCounter(ReplicationMetricType.ROWS_REPLICATED_TOTAL).add(1);
         const before = pg_utils.constructBeforeRecord(msg)!;
 
         return await batch.save({
@@ -697,7 +705,7 @@ WHERE  oid = $1::regclass`,
             } else if (msg.tag == 'begin') {
               inTx = true;
             } else if (msg.tag == 'commit') {
-              Metrics.getInstance().transactions_replicated_total.add(1);
+              this.metrics.getCounter(ReplicationMetricType.TRANSACTIONS_REPLICATED_TOTAL).add(1);
               inTx = false;
               await batch.commit(msg.lsn!, { createEmptyCheckpoints });
               await this.ack(msg.lsn!, replicationStream);
@@ -739,7 +747,7 @@ WHERE  oid = $1::regclass`,
             await this.ack(chunkLastLsn, replicationStream);
           }
 
-          Metrics.getInstance().chunks_replicated_total.add(1);
+          this.metrics.getCounter(ReplicationMetricType.CHUNKS_REPLICATED_TOTAL).add(1);
         }
       }
     );