sql-333: Add application_name to peek duration metrics (#36998)

Adding to provide better observability for catalog server incidents.

Unsure if a test for the metric is necessary 🤷

Author: SangJunBak

src/adapter/src/client.rs

@@ -1583,12 +1583,9 @@ impl RecordFirstRowStream {
         instance_id: Option<ComputeInstanceId>,
         strategy: Option<StatementExecutionStrategy>,
     ) -> Histogram {
-        let isolation_level = *client
-            .session
-            .as_ref()
-            .expect("session invariant")
-            .vars()
-            .transaction_isolation();
+        let session = client.session.as_ref().expect("session invariant");
+        let isolation_level = *session.vars().transaction_isolation();
+        let name_hint = ApplicationNameHint::from_str(session.application_name());
         let instance = match instance_id {
             Some(i) => Cow::Owned(i.to_string()),
             None => Cow::Borrowed("none"),
@@ -1606,6 +1603,7 @@ impl RecordFirstRowStream {
                 instance.as_ref(),
                 isolation_level.as_variant_str(),
                 strategy,
+                name_hint.as_str(),
             ])
     }
 

src/adapter/src/metrics.rs

@@ -140,7 +140,7 @@ impl Metrics {
             time_to_first_row_seconds: registry.register(metric! {
                 name: "mz_time_to_first_row_seconds",
                 help: "Latency of an execute for a successful query from pgwire's perspective",
-                var_labels: ["instance_id", "isolation_level", "strategy"],
+                var_labels: ["instance_id", "isolation_level", "strategy", "application_name"],
                 buckets: histogram_seconds_buckets(0.000_128, 32.0)
             }),
             statement_logging_records: registry.register(metric! {

src/environmentd/tests/server.rs

@@ -3547,6 +3547,52 @@ async fn test_http_metrics() {
     assert_eq!(failure_metric.get_label()[2].value(), "422");
 }
 
+#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))]
+#[cfg_attr(miri, ignore)]
+#[allow(clippy::disallowed_methods)]
+async fn test_time_to_first_row_metric() {
+    let server = test_util::TestHarness::default().start().await;
+
+    let client = server.connect().application_name("dbt").await.unwrap();
+    let _ = client.query_one("SELECT 1", &[]).await.unwrap();
+
+    let metrics = server.metrics_registry.gather();
+    let metric = metrics
+        .into_iter()
+        .find(|m| m.name() == "mz_time_to_first_row_seconds")
+        .expect("mz_time_to_first_row_seconds metric should exist");
+
+    // Find the series produced by our query.
+    let series = metric
+        .get_metric()
+        .iter()
+        .find(|m| {
+            m.get_label()
+                .iter()
+                .any(|l| l.name() == "application_name" && l.value() == "dbt")
+        })
+        .expect("a `mz_time_to_first_row_seconds` series for application_name=dbt should exist");
+
+    // Validate the labels are correct
+    let mut labels: Vec<(&str, &str)> = series
+        .get_label()
+        .iter()
+        .map(|l| (l.name(), l.value()))
+        .collect();
+
+    labels.sort();
+
+    assert_eq!(
+        labels,
+        vec![
+            ("application_name", "dbt"),
+            ("instance_id", "none"),
+            ("isolation_level", "strict serializable"),
+            ("strategy", "constant"),
+        ],
+    );
+}
+
 #[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 2))]
 #[cfg_attr(miri, ignore)] // too slow
 #[allow(clippy::disallowed_methods)]

src/sql/src/session/hint.rs

@@ -91,7 +91,6 @@ impl ApplicationNameHint {
     pub fn from_str(s: &str) -> Self {
         match s.to_lowercase().as_str() {
             "psql" => ApplicationNameHint::Psql(Private),
-            "dbt" => ApplicationNameHint::Dbt(Private),
             "web_console" => ApplicationNameHint::WebConsole(Private),
             "web_console_shell" => ApplicationNameHint::WebConsoleShell(Private),
             "mz_psql" => ApplicationNameHint::MzPsql(Private),
@@ -108,6 +107,8 @@ impl ApplicationNameHint {
             x if x.starts_with("terraform-provider-materialize") => {
                 ApplicationNameHint::TerraformProviderMaterialize(Private)
             }
+            // dbt provides the version as a suffix.
+            x if x.starts_with("dbt") => ApplicationNameHint::Dbt(Private),
             // DataGrip provides the version as a suffix.
             x if x.starts_with("datagrip") => ApplicationNameHint::DataGrip(Private),
             // DBeaver provides the version as a suffix.