chore(spanner): support built-in metrics for ResultSet streaming RPCs - #6302
chore(spanner): support built-in metrics for ResultSet streaming RPCs#6302olavloite wants to merge 2 commits into
Conversation
Adds OpenTelemetry built-in metrics instrumentation for streaming operations (`ExecuteStreamingSql` and `StreamingRead`) in Cloud Spanner `ResultSet`. - Records `attempt_count`, `attempt_latencies`, `gfe_latencies`, and `afe_latencies` across all initial attempts and stream restarts. - Records end-to-end `operation_count` and `operation_latencies` spanning the entire query lifecycle until stream completion or permanent failure. - Preserves attempt response headers across background stream trailer drains (`seen_last`) and early drops. - Standardizes metric `method` attribute formatting to `"Spanner.<MethodName>"` (`Spanner.ExecuteStreamingSql`, `Spanner.StreamingRead`) matching Java and Go reference clients. - Parses single and multiple `server-timing` response headers for GFE and AFE timings.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces attempt and operation metrics tracking for Spanner streaming queries and reads, including parsing server-timing headers for GFE/AFE latencies and normalising method names. The review feedback highlights several critical gaps in error handling and metrics accuracy: stream initialization failures are currently not recorded (and incorrectly reported as successful upon drop), stale headers from previous attempts are not cleared during retries, and complete connection failures prior to ResultSet creation are missed entirely by the metrics recorder.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #6302 +/- ##
==========================================
+ Coverage 96.26% 96.29% +0.03%
==========================================
Files 282 282
Lines 72762 73529 +767
==========================================
+ Hits 70044 70806 +762
- Misses 2718 2723 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
7f24278 to
3e80e52
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request implements built-in metrics collection for Spanner streaming operations, tracking attempt and operation counts, latencies, and GFE/AFE latencies parsed from response headers. It updates ResultSet to record these metrics during stream initialization, execution, retries, and upon being dropped, and adds comprehensive unit and integration tests to verify the metrics recording behavior. The review feedback suggests extending the stream auto-trait assertions to explicitly verify the Sync trait, ensuring compatibility with thread-safe types.
3e80e52 to
499440c
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request implements attempt and operation metrics tracking for Spanner streaming RPCs (ExecuteStreamingSql and StreamingRead), including parsing GFE/AFE latencies from response headers and recording metrics on completion or failure. While the implementation is well-tested, the review feedback highlights a critical blind spot: initial stream creation failures (when the initial send().await fails before a ResultSet is instantiated) are currently not recorded in the metrics. To ensure complete observability, these initial failure paths in both batch_read_only_transaction.rs and the retry macro in read_only_transaction.rs should be updated to record attempt and operation metrics.
| /// Helper macro to execute a streaming SQL or streaming read RPC with retry logic. | ||
| macro_rules! execute_stream_with_retry { | ||
| ($self:expr, $request:ident, $gax_options:ident, $rpc_method:ident, $operation_variant:path, $method_name:expr) => {{ | ||
| let attempt_start_time = Instant::now(); |
There was a problem hiding this comment.
Missing Metrics for Initial Stream Creation Failures
If the initial send().await call fails, the macro returns an error immediately (or after a failed explicit begin fallback) without ever creating a ResultSet. Consequently, no attempt or operation metrics are recorded for these initial failures. This creates a significant blind spot in monitoring, as initial connection/handshake/validation failures will not be reflected in the metrics.
Additionally, if the first attempt fails but the retry succeeds, the first failed attempt is not recorded, and the second attempt's start time is incorrectly measured from the beginning of the first attempt.
To fix this, we should:
- Record attempt and operation metrics on all initial failure paths.
- Record the first failed attempt and reset
attempt_start_timebefore retrying.
Since the middle of the macro is not part of the diff hunks, a direct code suggestion cannot be applied here. However, you can refactor the macro as follows:
macro_rules! execute_stream_with_retry {
($self:expr, $request:ident, $gax_options:ident, $rpc_method:ident, $operation_variant:path, $method_name:expr) => {{
let mut attempt_start_time = Instant::now();
let stream = match $self
.client
.spanner
.$rpc_method($request.clone(), $gax_options.clone(), $self.channel_hint)
.send()
.await
{
Ok(s) => s,
Err(e) => {
let elapsed = attempt_start_time.elapsed();
let is_starting = matches!(
$request
.transaction
.as_ref()
.and_then(|t| t.selector.as_ref()),
Some(crate::model::transaction_selector::Selector::Begin(_))
);
if is_starting {
if $self.transaction_selector.is_read_write() {
$self.client.o11y.record_attempt_with_status($method_name, elapsed, Some(&e), None, None);
$self.client.o11y.record_operation_with_status($method_name, elapsed, Some(&e));
$self.transaction_selector.set_failed(&e);
return Err(e);
} else {
if is_aborted(&e) {
$self.client.o11y.record_attempt_with_status($method_name, elapsed, Some(&e), None, None);
$self.client.o11y.record_operation_with_status($method_name, elapsed, Some(&e));
return Err(e);
}
$self.client.o11y.record_attempt_with_status($method_name, elapsed, Some(&e), None, None);
if $self
.begin_explicitly_if_not_started($gax_options.clone(), true, None)
.await?
{
$request.transaction =
Some($self.transaction_selector.selector().await?);
attempt_start_time = Instant::now();
match $self
.client
.spanner
.$rpc_method(
$request.clone(),
$gax_options.clone(),
$self.channel_hint,
)
.send()
.await
{
Ok(s) => s,
Err(retry_err) => {
let retry_elapsed = attempt_start_time.elapsed();
$self.client.o11y.record_attempt_with_status($method_name, retry_elapsed, Some(&retry_err), None, None);
$self.client.o11y.record_operation_with_status($method_name, retry_elapsed, Some(&retry_err));
return Err(retry_err);
}
}
} else {
$self.client.o11y.record_operation_with_status($method_name, elapsed, Some(&e));
return Err(e);
}
}
} else {
$self.client.o11y.record_attempt_with_status($method_name, elapsed, Some(&e), None, None);
$self.client.o11y.record_operation_with_status($method_name, elapsed, Some(&e));
return Err(e);
}
}
};
// ... rest of the macro ...References
- Scrutinize Edge Cases: Always consider failure modes (network failures, unexpected inputs). Do not assume a perfect world. (link)
| ) -> crate::Result<ResultSet> { | ||
| let channel_hint = client.spanner.next_channel_hint(); | ||
| let gax_options = client.spanner.attach_request_id(gax_options, channel_hint); | ||
| let attempt_start_time = Instant::now(); |
There was a problem hiding this comment.
Missing Metrics for Initial Stream Creation Failures
If the initial send().await call fails, the error is returned immediately via ? without ever creating a ResultSet. Consequently, no attempt or operation metrics are recorded for these initial failures. This creates a significant blind spot in monitoring, as initial connection/handshake/validation failures will not be reflected in the metrics.
We should match on the result of send().await and record the metrics on error:
let attempt_start_time = Instant::now();
let stream = match client
.spanner
.execute_streaming_sql(req.clone(), gax_options.clone(), channel_hint)
.send()
.await
{
Ok(s) => s,
Err(e) => {
let elapsed = attempt_start_time.elapsed();
client.o11y.record_attempt_with_status("ExecuteStreamingSql", elapsed, Some(&e), None, None);
client.o11y.record_operation_with_status("ExecuteStreamingSql", elapsed, Some(&e));
return Err(e);
}
};References
- Scrutinize Edge Cases: Always consider failure modes (network failures, unexpected inputs). Do not assume a perfect world. (link)
| ) -> crate::Result<ResultSet> { | ||
| let channel_hint = client.spanner.next_channel_hint(); | ||
| let gax_options = client.spanner.attach_request_id(gax_options, channel_hint); | ||
| let attempt_start_time = Instant::now(); |
There was a problem hiding this comment.
Missing Metrics for Initial Stream Creation Failures
If the initial send().await call fails, the error is returned immediately via ? without ever creating a ResultSet. Consequently, no attempt or operation metrics are recorded for these initial failures. This creates a significant blind spot in monitoring, as initial connection/handshake/validation failures will not be reflected in the metrics.
We should match on the result of send().await and record the metrics on error:
let attempt_start_time = Instant::now();
let stream = match client
.spanner
.streaming_read(req.clone(), gax_options.clone(), channel_hint)
.send()
.await
{
Ok(s) => s,
Err(e) => {
let elapsed = attempt_start_time.elapsed();
client.o11y.record_attempt_with_status("StreamingRead", elapsed, Some(&e), None, None);
client.o11y.record_operation_with_status("StreamingRead", elapsed, Some(&e));
return Err(e);
}
};References
- Scrutinize Edge Cases: Always consider failure modes (network failures, unexpected inputs). Do not assume a perfect world. (link)
1 participant
Adds OpenTelemetry built-in metrics instrumentation for streaming operations (
ExecuteStreamingSqlandStreamingRead) in Cloud SpannerResultSet.attempt_count,attempt_latencies,gfe_latencies, andafe_latenciesacross all initial attempts and stream restarts.operation_countandoperation_latenciesspanning the entire query lifecycle until stream completion or permanent failure.seen_last) and early drops.methodattribute formatting to"Spanner.<MethodName>"(Spanner.ExecuteStreamingSql,Spanner.StreamingRead) matching Java and Go reference clients.server-timingresponse headers for GFE and AFE timings.