Structured tracing on the client

This adds some basic structured tracing on the client. This is useful
both larger applications, in which case you can use a structured
tracing format like opentelemetry to create structured spans with timing
info, contextualized logs, etc.

It also serves as more detailed debug logging on simpler clients, which
can help with debugging.

Author: einarmo

TODO.md

@@ -5,10 +5,8 @@ This is a list of things that are known to be missing, or ideas that could be im
  - Flesh out the server and client SDK with tooling for ease if use.
    - Make it even easier to implement custom node managers.
  - Implement Part 4 7.41.2.3, encrypted secrets. We currently only support legacy secrets. We should also support more encryption algorithms for secrets.
- - Write some form of support for IssuedToken based authentication on the client.
  - Implement a better framework for security checks on the server.
  - Write a sophisticated server example with a persistent store. This would be a great way to verify the flexibility of the server.
  - Write some "bad ideas" servers, it would be nice to showcase how flexible this is.
  - Write a framework for method calls. The foundation for this has been laid with `TryFromVariant`, if we really wanted to we could use clever trait magic to let users simply define a rust method that takes in values that each implement a trait `MethodArg`, with a blanket impl for `TryFromVariant`, and return a tuple of results. Could be really powerful, but methods are a little niche.
  - Implement `Query`. I never got around to this, because the service is just so complex. Currently there is no way to actually implement it, since it won't work unless _all_ node managers implement it, and the core node managers don't.
- - Tracing and detailed logging in the client.

async-opcua-client/src/session/connect.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use tokio::{pin, select};
-use tracing::info;
+use tracing::{info, info_span, Instrument};
 
 use crate::transport::{Connector, SecureChannelEventLoop, TransportPollResult};
 use opcua_types::{NodeId, StatusCode};
@@ -40,9 +40,18 @@ impl SessionConnector {
         &self,
         connector: &T,
     ) -> Result<(SecureChannelEventLoop<T::Transport>, SessionConnectMode), StatusCode> {
-        let mut event_loop = self.inner.channel.connect_no_retry(connector).await?;
-
-        let activate_fut = self.ensure_and_activate_session();
+        let span = info_span!(
+            "Attempting to create and activate session to server",
+            endpoint_url = connector.default_endpoint().endpoint_url.as_ref()
+        );
+        let mut event_loop = self
+            .inner
+            .channel
+            .connect_no_retry(connector)
+            .instrument(span.clone())
+            .await?;
+
+        let activate_fut = self.ensure_and_activate_session().instrument(span.clone());
         pin!(activate_fut);
 
         let res = loop {
@@ -59,7 +68,7 @@ impl SessionConnector {
         let id = match res {
             Ok(id) => id,
             Err(e) => {
-                self.inner.channel.close_channel().await;
+                self.inner.channel.close_channel().instrument(span).await;
 
                 loop {
                     if matches!(event_loop.poll().await, TransportPollResult::Closed(_)) {
@@ -78,18 +87,18 @@ impl SessionConnector {
         let should_create_session = self.inner.session_id.load().is_null();
 
         if should_create_session {
-            self.inner.create_session().await?;
+            self.inner.create_session().in_current_span().await?;
         }
 
-        let reconnect = match self.inner.activate_session().await {
+        let reconnect = match self.inner.activate_session().in_current_span().await {
             Err(status_code) if !should_create_session => {
                 info!(
                     "Session activation failed on reconnect, error = {}, creating a new session",
                     status_code
                 );
                 self.inner.reset();
-                let id = self.inner.create_session().await?;
-                self.inner.activate_session().await?;
+                let id = self.inner.create_session().in_current_span().await?;
+                self.inner.activate_session().in_current_span().await?;
                 SessionConnectMode::NewSession(id)
             }
             Err(e) => return Err(e),
@@ -104,7 +113,10 @@ impl SessionConnector {
         };
 
         if self.inner.recreate_subscriptions {
-            self.inner.transfer_subscriptions_from_old_session().await;
+            self.inner
+                .transfer_subscriptions_from_old_session()
+                .in_current_span()
+                .await;
         }
 
         Ok(reconnect)

async-opcua-client/src/session/mod.rs

@@ -146,7 +146,10 @@ pub(crate) fn process_unexpected_response(response: ResponseMessage) -> StatusCo
             service_fault.response_header.service_result
         }
         _ => {
-            error!("Received an unexpected response to the request");
+            error!(
+                "Received an unexpected response to the request: {}",
+                response.type_name()
+            );
             StatusCode::BadUnknownResponse
         }
     }

async-opcua-client/src/session/services/attributes.rs

@@ -3,7 +3,9 @@ use std::time::Duration;
 use crate::{
     session::{
         process_service_result, process_unexpected_response,
-        request_builder::{builder_base, builder_debug, builder_error, RequestHeaderBuilder},
+        request_builder::{
+            builder_base, builder_debug, builder_error, builder_trace, RequestHeaderBuilder,
+        },
         UARequest,
     },
     AsyncSecureChannel, Session,
@@ -17,6 +19,7 @@ use opcua_types::{
     ReadResponse, ReadValueId, StatusCode, TimestampsToReturn, UpdateDataDetails,
     UpdateEventDetails, UpdateStructureDataDetails, WriteRequest, WriteResponse, WriteValue,
 };
+use tracing::{debug_span, Instrument};
 
 /// Enumeration used with Session::history_read()
 #[derive(Debug, Clone)]
@@ -175,17 +178,32 @@ impl UARequest for Read {
     where
         Self: 'b,
     {
-        if self.nodes_to_read.is_empty() {
-            builder_error!(self, "read(), was not supplied with any nodes to read");
-            return Err(StatusCode::BadNothingToDo);
-        }
-        let request = ReadRequest {
-            request_header: self.header.header,
-            max_age: self.max_age,
-            timestamps_to_return: self.timestamps_to_return,
-            nodes_to_read: Some(self.nodes_to_read),
+        let span = debug_span!(
+            "Sending Read request",
+            nodes_to_read = self.nodes_to_read.len(),
+            timestamps_to_return = ?self.timestamps_to_return,
+            max_age = self.max_age
+        );
+        let request = {
+            let _h = span.enter();
+            if self.nodes_to_read.is_empty() {
+                builder_error!(self, "read(), was not supplied with any nodes to read");
+                return Err(StatusCode::BadNothingToDo);
+            }
+            ReadRequest {
+                request_header: self.header.header,
+                max_age: self.max_age,
+                timestamps_to_return: self.timestamps_to_return,
+                nodes_to_read: Some(self.nodes_to_read),
+            }
         };
-        let response = channel.send(request, self.header.timeout).await?;
+
+        let response = channel
+            .send(request, self.header.timeout)
+            .instrument(span.clone())
+            .await?;
+
+        let _h = span.enter();
         if let ResponseMessage::Read(response) = response {
             builder_debug!(self, "read(), success");
             process_service_result(&response.response_header)?;
@@ -282,25 +300,39 @@ impl UARequest for HistoryRead {
     where
         Self: 'b,
     {
-        let history_read_details = ExtensionObject::from(self.details);
-        builder_debug!(
-            self,
-            "history_read() requested to read nodes {:?}",
-            self.nodes_to_read
+        let span = debug_span!(
+            "Sending HistoryRead request",
+            details = ?self.details,
+            timestamps_to_return = ?self.timestamps_to_return,
+            release_continuation_points = self.release_continuation_points,
+            num_nodes_to_read = self.nodes_to_read.len()
         );
-        let request = HistoryReadRequest {
-            request_header: self.header.header,
-            history_read_details,
-            timestamps_to_return: self.timestamps_to_return,
-            release_continuation_points: self.release_continuation_points,
-            nodes_to_read: if self.nodes_to_read.is_empty() {
-                None
-            } else {
-                Some(self.nodes_to_read)
-            },
+        let request = {
+            let _h = span.enter();
+            let history_read_details = ExtensionObject::from(self.details);
+            builder_trace!(
+                self,
+                "history_read() requested to read nodes {:?}",
+                self.nodes_to_read
+            );
+            HistoryReadRequest {
+                request_header: self.header.header,
+                history_read_details,
+                timestamps_to_return: self.timestamps_to_return,
+                release_continuation_points: self.release_continuation_points,
+                nodes_to_read: if self.nodes_to_read.is_empty() {
+                    None
+                } else {
+                    Some(self.nodes_to_read)
+                },
+            }
         };
 
-        let response = channel.send(request, self.header.timeout).await?;
+        let response = channel
+            .send(request, self.header.timeout)
+            .instrument(span.clone())
+            .await?;
+        let _h = span.enter();
         if let ResponseMessage::HistoryRead(response) = response {
             builder_debug!(self, "history_read(), success");
             process_service_result(&response.response_header)?;
@@ -367,16 +399,26 @@ impl UARequest for Write {
     where
         Self: 'a,
     {
-        if self.nodes_to_write.is_empty() {
-            builder_error!(self, "write() was not supplied with any nodes to write");
-            return Err(StatusCode::BadNothingToDo);
-        }
-
-        let request = WriteRequest {
-            request_header: self.header.header,
-            nodes_to_write: Some(self.nodes_to_write.to_vec()),
+        let span = debug_span!(
+            "Sending Write request",
+            num_nodes_to_write = self.nodes_to_write.len()
+        );
+        let request = {
+            let _h = span.enter();
+            if self.nodes_to_write.is_empty() {
+                builder_error!(self, "write(), was not supplied with any nodes to write");
+                return Err(StatusCode::BadNothingToDo);
+            }
+            WriteRequest {
+                request_header: self.header.header,
+                nodes_to_write: Some(self.nodes_to_write),
+            }
         };
-        let response = channel.send(request, self.header.timeout).await?;
+        let response = channel
+            .send(request, self.header.timeout)
+            .instrument(span.clone())
+            .await?;
+        let _h = span.enter();
         if let ResponseMessage::Write(response) = response {
             builder_debug!(self, "write(), success");
             process_service_result(&response.response_header)?;
@@ -451,23 +493,35 @@ impl UARequest for HistoryUpdate {
     where
         Self: 'a,
     {
-        if self.details.is_empty() {
-            builder_error!(
-                self,
-                "history_update(), was not supplied with any detail to update"
-            );
-            return Err(StatusCode::BadNothingToDo);
-        }
-        let details = self
-            .details
-            .into_iter()
-            .map(ExtensionObject::from)
-            .collect();
-        let request = HistoryUpdateRequest {
-            request_header: self.header.header,
-            history_update_details: Some(details),
+        let span = debug_span!(
+            "Sending HistoryUpdate request",
+            num_details = self.details.len()
+        );
+        let request = {
+            let _h = span.enter();
+            if self.details.is_empty() {
+                builder_error!(
+                    self,
+                    "history_update(), was not supplied with any detail to update"
+                );
+                return Err(StatusCode::BadNothingToDo);
+            }
+            let details = self
+                .details
+                .into_iter()
+                .map(ExtensionObject::from)
+                .collect();
+            HistoryUpdateRequest {
+                request_header: self.header.header,
+                history_update_details: Some(details),
+            }
         };
-        let response = channel.send(request, self.header.timeout).await?;
+
+        let response = channel
+            .send(request, self.header.timeout)
+            .instrument(span.clone())
+            .await?;
+        let _h = span.enter();
         if let ResponseMessage::HistoryUpdate(response) = response {
             builder_error!(self, "history_update(), success");
             process_service_result(&response.response_header)?;

async-opcua-client/src/session/services/method.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 use crate::{
     session::{
         process_unexpected_response,
-        request_builder::{builder_base, builder_debug, builder_error, RequestHeaderBuilder},
+        request_builder::{builder_base, builder_error, RequestHeaderBuilder},
         session_error,
     },
     AsyncSecureChannel, Session, UARequest,
@@ -13,6 +13,7 @@ use opcua_types::{
     CallMethodRequest, CallMethodResult, CallRequest, CallResponse, IntegerId, MethodId, NodeId,
     ObjectId, StatusCode, TryFromVariant, Variant,
 };
+use tracing::{debug_span, Instrument};
 
 #[derive(Debug, Clone)]
 /// Calls a list of methods on the server by sending a [`CallRequest`] to the server.
@@ -68,18 +69,29 @@ impl UARequest for Call {
     where
         Self: 'a,
     {
-        if self.methods.is_empty() {
-            builder_error!(self, "call(), was not supplied with any methods to call");
-            return Err(StatusCode::BadNothingToDo);
-        }
-
-        builder_debug!(self, "call()");
+        let span = debug_span!(
+            "Sending Call request",
+            num_method_calls = self.methods.len()
+        );
         let cnt = self.methods.len();
-        let request = CallRequest {
-            request_header: self.header.header,
-            methods_to_call: Some(self.methods),
+        let request = {
+            let _h = span.enter();
+            if self.methods.is_empty() {
+                builder_error!(self, "call(), was not supplied with any methods to call");
+                return Err(StatusCode::BadNothingToDo);
+            }
+
+            CallRequest {
+                request_header: self.header.header,
+                methods_to_call: Some(self.methods),
+            }
         };
-        let response = channel.send(request, self.header.timeout).await?;
+
+        let response = channel
+            .send(request, self.header.timeout)
+            .instrument(span.clone())
+            .await?;
+        let _h = span.enter();
         if let ResponseMessage::Call(response) = response {
             if let Some(results) = &response.results {
                 if results.len() != cnt {