feat(metrics): 3 Prometheus counters for scaling dive (#7756) (#7762)

* feat(metrics): expose 3 Prometheus counters for the scaling dive

Per the spec section 6 of #7756: enables the load-test harness to
attribute *where* time goes on the server, not just the gauge headline
(CPU / event-loop / memory) the dive doc starts from.

New /stats/prometheus rows:

- etherpad_pad_users{padId} — gauge, derived from sessioninfos on
  each scrape. Lets the harness confirm the pad it points at actually
  has the expected concurrency.

- etherpad_changeset_apply_duration_seconds — histogram observed
  inside handleUserChanges. Separates "apply path is slow" from
  "fan-out is slow" when latency rises.

- etherpad_socket_emits_total{type} — counter at the broadcast
  emit sites (handleCustomObjectMessage, handleCustomMessage,
  sendChatMessageToPadClients) and inside the NEW_CHANGES per-socket
  loop in updatePadClients. Bucketed by message type so the harness
  can measure the amplification factor of each lever (especially the
  fan-out batching lever).

Metric handles live in a new prom-instruments.ts module rather than
in prometheus.ts itself, so PadMessageHandler can import the
recording helpers without creating a circular dependency
(prometheus.ts already requires PadMessageHandler).

Tests: smoke test verifies recordSocketEmit + recordChangesetApply
move the underlying counters/histogram.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(metrics): address Qodo review — flag-gate, scope histogram, bound label cardinality

Three issues raised on the initial PR:

1. **Feature flag.** Per project compliance rule, new features must
   be behind a flag and disabled by default. Adds
   `settings.scalingDiveMetrics` (default `false`). When off,
   recordSocketEmit() / recordChangesetApply() short-circuit to
   no-ops and the metrics are never even registered with the
   Prometheus register. Enable only when running the
   ether/etherpad-load-test scaling-dive harness.

2. **Histogram scope.** Previously the
   etherpad_changeset_apply_duration_seconds timer wrapped the
   whole handleUserChanges() body — including
   `await exports.updatePadClients(pad)` — so the histogram
   measured apply+fan-out, defeating its stated purpose. Now
   stopped immediately after the apply work (`assert.equal(...rev,
   r)`), before the ACCEPT_COMMIT socket emit and the
   updatePadClients call. Failed applies deliberately don't observe
   so the success-path distribution stays clean.

3. **Label cardinality.** handleCustomMessage was passing the
   user-supplied msgString (an HTTP-API param) directly as the
   `type` label value. A misbehaving API caller could grow
   prom-client's internal label map until OOM. Now bucketed against
   a known-types allowlist; anything outside it lands in `other`.

Tests updated: 5/5 — covers happy path, "other" bucketing of
unknown/unsafe labels, and that the flag-disabled state is a true
no-op.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/node/handler/PadMessageHandler.ts

@@ -47,6 +47,7 @@ const accessLogger = log4js.getLogger('access');
 const hooks = require('../../static/js/pluginfw/hooks');
 const stats = require('../stats')
 const assert = require('assert').strict;
+import {recordChangesetApply, recordSocketEmit} from '../prom-instruments';
 import {RateLimiterMemory} from 'rate-limiter-flexible';
 import {ChangesetRequest, PadUserInfo, SocketClientRequest} from "../types/SocketClientRequest";
 import {APool, AText, PadAuthor, PadType} from "../types/PadType";
@@ -117,6 +118,20 @@ function getActivePadCountFromSessionInfos() {
 }
 exports.getActivePadCountFromSessionInfos = getActivePadCountFromSessionInfos;
 
+// Per-pad user counts derived on demand from sessioninfos. Used by
+// prometheus.ts to populate `etherpad_pad_users{padId}` so the #7756
+// scaling-dive harness can confirm the pad it's pointing at actually
+// has the expected concurrency.
+function getPadUsersMap(): Map<string, number> {
+  const out = new Map<string, number>();
+  for (const {padId} of Object.values(sessioninfos)) {
+    if (!padId) continue;
+    out.set(padId, (out.get(padId) ?? 0) + 1);
+  }
+  return out;
+}
+exports.getPadUsersMap = getPadUsersMap;
+
 /**
  * Build a sanitized copy of the plugins registry suitable for sending to the
  * client as part of clientVars. The shape is preserved but each plugin's
@@ -625,6 +640,7 @@ exports.handleCustomObjectMessage = (msg: CustomMessage, sessionID: string) => {
     } else {
       // broadcast to all clients on this pad
       socketio.sockets.in(msg.data.payload.padId).emit('message', msg);
+      recordSocketEmit(msg.data.type);
     }
   }
 };
@@ -645,6 +661,7 @@ exports.handleCustomMessage = (padID: string, msgString:string) => {
     },
   };
   socketio.sockets.in(padID).emit('message', msg);
+  recordSocketEmit(msg.data.type);
 };
 
 /**
@@ -685,6 +702,7 @@ exports.sendChatMessageToPadClients = async (mt: ChatMessage|number, puId: strin
     type: 'COLLABROOM',
     data: {type: 'CHAT_MESSAGE', message},
   });
+  recordSocketEmit('CHAT_MESSAGE');
   await promise;
 };
 
@@ -802,8 +820,14 @@ const handleUserChanges = async (socket:any, message: {
   // if the session was valid when the message arrived in the first place
   if (!thisSession) throw new Error('client disconnected');
 
-  // Measure time to process edit
+  // Measure time to process edit. stats.timer('edits') spans the full handler
+  // (apply + fan-out) for backwards-compat; the new Prometheus histogram below
+  // wraps only the apply path so the scaling-dive harness can distinguish
+  // "apply is slow" from "fan-out is slow". Failed applies do not call the
+  // stopper — leaving the timer un-observed keeps the success-path
+  // distribution clean.
   const stopWatch = stats.timer('edits').start();
+  const stopApplyHistogram = recordChangesetApply();
   try {
     const {data: {baseRev, apool, changeset}} = message;
     if (baseRev == null) throw new Error('missing baseRev');
@@ -905,6 +929,10 @@ const handleUserChanges = async (socket:any, message: {
     // The client assumes that ACCEPT_COMMIT and NEW_CHANGES messages arrive in order. Make sure we
     // have already sent any previous ACCEPT_COMMIT and NEW_CHANGES messages.
     assert.equal(thisSession.rev, r);
+    // End of the apply path. The Prometheus histogram observes here so that
+    // fan-out (socket emit + updatePadClients) does NOT inflate the apply
+    // duration. Failed applies are deliberately not recorded.
+    stopApplyHistogram();
     socket.emit('message', {type: 'COLLABROOM', data: {type: 'ACCEPT_COMMIT', newRev}});
     thisSession.rev = newRev;
     if (newRev !== r) thisSession.time = await pad.getRevisionDate(newRev);
@@ -968,6 +996,7 @@ exports.updatePadClients = async (pad: PadType) => {
       };
       try {
         socket.emit('message', msg);
+        recordSocketEmit('NEW_CHANGES');
       } catch (err:any) {
         messageLogger.error(`Failed to notify user of new revision: ${err.stack || err}`);
         return;

src/node/prom-instruments.ts

@@ -0,0 +1,67 @@
+// Prometheus instruments referenced from the hot path (PadMessageHandler).
+//
+// Defined in a separate file so that PadMessageHandler can import the
+// recording helpers without creating a circular import with prometheus.ts
+// (which already requires PadMessageHandler to read sessioninfos).
+//
+// The metrics themselves are added to the central Registry by prometheus.ts.
+//
+// Everything here is gated behind settings.scalingDiveMetrics (default false).
+// When the flag is off the recording helpers short-circuit to no-ops and the
+// metrics are never registered, so production deployments don't pay for
+// instrumentation they don't use.
+
+import client from 'prom-client';
+import settings from './utils/Settings';
+
+export const enabled = (): boolean => settings.scalingDiveMetrics === true;
+
+export const changesetApplyDuration = new client.Histogram({
+  name: 'etherpad_changeset_apply_duration_seconds',
+  help: 'Time spent applying an incoming USER_CHANGES message on the server (apply path only, excludes fan-out to other clients)',
+  buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 5],
+});
+
+export const socketEmitsTotal = new client.Counter({
+  name: 'etherpad_socket_emits_total',
+  help: 'Number of socket.io broadcast emits, bucketed by message type',
+  labelNames: ['type'],
+});
+
+export const padUsersGauge = new client.Gauge({
+  name: 'etherpad_pad_users',
+  help: 'Active users connected to a pad, keyed by padId',
+  labelNames: ['padId'],
+});
+
+// Allowlist of message-type label values. Anything outside this set is rolled
+// into 'other' so a misbehaving plugin or HTTP-API caller passing a
+// user-controlled msgString cannot explode prom-client's internal label-cardinality
+// state.
+const KNOWN_TYPES = new Set([
+  'NEW_CHANGES',
+  'ACCEPT_COMMIT',
+  'CHAT_MESSAGE',
+  'CLIENT_VARS',
+  'CLIENT_MESSAGE',
+  'CUSTOM',
+  'USER_NEWINFO',
+  'USERINFO_UPDATE',
+  'USER_LEAVE',
+]);
+
+/** Start a timer for the changeset apply path. Call the returned function when done.
+ *  Returns a no-op stopper when the feature flag is off. */
+export const recordChangesetApply = (): (() => void) => {
+  if (!enabled()) return () => {};
+  return changesetApplyDuration.startTimer();
+};
+
+/** Increment the socket-emit counter for the given message type.
+ *  No-op when the feature flag is off. Unknown/missing types are bucketed as
+ *  'other' to keep label cardinality bounded. */
+export const recordSocketEmit = (type: string | undefined): void => {
+  if (!enabled()) return;
+  const label = type && KNOWN_TYPES.has(type) ? type : 'other';
+  socketEmitsTotal.labels(label).inc();
+};

src/node/prometheus.ts

@@ -23,6 +23,19 @@ const activePadsGauge = new client.Gauge({
 });
 register.registerMetric(activePadsGauge);
 
+// Added for the #7756 scaling dive: lets the load-test harness attribute
+// where time goes (apply path vs. fan-out) and confirm per-pad concurrency.
+// The metric handles live in prom-instruments.ts to avoid a circular import
+// with PadMessageHandler (which records into them on the hot path).
+// Gated behind settings.scalingDiveMetrics so production deployments don't
+// pay for the instrumentation by default.
+import {padUsersGauge, changesetApplyDuration, socketEmitsTotal, enabled as scalingDiveMetricsEnabled} from './prom-instruments';
+if (scalingDiveMetricsEnabled()) {
+  register.registerMetric(padUsersGauge);
+  register.registerMetric(changesetApplyDuration);
+  register.registerMetric(socketEmitsTotal);
+}
+
 client.collectDefaultMetrics({register});
 
 const monitor = async function () {
@@ -32,6 +45,13 @@ const monitor = async function () {
   }
   activePadsGauge.set(PadMessageHandler.getActivePadCountFromSessionInfos());
   totalUsersGauge.set(PadMessageHandler.getTotalActiveUsers());
+  if (scalingDiveMetricsEnabled()) {
+    // Per-pad concurrency: reset to avoid stale labels for pads that drained.
+    padUsersGauge.reset();
+    for (const [padId, count] of PadMessageHandler.getPadUsersMap()) {
+      padUsersGauge.set({padId}, count);
+    }
+  }
   return register;
 };
 

src/node/utils/Settings.ts

@@ -271,6 +271,7 @@ export type SettingsType = {
   ipLogging: 'full' | 'truncated' | 'anonymous',
   automaticReconnectionTimeout: number,
   loadTest: boolean,
+  scalingDiveMetrics: boolean,
   dumpOnUncleanExit: boolean,
   indentationOnNewLine: boolean,
   logconfig: any | null,
@@ -650,6 +651,13 @@ const settings: SettingsType = {
    * Disable Load Testing
    */
   loadTest: false,
+  /**
+   * Expose extra Prometheus metrics designed for the scaling-dive load-test harness
+   * (ether/etherpad#7756): etherpad_pad_users{padId}, etherpad_changeset_apply_duration_seconds,
+   * etherpad_socket_emits_total{type}. Default false — enable only when running the harness so
+   * production deployments aren't paying for instrumentation they don't use.
+   */
+  scalingDiveMetrics: false,
   /**
    * Disable dump of objects preventing a clean exit
    */

src/tests/backend-new/specs/prom-instruments.test.ts

@@ -0,0 +1,78 @@
+// Smoke test for the Prometheus instruments added for #7756. Verifies the
+// recording helpers actually move the underlying metrics so the load-test
+// harness can rely on them, AND that the settings.scalingDiveMetrics flag
+// gates everything off when disabled.
+
+import {describe, it, expect, beforeEach, afterEach} from 'vitest';
+import settings from '../../../node/utils/Settings';
+import {
+  recordChangesetApply,
+  recordSocketEmit,
+  changesetApplyDuration,
+  socketEmitsTotal,
+} from '../../../node/prom-instruments';
+
+const originalFlag = settings.scalingDiveMetrics;
+
+beforeEach(() => {
+  socketEmitsTotal.reset();
+  changesetApplyDuration.reset();
+  settings.scalingDiveMetrics = true;
+});
+
+afterEach(() => { settings.scalingDiveMetrics = originalFlag; });
+
+describe('recordSocketEmit (flag enabled)', () => {
+  it('increments etherpad_socket_emits_total keyed by message type', async () => {
+    recordSocketEmit('NEW_CHANGES');
+    recordSocketEmit('NEW_CHANGES');
+    recordSocketEmit('CHAT_MESSAGE');
+    const values = await socketEmitsTotal.get();
+    const byType: Record<string, number> = {};
+    for (const v of values.values) byType[v.labels.type as string] = v.value;
+    expect(byType['NEW_CHANGES']).toBe(2);
+    expect(byType['CHAT_MESSAGE']).toBe(1);
+  });
+
+  it('buckets unknown / user-supplied label values as "other" to keep cardinality bounded', async () => {
+    recordSocketEmit(undefined);
+    recordSocketEmit('attacker-supplied-string-1');
+    recordSocketEmit('attacker-supplied-string-2');
+    const values = await socketEmitsTotal.get();
+    const byType: Record<string, number> = {};
+    for (const v of values.values) byType[v.labels.type as string] = v.value;
+    expect(byType['other']).toBe(3);
+    // No labels for the attacker strings — proves the allowlist is enforced.
+    expect(Object.keys(byType)).not.toContain('attacker-supplied-string-1');
+    expect(Object.keys(byType)).not.toContain('attacker-supplied-string-2');
+  });
+});
+
+describe('recordChangesetApply (flag enabled)', () => {
+  it('observes a duration in etherpad_changeset_apply_duration_seconds', async () => {
+    const end = recordChangesetApply();
+    await new Promise((r) => setTimeout(r, 5));
+    end();
+    const values = await changesetApplyDuration.get();
+    const countRow = values.values.find((v) => v.metricName === 'etherpad_changeset_apply_duration_seconds_count');
+    expect(countRow?.value).toBeGreaterThan(0);
+  });
+});
+
+describe('feature flag (disabled by default)', () => {
+  beforeEach(() => { settings.scalingDiveMetrics = false; });
+
+  it('recordSocketEmit is a no-op', async () => {
+    recordSocketEmit('NEW_CHANGES');
+    const values = await socketEmitsTotal.get();
+    expect(values.values.length).toBe(0);
+  });
+
+  it('recordChangesetApply returns a no-op stopper that does not observe', async () => {
+    const end = recordChangesetApply();
+    end();
+    const values = await changesetApplyDuration.get();
+    const countRow = values.values.find((v) => v.metricName === 'etherpad_changeset_apply_duration_seconds_count');
+    expect(countRow?.value ?? 0).toBe(0);
+  });
+});