Use tracing::span

disk/src/cache.rs

@@ -140,7 +140,7 @@ impl Stream for BinnedBytesForHttpStream {
 
 #[derive(Clone, Debug)]
 pub struct PreBinnedQuery {
-    patch: PreBinnedPatchCoord,
+    pub patch: PreBinnedPatchCoord,
     agg_kind: AggKind,
     channel: Channel,
 }

disk/src/raw.rs

@@ -49,7 +49,7 @@ pub async fn x_processed_stream_from_node(
     netout.forget();
     debug!("x_processed_stream_from_node   WRITTEN");
     let frames = InMemoryFrameAsyncReadStream::new(netin);
-    let s2 = MinMaxAvgScalarEventBatchStreamFromFrames { inp: frames };
+    let s2 = MinMaxAvgScalarEventBatchStreamFromFrames::new(frames);
     debug!("x_processed_stream_from_node   HAVE STREAM INSTANCE");
     let s3: Pin<Box<dyn Stream<Item = Result<_, Error>> + Send>> = Box::pin(s2);
     debug!("x_processed_stream_from_node   RETURN");
@@ -63,6 +63,15 @@ where
     inp: InMemoryFrameAsyncReadStream<T>,
 }
 
+impl<T> MinMaxAvgScalarEventBatchStreamFromFrames<T>
+where
+    T: AsyncRead + Unpin,
+{
+    pub fn new(inp: InMemoryFrameAsyncReadStream<T>) -> Self {
+        Self { inp }
+    }
+}
+
 impl<T> Stream for MinMaxAvgScalarEventBatchStreamFromFrames<T>
 where
     T: AsyncRead + Unpin,

httpret/src/lib.rs

@@ -14,8 +14,9 @@ use pin::Pin;
 use std::{future, net, panic, pin, sync, task};
 use sync::Arc;
 use task::{Context, Poll};
+use tracing::field::Empty;
 #[allow(unused_imports)]
-use tracing::{debug, error, info, trace, warn};
+use tracing::{debug, error, info, span, trace, warn, Level};
 
 pub async fn host(node_config: Arc<NodeConfig>) -> Result<(), Error> {
     let rawjh = taskrun::spawn(disk::raw::raw_service(node_config.clone()));
@@ -167,6 +168,7 @@ impl hyper::body::HttpBody for BodyStreamWrap {
 
 struct BodyStream<S> {
     inp: S,
+    desc: String,
 }
 
 impl<S, I> BodyStream<S>
@@ -174,12 +176,12 @@ where
     S: Stream<Item = Result<I, Error>> + Unpin + Send + 'static,
     I: Into<Bytes> + Sized + 'static,
 {
-    pub fn new(inp: S) -> Self {
+    pub fn new(inp: S, desc: String) -> Self {
-        Self { inp }
+        Self { inp, desc }
     }
 
-    pub fn wrapped(inp: S) -> Body {
+    pub fn wrapped(inp: S, desc: String) -> Body {
-        Body::wrap_stream(Self::new(inp))
+        Body::wrap_stream(Self::new(inp, desc))
     }
 }
 
@@ -191,24 +193,28 @@ where
     type Item = Result<I, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        use Poll::*;
+        let span1 = span!(Level::INFO, "httpret::BodyStream", desc = Empty);
-        let t = std::panic::catch_unwind(AssertUnwindSafe(|| self.inp.poll_next_unpin(cx)));
+        span1.record("desc", &self.desc.as_str());
-        match t {
+        span1.in_scope(|| {
-            Ok(r) => match r {
+            use Poll::*;
-                Ready(Some(Ok(k))) => Ready(Some(Ok(k))),
+            let t = std::panic::catch_unwind(AssertUnwindSafe(|| self.inp.poll_next_unpin(cx)));
-                Ready(Some(Err(e))) => {
+            match t {
-                    error!("body stream error: {:?}", e);
+                Ok(r) => match r {
-                    Ready(Some(Err(e.into())))
+                    Ready(Some(Ok(k))) => Ready(Some(Ok(k))),
+                    Ready(Some(Err(e))) => {
+                        error!("body stream error: {:?}", e);
+                        Ready(Some(Err(e.into())))
+                    }
+                    Ready(None) => Ready(None),
+                    Pending => Pending,
+                },
+                Err(e) => {
+                    error!("PANIC CAUGHT in httpret::BodyStream: {:?}", e);
+                    let e = Error::with_msg(format!("PANIC CAUGHT in httpret::BodyStream: {:?}", e));
+                    Ready(Some(Err(e)))
                 }
-                Ready(None) => Ready(None),
-                Pending => Pending,
-            },
-            Err(e) => {
-                error!("PANIC CAUGHT in httpret::BodyStream: {:?}", e);
-                let e = Error::with_msg(format!("PANIC CAUGHT in httpret::BodyStream: {:?}", e));
-                Ready(Some(Err(e)))
             }
-        }
+        })
     }
 }
 
@@ -217,7 +223,7 @@ async fn binned(req: Request<Body>, node_config: Arc<NodeConfig>) -> Result<Resp
     let (head, _body) = req.into_parts();
     let query = disk::cache::Query::from_request(&head)?;
     let ret = match disk::cache::binned_bytes_for_http(node_config, &query).await {
-        Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(s))?,
+        Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(s, format!("desc-BINNED")))?,
         Err(e) => {
             error!("{:?}", e);
             response(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())?
@@ -227,15 +233,22 @@ async fn binned(req: Request<Body>, node_config: Arc<NodeConfig>) -> Result<Resp
 }
 
 async fn prebinned(req: Request<Body>, node_config: Arc<NodeConfig>) -> Result<Response<Body>, Error> {
-    info!("--------------------------------------------------------   PRE-BINNED");
     let (head, _body) = req.into_parts();
     let q = PreBinnedQuery::from_request(&head)?;
-    let ret = match disk::cache::pre_binned_bytes_for_http(node_config, &q) {
+    let span1 = span!(Level::INFO, "httpret::prebinned", bin_t_len = 0);
-        Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(s))?,
+    span1.record("bin_t_len", &q.patch.bin_t_len());
-        Err(e) => {
+    span1.in_scope(|| {
-            error!("{:?}", e);
+        trace!("prebinned");
-            response(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())?
+        let ret = match disk::cache::pre_binned_bytes_for_http(node_config, &q) {
-        }
+            Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(
-    };
+                s,
-    Ok(ret)
+                format!("prebinned-bin-{}-path-{}", q.patch.bin_t_len(), q.patch.patch_beg()),
+            ))?,
+            Err(e) => {
+                error!("{:?}", e);
+                response(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())?
+            }
+        };
+        Ok(ret)
+    })
 }