Skip to content
This repository was archived by the owner on Mar 26, 2026. It is now read-only.

Commit 8af801b

Browse files
danieljbrucedanieljbruce
authored
feat: Record ReadRows application latencies for client side metrics (#1647)
## Description This PR makes it so that all readRows calls will start recording application latencies. With this change, data for application latencies will now be available on the Google Cloud dashboard. ## Impact Data should now be available for application latencies on ReadRows calls in the client side metrics dashboard. ## Testing Tests are added that mock out the hrtime module and make sure the right application latencies values are recorded.
1 parent 7601e4d commit 8af801b

17 files changed

Lines changed: 631 additions & 147 deletions

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,8 @@
5757
"@opentelemetry/resources": "^1.30.0",
5858
"@opentelemetry/sdk-metrics": "^1.30.0",
5959
"@types/long": "^4.0.0",
60-
"arrify": "2.0.0",
6160
"abort-controller": "^3.0.0",
61+
"arrify": "2.0.0",
6262
"concat-stream": "^2.0.0",
6363
"dot-prop": "6.0.0",
6464
"escape-string-regexp": "4.0.0",

‎src/client-side-metrics/gcp-metrics-handler.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,9 +255,9 @@ export class GCPMetricsHandler implements IMetricsHandler {
255255
status: data.status,
256256
...commonAttributes,
257257
});
258-
for (const applicationLatency of data.applicationLatencies) {
258+
if (data.applicationLatency) {
259259
otelInstruments.applicationBlockingLatencies.record(
260-
applicationLatency,
260+
data.applicationLatency,
261261
commonAttributes,
262262
);
263263
}

‎src/client-side-metrics/metrics-handler.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ interface StandardData {
3939
export interface OnOperationCompleteData extends StandardData {
4040
firstResponseLatency?: number;
4141
operationLatency: number;
42-
applicationLatencies: number[];
42+
applicationLatency?: number;
4343
retryCount?: number;
4444
}
4545

‎src/client-side-metrics/operation-metrics-collector.ts‎

Lines changed: 8 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import * as gax from 'google-gax';
1919
import {AbortableDuplex, BigtableOptions} from '../index';
2020
import * as path from 'path';
2121
import {IMetricsHandler} from './metrics-handler';
22+
import {TimedStream} from '../timed-stream';
2223

2324
// When this environment variable is set then print any errors associated
2425
// with failures in the metrics collector.
@@ -115,9 +116,8 @@ export class OperationMetricsCollector {
115116
private serverTime: number | null;
116117
private connectivityErrorCount: number;
117118
private streamingOperation: StreamingState;
118-
private applicationLatencies: number[];
119-
private lastRowReceivedTime: bigint | null;
120119
private handlers: IMetricsHandler[];
120+
public userStream?: TimedStream;
121121

122122
/**
123123
* @param {ITabularApiSurface} tabularApiSurface Information about the Bigtable table being accessed.
@@ -143,8 +143,6 @@ export class OperationMetricsCollector {
143143
this.serverTime = null;
144144
this.connectivityErrorCount = 0;
145145
this.streamingOperation = streamingOperation;
146-
this.lastRowReceivedTime = null;
147-
this.applicationLatencies = [];
148146
this.handlers = handlers;
149147
}
150148

@@ -197,7 +195,6 @@ export class OperationMetricsCollector {
197195
checkState(this.state, [MetricsCollectorState.OPERATION_NOT_STARTED]);
198196
this.operationStartTime = hrtime.bigint();
199197
this.firstResponseLatency = null;
200-
this.applicationLatencies = [];
201198
this.state =
202199
MetricsCollectorState.OPERATION_STARTED_ATTEMPT_NOT_IN_PROGRESS;
203200
});
@@ -254,7 +251,6 @@ export class OperationMetricsCollector {
254251
this.serverTime = null;
255252
this.serverTimeRead = false;
256253
this.connectivityErrorCount = 0;
257-
this.lastRowReceivedTime = null;
258254
});
259255
}
260256

@@ -287,8 +283,12 @@ export class OperationMetricsCollector {
287283
* Called when an operation completes (successfully or unsuccessfully).
288284
* Records operation latencies, retry counts, and connectivity error counts.
289285
* @param {grpc.status} finalOperationStatus Information about the completed operation.
286+
* @param {number} applicationLatency The application latency measurement.
290287
*/
291-
onOperationComplete(finalOperationStatus: grpc.status) {
288+
onOperationComplete(
289+
finalOperationStatus: grpc.status,
290+
applicationLatency?: number,
291+
) {
292292
this.onAttemptComplete(finalOperationStatus);
293293
withMetricsDebug(() => {
294294
checkState(this.state, [
@@ -311,7 +311,7 @@ export class OperationMetricsCollector {
311311
operationLatency: totalMilliseconds,
312312
retryCount: this.attemptCount - 1,
313313
firstResponseLatency: this.firstResponseLatency ?? undefined,
314-
applicationLatencies: this.applicationLatencies,
314+
applicationLatency: applicationLatency ?? 0,
315315
});
316316
}
317317
});
@@ -355,38 +355,6 @@ export class OperationMetricsCollector {
355355
}
356356
}
357357

358-
/**
359-
* Called when a row from the Bigtable stream reaches the application user.
360-
*
361-
* This method is used to calculate the latency experienced by the application
362-
* when reading rows from a Bigtable stream. It records the time between the
363-
* previous row being received and the current row reaching the user. These
364-
* latencies are then collected and reported as `applicationBlockingLatencies`
365-
* when the operation completes.
366-
*/
367-
onRowReachesUser() {
368-
if (
369-
this.state ===
370-
MetricsCollectorState.OPERATION_STARTED_ATTEMPT_IN_PROGRESS_NO_ROWS_YET ||
371-
this.state ===
372-
MetricsCollectorState.OPERATION_STARTED_ATTEMPT_IN_PROGRESS_SOME_ROWS_RECEIVED ||
373-
this.state ===
374-
MetricsCollectorState.OPERATION_STARTED_ATTEMPT_NOT_IN_PROGRESS
375-
) {
376-
const currentTime = hrtime.bigint();
377-
if (this.lastRowReceivedTime) {
378-
// application latency is measured in total milliseconds.
379-
const applicationLatency = Number(
380-
(currentTime - this.lastRowReceivedTime) / BigInt(1000000),
381-
);
382-
this.applicationLatencies.push(applicationLatency);
383-
}
384-
this.lastRowReceivedTime = currentTime;
385-
} else {
386-
console.warn('Invalid state transition attempted');
387-
}
388-
}
389-
390358
/**
391359
* Called when status information is received. Extracts zone and cluster information.
392360
* @param {object} status The received status information.

‎src/timed-stream.ts‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
// limitations under the License.
1414

1515
import {PassThrough, TransformCallback, TransformOptions} from 'stream';
16+
const {hrtime} = require('node:process');
1617

1718
/**
1819
* This interface is the usual options that can be passed into a Transform plus
@@ -42,11 +43,11 @@ class StreamTimer {
4243
}
4344

4445
start() {
45-
this.startTime = process.hrtime.bigint();
46+
this.startTime = hrtime.bigint();
4647
}
4748

4849
stop() {
49-
const endTime = process.hrtime.bigint();
50+
const endTime = hrtime.bigint();
5051
const duration = endTime - this.startTime;
5152
this.totalDuration += duration;
5253
}
@@ -65,7 +66,8 @@ export class TimedStream extends PassThrough {
6566
super({
6667
...options,
6768
objectMode: true,
68-
highWaterMark: 0,
69+
readableHighWaterMark: 0, // We need to disable readside buffering to allow for acceptable behavior when the end user cancels the stream early.
70+
writableHighWaterMark: 0, // We need to disable writeside buffering because in nodejs 14 the call to _transform happens after write buffering. This creates problems for tracking the last seen row key.
6971
transform: (event, _encoding, callback) => {
7072
/* When we iterate through a stream, time spent waiting for the user's
7173
application is added to totalDurationTransform. When we use handlers,
@@ -79,7 +81,6 @@ export class TimedStream extends PassThrough {
7981
if (options?.transformHook) {
8082
options?.transformHook(event, _encoding, callback);
8183
}
82-
callback(null, event);
8384
this.transformTimer.stop();
8485
},
8586
});

‎src/utils/createReadStreamInternal.ts‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ import {
4242
TabularApiSurface,
4343
} from '../tabular-api-surface';
4444
import {OperationMetricsCollector} from '../client-side-metrics/operation-metrics-collector';
45+
import {TimedStream} from '../timed-stream';
4546

4647
/**
4748
* Creates a readable stream of rows from a Bigtable table or authorized view.
@@ -128,11 +129,8 @@ export function createReadStreamInternal(
128129
// discarded in the per attempt subpipeline (rowStream)
129130
let lastRowKey = '';
130131
let rowsRead = 0;
131-
const userStream = new PassThrough({
132-
objectMode: true,
133-
readableHighWaterMark: 0, // We need to disable readside buffering to allow for acceptable behavior when the end user cancels the stream early.
134-
writableHighWaterMark: 0, // We need to disable writeside buffering because in nodejs 14 the call to _transform happens after write buffering. This creates problems for tracking the last seen row key.
135-
transform(event, _encoding, callback) {
132+
const userStream = new TimedStream({
133+
transformHook(event, _encoding, callback) {
136134
if (userCanceled) {
137135
callback();
138136
return;
@@ -373,7 +371,10 @@ export function createReadStreamInternal(
373371
// We ignore the `cancelled` "error", since we are the ones who cause
374372
// it when the user calls `.abort()`.
375373
userStream.end();
376-
metricsCollector.onOperationComplete(error.code);
374+
metricsCollector.onOperationComplete(
375+
error.code,
376+
userStream.getTotalDurationMs(),
377+
);
377378
return;
378379
}
379380
numConsecutiveErrors++;
@@ -405,7 +406,10 @@ export function createReadStreamInternal(
405406
//
406407
error.code = grpc.status.CANCELLED;
407408
}
408-
metricsCollector.onOperationComplete(error.code);
409+
metricsCollector.onOperationComplete(
410+
error.code,
411+
userStream.getTotalDurationMs(),
412+
);
409413
userStream.emit('error', error);
410414
}
411415
})
@@ -416,7 +420,11 @@ export function createReadStreamInternal(
416420
})
417421
.on('end', () => {
418422
activeRequestStream = null;
419-
metricsCollector.onOperationComplete(grpc.status.OK);
423+
const applicationLatency = userStream.getTotalDurationMs();
424+
metricsCollector.onOperationComplete(
425+
grpc.status.OK,
426+
userStream.getTotalDurationMs(),
427+
);
420428
});
421429
rowStreamPipe(rowStream, userStream);
422430
};

‎system-test/client-side-metrics-all-methods.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -100,10 +100,10 @@ function readRowsAssertionCheck(
100100
// them from the comparison after checking they exist.
101101
assert(secondRequest.operationLatency);
102102
assert(secondRequest.firstResponseLatency);
103-
assert(secondRequest.applicationLatencies);
103+
assert.strictEqual(secondRequest.applicationLatency, 0);
104104
delete secondRequest.operationLatency;
105105
delete secondRequest.firstResponseLatency;
106-
delete secondRequest.applicationLatencies;
106+
delete secondRequest.applicationLatency;
107107
delete secondRequest.metricsCollectorData.appProfileId;
108108
assert.deepStrictEqual(secondRequest, {
109109
status: '0',
@@ -146,10 +146,10 @@ function readRowsAssertionCheck(
146146
// them from the comparison after checking they exist.
147147
assert(fourthRequest.operationLatency);
148148
assert(fourthRequest.firstResponseLatency);
149-
assert(fourthRequest.applicationLatencies);
149+
assert.strictEqual(fourthRequest.applicationLatency, 0);
150150
delete fourthRequest.operationLatency;
151151
delete fourthRequest.firstResponseLatency;
152-
delete fourthRequest.applicationLatencies;
152+
delete fourthRequest.applicationLatency;
153153
delete fourthRequest.metricsCollectorData.appProfileId;
154154
assert.deepStrictEqual(fourthRequest, {
155155
status: '0',

0 commit comments

Comments
 (0)