impl(gax-internal): recorder and logging (#5148)

Introduce `WithClientLogging` a decorator for `std::future::Future` that
creates the L3 logs entries with the future completes.

Author: coryan

Cargo.lock

@@ -134,6 +134,7 @@ httptest.workspace                = true
 mockall.workspace                 = true
 opentelemetry_sdk                 = { workspace = true, features = ["logs", "metrics", "testing"] }
 opentelemetry-appender-tracing    = { workspace = true }
+pretty_assertions.workspace       = true
 scoped-env.workspace              = true
 serde_with.workspace              = true
 serial_test.workspace             = true

src/gax-internal/Cargo.toml

@@ -44,6 +44,7 @@ mod client_signals;
 #[cfg(google_cloud_unstable_tracing)]
 pub use client_signals::{
     ClientRequestAttributes, ClientSignalsExt, DurationMetric, RequestRecorder, RequestStart,
+    WithClientLogging,
 };
 
 #[cfg(google_cloud_unstable_tracing)]

src/gax-internal/src/observability.rs

@@ -16,12 +16,14 @@ mod client_signals_ext;
 mod duration_metric;
 mod recorder;
 mod request_start;
+mod with_client_logging;
 mod with_client_signals;
 
 pub use client_signals_ext::ClientSignalsExt;
 pub use duration_metric::DurationMetric;
 pub use recorder::{ClientRequestAttributes, RequestRecorder};
 pub use request_start::RequestStart;
+pub use with_client_logging::WithClientLogging;
 pub use with_client_signals::WithClientSignals;
 
 /// An extension to disable terminal actionable error logging.
@@ -213,7 +215,7 @@ mod tests {
         default_host: "example.com",
     };
     pub(crate) static URL_TEMPLATE: &str = "/v1/projects/{}:test_method";
-    pub(crate) static METHOD: &str = "test-method";
+    pub(crate) static METHOD: &str = "google.test.v1.Service/TestMethod";
     pub(crate) const DELAY: Duration = Duration::from_millis(750);
     const COMMON_ATTRIBUTES: [(&str, &str); 3] = [
         ("rpc.system.name", "http"),

src/gax-internal/src/observability/client_signals.rs

@@ -210,7 +210,7 @@ mod tests {
             .build();
         let metric = DurationMetric::new_with_provider(&TEST_INFO, Arc::new(provider.clone()));
         let options = RequestOptions::default().insert_extension(PathTemplate(URL_TEMPLATE));
-        let start = RequestStart::new(&TEST_INFO, &options, "test-method");
+        let start = RequestStart::new(&TEST_INFO, &options, METHOD);
         // Use a long pause so it gets recorded as such.
         tokio::time::sleep(DELAY).await;
         let error = Error::service(
@@ -241,7 +241,7 @@ mod tests {
             .build();
         let metric = DurationMetric::new_with_provider(&TEST_INFO, Arc::new(provider.clone()));
         let options = RequestOptions::default().insert_extension(PathTemplate(URL_TEMPLATE));
-        let start = RequestStart::new(&TEST_INFO, &options, "test-method");
+        let start = RequestStart::new(&TEST_INFO, &options, METHOD);
         // Use a long pause so it gets recorded as such.
         tokio::time::sleep(DELAY).await;
         let error = Error::http(429, http::HeaderMap::new(), bytes::Bytes::new());

src/gax-internal/src/observability/client_signals/duration_metric.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::observability::attributes::RPC_SYSTEM_HTTP;
+use crate::observability::attributes::{GCP_CLIENT_REPO_GOOGLEAPIS, RPC_SYSTEM_HTTP};
 use crate::options::InstrumentationClientInfo;
 #[cfg(feature = "_internal-http-client")]
 use google_cloud_gax::error::Error;
@@ -259,6 +259,34 @@ impl ClientSnapshot {
         self.info.default_host
     }
 
+    /// Returns the service name (e.g. `storage`).
+    ///
+    /// Use with the "gcp.client.service" attribute.
+    pub fn service_name(&self) -> &'static str {
+        self.info.service_name
+    }
+
+    /// Returns the service version (e.g. `1.2.3`).
+    ///
+    /// Use with the "gcp.client.version" attribute.
+    pub fn client_version(&self) -> &'static str {
+        self.info.client_version
+    }
+
+    /// Returns the GitHub repository.
+    ///
+    /// Use with the "gcp.client.repo" attribute.
+    pub fn client_repo(&self) -> &'static str {
+        GCP_CLIENT_REPO_GOOGLEAPIS
+    }
+
+    /// Returns the Rust crate.
+    ///
+    /// Use as instrumentation name, and with the "gcp.client.artifact" attribute.
+    pub fn client_artifact(&self) -> &'static str {
+        self.info.client_artifact
+    }
+
     /// Returns the RPC system (HTTP or gRPC) used in the last low-level request.
     ///
     /// Use with the "rpc.system.name" attribute.
@@ -313,6 +341,30 @@ impl ClientSnapshot {
             .and_then(|s| s.http_status_code)
     }
 
+    /// Returns the HTTP method (e.g. POST) used in the last request.
+    ///
+    /// Note that this may not be populated for gRPC requests.
+    ///
+    /// Use with the "http.request.method" attribute.
+    pub fn http_method(&self) -> Option<&str> {
+        self.transport_snapshot
+            .as_ref()
+            .and_then(|s| s.http_method.as_ref().map(|m| m.as_str()))
+    }
+
+    /// Returns the "resend count" of the last request, if it was a retry.
+    ///
+    /// The resend count of the initial attempt is `None`, and starts at 1 for each retry attempt
+    /// made.
+    ///
+    /// Use with the "http.request.resend_count" attribute.
+    pub fn http_resend_count(&self) -> Option<u32> {
+        if self.attempt_count <= 1 {
+            return None;
+        }
+        Some(self.attempt_count - 1)
+    }
+
     /// Returns the full URL used in the last request.
     ///
     /// Note that this may not be populated for gRPC requests.