feat(analytics): add replicated mode to get_analytics

Add a `replicated: true` option to the `get_analytics` operation that fans
the query out to every peer node and merges each node's analytics into one
cluster-wide result set. The local query is streamed first, then each peer's
rows (already labeled with their origin `node` attribute) are appended.

The operation is forwarded to peers with `replicated` cleared so they only
return their own local metrics (no recursive fan-out), mirroring the existing
`restart` fan-out convention. Fan-out is best-effort: a peer that errors is
logged and omitted rather than failing the whole query. It is skipped when
running standalone (no peers) and when `hdb_analytics` already replicates
across the cluster (`analytics_replicate`), which would otherwise double-count.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: kriszyp

resources/analytics/read.ts

@@ -8,11 +8,13 @@ import { CONFIG_PARAMS } from '../../utility/hdbTerms.ts';
 import { get as envGet } from '../../utility/environment/environmentManager.ts';
 import { validateGetAnalytics } from '../../validation/analyticsValidator.ts';
 import { handleHDBError, hdbErrors } from '../../utility/errors/hdbError.ts';
+import { getThisNodeName } from '../../server/nodeName.ts';
 
 // default to one week time window for finding custom metrics
 const defaultCustomMetricWindow = 1000 * 60 * 60 * 24 * 7;
 
-const log = forComponent('analytics').conditional;
+const logger = forComponent('analytics');
+const log = logger.conditional;
 
 async function lookupHostname(nodeId: number): Promise<string | undefined> {
 	const result = await getAnalyticsHostnameTable().get(nodeId);
@@ -26,15 +28,19 @@ function isSelected(querySelect: string[], attr: string) {
 }
 
 interface GetAnalyticsRequest {
+	operation?: string;
 	metric: string;
 	start_time?: number;
 	end_time?: number;
 	get_attributes?: string[];
 	coalesce_time?: boolean;
 	conditions?: Conditions;
+	// When true, fan the query out to every peer node and merge the results into one
+	// cluster-wide response. Cleared before forwarding so peers only return their own.
+	replicated?: boolean;
 }
 
-type GetAnalyticsResponse = Metric[];
+type GetAnalyticsResponse = AsyncIterable<Metric> | Metric[];
 
 /**
  * Validates the `get_analytics` request and returns the analytics results.
@@ -54,13 +60,75 @@ export async function getOp(req: GetAnalyticsRequest): Promise<GetAnalyticsRespo
 			true
 		);
 	}
-	return get(req.metric, {
-		getAttributes: req.get_attributes,
+	// `replicated` fans the query out to every peer node and merges each node's
+	// analytics into one cluster-wide result set. Fan-out is skipped when:
+	//  - the request did not ask for it;
+	//  - this is standalone core, which has no `server.nodes` (harper-pro populates it);
+	//  - `hdb_analytics` already replicates across the cluster, in which case a local
+	//    query already holds every node's rows and fanning out would double-count.
+	//    The DB layer marks the table `replicate === false` only when it is *not*
+	//    replicated (`analytics_replicate: false`), which is exactly when fan-out helps.
+	const analyticsReplicatedByDb = databases.system.hdb_analytics.replicate !== false;
+	const peers = req.replicated && !analyticsReplicatedByDb && server.nodes?.length ? server.nodes : undefined;
+
+	// When merging across the cluster, make sure every row keeps its origin `node`
+	// attribute so callers can tell the nodes apart. An empty/absent `get_attributes`
+	// already selects everything (including `node`), so only an explicit list needs it.
+	let getAttributes = req.get_attributes;
+	if (peers && getAttributes?.length && !getAttributes.includes('node')) {
+		getAttributes = [...getAttributes, 'node'];
+	}
+
+	const localResults = await get(req.metric, {
+		getAttributes,
 		startTime: req.start_time,
 		endTime: req.end_time,
 		coalesceTime: req.coalesce_time,
 		additionalConditions: req.conditions,
 	});
+
+	if (!peers) return localResults;
+	return mergeAnalyticsFromPeers(localResults, { ...req, get_attributes: getAttributes }, peers);
+}
+
+/**
+ * Streams the local analytics, then appends each peer node's analytics. The same
+ * query is forwarded to every peer with `replicated` cleared so each returns only
+ * its own local metrics (no recursive fan-out). Best-effort: a peer that fails is
+ * logged and omitted rather than failing the whole query.
+ */
+async function* mergeAnalyticsFromPeers(
+	localResults: AsyncIterable<Metric> | Iterable<Metric>,
+	req: GetAnalyticsRequest,
+	peers: { name: string }[]
+): AsyncGenerator<Metric> {
+	const thisNode = getThisNodeName();
+	const peerReq = { ...req, replicated: false };
+	// `sendOperationToNode` is typed for a node name string, but the replication
+	// implementation expects the full node object (the stub-vs-impl mismatch that
+	// `restart` works around the same way).
+	const sendOperationToNode = server.replication.sendOperationToNode as unknown as (
+		node: { name: string },
+		operation: unknown
+	) => Promise<{ results?: Metric[] } | Metric[]>;
+
+	const peerResults = peers
+		.filter((node) => node.name !== thisNode)
+		.map((node) =>
+			sendOperationToNode(node, peerReq).then(
+				// an array response is wrapped as `{ results }` over the replication channel
+				(response): Metric[] => (Array.isArray(response) ? response : (response?.results ?? [])),
+				(error: Error): Metric[] => {
+					logger.warn(`get_analytics replication to node '${node.name}' failed; omitting its results`, error);
+					return [];
+				}
+			)
+		);
+
+	yield* localResults;
+	for (const peerResult of peerResults) {
+		yield* await peerResult;
+	}
 }
 
 function conformCondition(condition: Condition): Condition {

unitTests/resources/analytics/read.test.js

@@ -4,7 +4,32 @@ const { expect } = require('chai');
 const { describe, it } = require('mocha');
 const sinon = require('sinon');
 const { METRIC } = require('#src/resources/analytics/metadata');
-const { listMetrics, describeMetric /* collectDistinctValues */ } = require('#src/resources/analytics/read');
+const { getOp, listMetrics, describeMetric /* collectDistinctValues */ } = require('#src/resources/analytics/read');
+const { getThisNodeName } = require('#src/server/nodeName');
+const hostnames = require('#src/resources/analytics/hostnames');
+
+// Mimics the Harper search iterable: array-like with a lazy async `.map`, which is
+// what resources/analytics/read.ts `get()` consumes.
+function mockSearchIterable(items) {
+	return {
+		[Symbol.asyncIterator]: async function* () {
+			for (const item of items) yield item;
+		},
+		map(fn) {
+			return {
+				[Symbol.asyncIterator]: async function* () {
+					for (const item of items) yield await fn(item);
+				},
+			};
+		},
+	};
+}
+
+async function collect(result) {
+	const out = [];
+	for await (const item of result) out.push(item);
+	return out;
+}
 
 describe('listMetrics', () => {
 	let searchStub;
@@ -301,3 +326,155 @@ describe('describeMetric', () => {
 		}
 	});
 });
+
+describe('getOp (replicated fan-out)', () => {
+	let searchStub;
+	let sendOperationStub;
+	let originalServer;
+	let originalDatabases;
+
+	beforeEach(() => {
+		// `server` and `databases` are process-wide globals established at module load;
+		// stash and restore them rather than deleting so later test files still see them.
+		originalServer = global.server;
+		originalDatabases = global.databases;
+
+		searchStub = sinon.stub().returns(mockSearchIterable([]));
+		// `replicate === false` => analytics are NOT replicated by the DB layer, so the
+		// fan-out is needed (and enabled). The skip case is covered explicitly below.
+		global.databases = { system: { hdb_analytics: { search: searchStub, replicate: false } } };
+
+		sendOperationStub = sinon.stub();
+		global.server = {
+			hostname: 'local-host',
+			nodes: [],
+			replication: { sendOperationToNode: sendOperationStub },
+		};
+	});
+
+	afterEach(() => {
+		sinon.restore();
+		global.server = originalServer;
+		global.databases = originalDatabases;
+	});
+
+	it('merges metrics from every peer node into one flat result set', async () => {
+		global.server.nodes = [{ name: 'peer-a' }, { name: 'peer-b' }];
+		sendOperationStub
+			.withArgs(sinon.match({ name: 'peer-a' }))
+			.resolves({ results: [{ id: 1, metric: 'm', node: 'peer-a' }] });
+		sendOperationStub
+			.withArgs(sinon.match({ name: 'peer-b' }))
+			.resolves({ results: [{ id: 2, metric: 'm', node: 'peer-b' }] });
+
+		const result = await collect(await getOp({ operation: 'get_analytics', metric: 'm', replicated: true }));
+
+		expect(result).to.deep.equal([
+			{ id: 1, metric: 'm', node: 'peer-a' },
+			{ id: 2, metric: 'm', node: 'peer-b' },
+		]);
+		expect(sendOperationStub.calledTwice).to.be.true;
+	});
+
+	it('forwards the query to peers with `replicated` cleared (no recursive fan-out)', async () => {
+		global.server.nodes = [{ name: 'peer-a' }];
+		sendOperationStub.resolves({ results: [] });
+
+		await collect(await getOp({ operation: 'get_analytics', metric: 'm', replicated: true }));
+
+		const forwarded = sendOperationStub.firstCall.args[1];
+		expect(forwarded.replicated).to.equal(false);
+		expect(forwarded.metric).to.equal('m');
+	});
+
+	it('skips the local node when fanning out', async () => {
+		const thisNode = getThisNodeName();
+		global.server.nodes = [{ name: thisNode }, { name: 'peer-x' }];
+		sendOperationStub.resolves({ results: [] });
+
+		await collect(await getOp({ metric: 'm', replicated: true }));
+
+		expect(sendOperationStub.calledOnce).to.be.true;
+		expect(sendOperationStub.firstCall.args[0]).to.deep.equal({ name: 'peer-x' });
+	});
+
+	it('omits a peer that errors and still returns the others (best-effort)', async () => {
+		global.server.nodes = [{ name: 'peer-good' }, { name: 'peer-bad' }];
+		sendOperationStub
+			.withArgs(sinon.match({ name: 'peer-good' }))
+			.resolves({ results: [{ id: 1, metric: 'm', node: 'peer-good' }] });
+		sendOperationStub.withArgs(sinon.match({ name: 'peer-bad' })).rejects(new Error('connection refused'));
+
+		const result = await collect(await getOp({ metric: 'm', replicated: true }));
+
+		expect(result).to.deep.equal([{ id: 1, metric: 'm', node: 'peer-good' }]);
+	});
+
+	it('accepts a bare-array peer response (defensive unwrap)', async () => {
+		global.server.nodes = [{ name: 'peer-a' }];
+		sendOperationStub.resolves([{ id: 5, metric: 'm', node: 'peer-a' }]);
+
+		const result = await collect(await getOp({ metric: 'm', replicated: true }));
+
+		expect(result).to.deep.equal([{ id: 5, metric: 'm', node: 'peer-a' }]);
+	});
+
+	it('includes local node results ahead of peer results', async () => {
+		sinon
+			.stub(hostnames, 'getAnalyticsHostnameTable')
+			.returns({ get: sinon.stub().resolves({ hostname: 'local-host' }) });
+		searchStub.returns(mockSearchIterable([{ id: [10, 12345], metric: 'm', total: 1 }]));
+		global.server.nodes = [{ name: 'peer-a' }];
+		sendOperationStub.resolves({ results: [{ id: 20, metric: 'm', node: 'peer-a', total: 2 }] });
+
+		const result = await collect(await getOp({ metric: 'm', replicated: true }));
+
+		expect(result).to.deep.equal([
+			{ id: 10, metric: 'm', total: 1, node: 'local-host' },
+			{ id: 20, metric: 'm', node: 'peer-a', total: 2 },
+		]);
+	});
+
+	it('forces the `node` attribute into an explicit get_attributes list when replicated', async () => {
+		global.server.nodes = [{ name: 'peer-a' }];
+		sendOperationStub.resolves({ results: [] });
+
+		await collect(await getOp({ metric: 'm', get_attributes: ['metric', 'total'], replicated: true }));
+
+		const forwarded = sendOperationStub.firstCall.args[1];
+		expect(forwarded.get_attributes).to.include('node');
+	});
+
+	it('does not fan out when `replicated` is not set', async () => {
+		searchStub.returns(mockSearchIterable([{ id: [10, 1], metric: 'm', total: 1 }]));
+		global.server.nodes = [{ name: 'peer-a' }];
+
+		const result = await collect(await getOp({ metric: 'm', get_attributes: ['metric', 'total'] }));
+
+		expect(sendOperationStub.called).to.be.false;
+		expect(result).to.deep.equal([{ id: 10, metric: 'm', total: 1 }]);
+	});
+
+	it('does not fan out in standalone core (no server.nodes)', async () => {
+		global.server.nodes = undefined;
+		searchStub.returns(mockSearchIterable([{ id: [10, 1], metric: 'm' }]));
+
+		const result = await collect(await getOp({ metric: 'm', get_attributes: ['metric'], replicated: true }));
+
+		expect(sendOperationStub.called).to.be.false;
+		expect(result).to.deep.equal([{ id: 10, metric: 'm' }]);
+	});
+
+	it('does not fan out when the analytics table already replicates (replicate !== false)', async () => {
+		// `analytics_replicate: true` leaves the table `replicate` undefined; a local query
+		// already holds every node's rows, so fanning out would double-count.
+		delete global.databases.system.hdb_analytics.replicate;
+		global.server.nodes = [{ name: 'peer-a' }];
+		searchStub.returns(mockSearchIterable([{ id: [10, 1], metric: 'm', total: 1 }]));
+
+		const result = await collect(await getOp({ metric: 'm', get_attributes: ['metric', 'total'], replicated: true }));
+
+		expect(sendOperationStub.called).to.be.false;
+		expect(result).to.deep.equal([{ id: 10, metric: 'm', total: 1 }]);
+	});
+});

unitTests/validation/analyticsValidator.test.js

@@ -175,4 +175,16 @@ describe('validateGetAnalytics', function () {
 		const error = validateGetAnalytics({ metric: 'cpu-usage', conditions: [{ attribute: 'path' }] });
 		assert.ok(error instanceof Error);
 	});
+
+	// ── replicated ───────────────────────────────────────────────────────────
+
+	it('should accept replicated as a boolean', function () {
+		assert.strictEqual(validateGetAnalytics({ metric: 'cpu-usage', replicated: true }), undefined);
+	});
+
+	it('should reject replicated as the string "true"', function () {
+		const error = validateGetAnalytics({ metric: 'cpu-usage', replicated: 'true' });
+		assert.ok(error instanceof Error);
+		assert.ok(error.message.includes('replicated'), `expected "replicated" in: ${error.message}`);
+	});
 });

validation/analyticsValidator.ts

@@ -36,6 +36,7 @@ const getAnalyticsSchema = Joi.object({
 	get_attributes: Joi.array().items(Joi.string()),
 	coalesce_time: Joi.boolean(),
 	conditions: Joi.array().items(Joi.alternatives(groupConditionSchema, directConditionSchema)),
+	replicated: Joi.boolean(),
 }).strict();
 
 export function validateGetAnalytics(req: any): Error | undefined {