feat: instrument read_window_aggregate
@@ -356,13 +356,23 @@ where
info!(%db_name, ?range, ?window_every, ?offset, ?aggregate, ?window, predicate=%predicate.loggable(),"read_window_aggregate");
+ let ob = self.metrics.requests.observation();
+ let labels = &[
+ KeyValue::new("operation", "read_window_aggregate"),
+ KeyValue::new("db_name", db_name.to_string()),
+ ];
+
let aggregate_string = format!(
"aggregate: {:?}, window_every: {:?}, offset: {:?}, window: {:?}",
aggregate, window_every, offset, window
);
let gby_agg = expr::make_read_window_aggregate(aggregate, window_every, offset, window)
- .context(ConvertingWindowAggregate { aggregate_string })?;
+ .context(ConvertingWindowAggregate { aggregate_string })
+ .map_err(|e| {
+ ob.client_error_with_labels(labels);
+ e
+ })?;
let results = query_group_impl(
Arc::clone(&self.db_store),
@@ -372,11 +382,19 @@ where
gby_agg,
)
.await
- .map_err(|e| e.to_status())?
+ .map_err(|e| {
+ if e.is_internal() {
+ ob.error_with_labels(labels);
+ } else {
+ ob.client_error_with_labels(labels);
+ }
+ e.to_status()
+ })?
.into_iter()
.map(Ok)
.collect::<Vec<_>>();
+ ob.ok_with_labels(labels);
Ok(tonic::Response::new(futures::stream::iter(results)))
}
@@ -2034,7 +2052,7 @@ mod tests {
}
#[tokio::test]
- async fn test_read_window_aggegate_window_every() {
+ async fn test_read_window_aggregate_window_every() {
test_helpers::maybe_start_logging();
// Start a test gRPC server on a randomally allocated port
let mut fixture = Fixture::new().await.expect("Connecting to test server");
@@ -2089,6 +2107,19 @@ mod tests {
actual_frames, expected_frames,
"unexpected frames returned by query_groups"
);
+
+ fixture
+ .test_storage
+ .metrics_registry
+ .has_metric_family("gRPC_requests_total")
+ .with_labels(&[
+ ("operation", "read_window_aggregate"),
+ ("db_name", "000000000000007b_00000000000001c8"),
+ ("status", "ok"),
+ ])
+ .counter()
+ .eq(1.0)
+ .unwrap();
}
#[tokio::test]
@@ -2209,6 +2240,19 @@ mod tests {
.to_string();
assert_contains!(response_string, "Sugar we are going down");
+
+ fixture
+ .test_storage
+ .metrics_registry
+ .has_metric_family("gRPC_requests_total")
+ .with_labels(&[
+ ("operation", "read_window_aggregate"),
+ ("db_name", "000000000000007b_00000000000001c8"),
+ ("status", "client_error"),
+ ])
+ .counter()
+ .eq(1.0)
+ .unwrap();
}
#[tokio::test]
