chore(spanner): [WIP] implement built-in attempt metrics and server-timing recording - #6311
chore(spanner): [WIP] implement built-in attempt metrics and server-timing recording#6311olavloite wants to merge 1 commit into
Conversation
![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 extends the AttemptInterceptor trait with lifecycle hooks (on_attempt_start and on_attempt_complete) to track unary RPC attempts, and integrates these hooks into the Spanner client to collect experimental builtin metrics (such as GFE/AFE latencies and connectivity errors). Feedback on the changes highlights an opportunity to avoid unnecessary double indirection (Arc<Vec<...>>) when metrics are disabled, as well as a bug in the method name normalization fallback logic when handling custom methods prefixed with Spanner/.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #6311 +/- ##
==========================================
- Coverage 96.26% 96.15% -0.12%
==========================================
Files 282 282
Lines 72762 73375 +613
==========================================
+ Hits 70044 70552 +508
- Misses 2718 2823 +105 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
b650103 to
f89d3b1
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces lifecycle hooks (on_attempt_start and on_attempt_complete) to the AttemptInterceptor trait in the GAX library, allowing unary RPC attempts to be tracked. These hooks are integrated into the gRPC client and utilized in the Spanner client to collect detailed attempt-level metrics, such as GFE/AFE latencies and connectivity errors, when the _experimental-builtin-metrics feature is enabled. Feedback on the changes suggests simplifying the fallback pattern matching in normalize_method_name by utilizing the already-cleaned method name prefix instead of the raw trimmed string.
…recording - Implement SpannerMetricsInterceptor to record attempt latencies, attempt count, and server-timing metrics (GFE and AFE latencies) on RPC attempt completion. - Inject x-goog-spanner-enable-afe-server-timing request header on outgoing attempts, respecting SPANNER_DISABLE_AFE_SERVER_TIMING. - Normalize the method metric attribute to "Spanner.<RPC>" with zero-allocation static matching across operation and attempt metrics. - Fix GFE and AFE connectivity error recording to prevent false positive AFE errors on non-DirectPath traffic. - Disable Cloud Monitoring exporter initialization for plaintext mock/test endpoints, Omni, and emulator environments.
f89d3b1 to
3033b3e
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces lifecycle hooks (on_attempt_start and on_attempt_complete) to the AttemptInterceptor trait to track unary RPC attempts, and integrates them into the gRPC client and Spanner metrics interceptor to record attempt-level metrics. Feedback focuses on optimizing the gRPC client's on_attempt_complete method to take and return the result by value, enabling zero-allocation header extraction by destructuring the Response instead of cloning its metadata. Additionally, it is recommended to document a limitation in the composite Vec interceptor where custom Instant values returned by individual interceptors are discarded.
| pending.await | ||
| }; | ||
|
|
||
| self.on_attempt_complete(path.path(), attempt_number, start_time, &result, options); |
There was a problem hiding this comment.
Pass result by value to on_attempt_complete to allow zero-allocation header extraction by moving the metadata instead of cloning it.
| self.on_attempt_complete(path.path(), attempt_number, start_time, &result, options); | |
| let result = self.on_attempt_complete(path.path(), attempt_number, start_time, result, options); |
| fn on_attempt_complete<Response>( | ||
| &self, | ||
| method: &str, | ||
| attempt: u32, | ||
| start_time: Option<Instant>, | ||
| result: &Result<::tonic::Response<Response>>, | ||
| options: &RequestOptions, | ||
| ) { | ||
| if let (Some(interceptor), Some(start_time)) = (&self.attempt_interceptor, start_time) { | ||
| let res_headers; | ||
| let (response_headers, error) = match result { | ||
| Ok(response) => { | ||
| res_headers = response.metadata().clone().into_headers(); | ||
| (Some(&res_headers), None) | ||
| } | ||
| Err(error) => (error.http_headers(), Some(error)), | ||
| }; | ||
| interceptor.on_attempt_complete( | ||
| method, | ||
| attempt, | ||
| start_time, | ||
| response_headers, | ||
| error, | ||
| options, | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this comment.
Update on_attempt_complete to take and return result by value. By destructuring the successful Response using into_parts() and extracting the headers via into_headers(), we can pass the headers to the interceptor and reconstruct the Response without any cloning or extra allocations.
fn on_attempt_complete<Response>(
&self,
method: &str,
attempt: u32,
start_time: Option<Instant>,
result: Result<::tonic::Response<Response>>,
options: &RequestOptions,
) -> Result<::tonic::Response<Response>> {
if let (Some(interceptor), Some(start_time)) = (&self.attempt_interceptor, start_time) {
match result {
Ok(response) => {
let (metadata, message, extensions) = response.into_parts();
let headers = metadata.into_headers();
interceptor.on_attempt_complete(
method,
attempt,
start_time,
Some(&headers),
None,
options,
);
let metadata = ::tonic::metadata::MetadataMap::from_headers(headers);
let mut response = ::tonic::Response::from_parts(metadata, message);
*response.extensions_mut() = extensions;
Ok(response)
}
Err(error) => {
interceptor.on_attempt_complete(
method,
attempt,
start_time,
error.http_headers(),
Some(&error),
options,
);
Err(error)
}
}
} else {
result
}
}| fn on_attempt_start( | ||
| &self, | ||
| method: &str, | ||
| attempt: u32, | ||
| headers: &mut HeaderMap, | ||
| options: &RequestOptions, | ||
| ) -> Instant { | ||
| for interceptor in self { | ||
| interceptor.intercept(headers, attempt); | ||
| interceptor.on_attempt_start(method, attempt, headers, options); | ||
| } | ||
| Instant::now() | ||
| } |
There was a problem hiding this comment.
In the Vec<Arc<dyn AttemptInterceptor>> implementation of on_attempt_start, the Instant returned by each individual interceptor's on_attempt_start is discarded, and a generic Instant::now() is returned instead. If an individual interceptor overrides on_attempt_start to return a custom or mocked Instant, this custom timing will be ignored during on_attempt_complete. Consider documenting this limitation in the trait or the Vec implementation's docstring to prevent unexpected behavior for custom interceptors.
/// Callback invoked before a unary RPC attempt is dispatched.
///
/// # Note
/// Any custom `Instant` returned by individual interceptors in the `Vec` is discarded.
/// The composite implementation always returns a fresh `Instant::now()`.
fn on_attempt_start(
&self,
method: &str,
attempt: u32,
headers: &mut HeaderMap,
options: &RequestOptions,
) -> Instant {
for interceptor in self {
interceptor.on_attempt_start(method, attempt, headers, options);
}
Instant::now()
}1 participant
Do not review: Work in progress, the Gax changes will be split into a separate pull request.