Add log_request_and_response. Add Request::id field.

Author: mleonhard

Cargo.toml

@@ -24,6 +24,7 @@ include_dir = { version = "0.7", optional = true }
 #libflate = "1"
 once_cell = "1"
 permit = "^0.2.1"
+rand = { version = "^0.8.5", features = ["small_rng"] }
 safe-regex = "^0.2.5"
 safina-executor = { version = "^0.3.3" }
 safina-sync = { version = "^0.2.4" }

examples/events-sse.rs

@@ -23,7 +23,7 @@
 //! ```
 #![forbid(unsafe_code)]
 use permit::Permit;
-use servlin::log::log_response;
+use servlin::log::log_request_and_response;
 use servlin::reexport::{safina_executor, safina_timer};
 use servlin::{
     socket_addr_127_0_0_1, Error, Event, EventSender, HttpServerBuilder, Request, Response,
@@ -62,14 +62,14 @@ fn event_sender_thread(state: Arc<State>, permit: Permit) {
 }
 
 #[allow(clippy::unnecessary_wraps)]
-fn subscribe(state: &Arc<State>, _req: &Request) -> Result<Response, Error> {
+fn subscribe(state: Arc<State>, _req: Request) -> Result<Response, Error> {
     let (sender, response) = Response::event_stream();
     state.subscribers.lock().unwrap().push(sender);
     Ok(response)
 }
 
 #[allow(clippy::unnecessary_wraps)]
-fn handle_req(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn handle_req(state: Arc<State>, req: Request) -> Result<Response, Error> {
     match (req.method(), req.url().path()) {
         ("GET", "/health") => Ok(Response::text(200, "ok")),
         ("GET", "/subscribe") => subscribe(state, req),
@@ -85,7 +85,8 @@ pub fn main() {
     std::thread::spawn(move || event_sender_thread(state_clone, event_sender_thread_permit));
     safina_timer::start_timer_thread();
     let executor = safina_executor::Executor::default();
-    let request_handler = move |req: Request| log_response(&req, handle_req(&state, &req)).unwrap();
+    let request_handler =
+        move |req: Request| log_request_and_response(req, |req| handle_req(state, req));
     executor
         .block_on(
             HttpServerBuilder::new()

examples/html_form.rs

@@ -22,7 +22,7 @@
 //! <http://127.0.0.1:8000/>
 #![forbid(unsafe_code)]
 use serde::Deserialize;
-use servlin::log::log_response;
+use servlin::log::log_request_and_response;
 use servlin::reexport::{safina_executor, safina_timer};
 use servlin::{socket_addr_127_0_0_1, Error, HttpServerBuilder, Request, Response};
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -51,7 +51,7 @@ impl State {
     }
 }
 
-fn index(state: &Arc<State>) -> Response {
+fn index(state: Arc<State>) -> Response {
     Response::html(
         200,
         format!(
@@ -74,12 +74,12 @@ fn index(state: &Arc<State>) -> Response {
     )
 }
 
-fn increment(state: &Arc<State>) -> Response {
+fn increment(state: Arc<State>) -> Response {
     state.increment();
     Response::redirect_303("/")
 }
 
-fn add(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn add(state: Arc<State>, req: Request) -> Result<Response, Error> {
     #[derive(Deserialize)]
     struct Input {
         num: usize,
@@ -94,7 +94,7 @@ fn add(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
     Ok(Response::redirect_303("/"))
 }
 
-fn handle_req(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn handle_req(state: Arc<State>, req: Request) -> Result<Response, Error> {
     match (req.method(), req.url().path()) {
         ("GET", "/health") => Ok(Response::text(200, "ok")),
         ("GET", "/") => Ok(index(state)),
@@ -109,7 +109,8 @@ pub fn main() {
     safina_timer::start_timer_thread();
     let executor = safina_executor::Executor::default();
     let state = Arc::new(State::new());
-    let request_handler = move |req: Request| log_response(&req, handle_req(&state, &req)).unwrap();
+    let request_handler =
+        move |req: Request| log_request_and_response(req, |req| handle_req(state, req));
     executor
         .block_on(
             HttpServerBuilder::new()

examples/http-put.rs

@@ -22,7 +22,7 @@
 //! Upload received, body_len=5, upload_count=2
 //! ```
 #![forbid(unsafe_code)]
-use servlin::log::{log_response, set_global_logger, LogFileWriter};
+use servlin::log::{log_request_and_response, set_global_logger, LogFileWriter};
 use servlin::reexport::{safina_executor, safina_timer};
 use servlin::{socket_addr_127_0_0_1, Error, HttpServerBuilder, Request, Response};
 use std::io::Read;
@@ -43,7 +43,7 @@ impl State {
     }
 }
 
-fn put(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn put(state: Arc<State>, req: Request) -> Result<Response, Error> {
     if req.body.is_pending() {
         return Ok(Response::get_body_and_reprocess(1024 * 1024));
     }
@@ -59,7 +59,7 @@ fn put(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
     ))
 }
 
-fn handle_req(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn handle_req(state: Arc<State>, req: Request) -> Result<Response, Error> {
     match (req.method(), req.url().path()) {
         ("GET", "/health") => Ok(Response::text(200, "ok")),
         ("PUT", "/upload") => put(state, req),
@@ -80,7 +80,8 @@ pub fn main() {
     let executor = safina_executor::Executor::default();
     let cache_dir = TempDir::new().unwrap();
     let state = Arc::new(State::new());
-    let request_handler = move |req: Request| log_response(&req, handle_req(&state, &req)).unwrap();
+    let request_handler =
+        move |req: Request| log_request_and_response(req, |req| handle_req(state, req));
     executor
         .block_on(
             HttpServerBuilder::new()

examples/json_api.rs

@@ -37,7 +37,7 @@
 #![forbid(unsafe_code)]
 use serde::Deserialize;
 use serde_json::json;
-use servlin::log::log_response;
+use servlin::log::log_request_and_response;
 use servlin::reexport::{safina_executor, safina_timer};
 use servlin::{socket_addr_127_0_0_1, Error, HttpServerBuilder, Request, Response};
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -66,16 +66,16 @@ impl State {
     }
 }
 
-fn get_count(state: &Arc<State>) -> Response {
+fn get_count(state: Arc<State>) -> Response {
     Response::json(200, json!({ "count": state.get() })).unwrap()
 }
 
-fn increment(state: &Arc<State>) -> Response {
+fn increment(state: Arc<State>) -> Response {
     state.increment();
     Response::json(200, json!({ "count": state.get() })).unwrap()
 }
 
-fn add(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn add(state: Arc<State>, req: Request) -> Result<Response, Error> {
     #[derive(Deserialize)]
     struct Input {
         num: usize,
@@ -90,7 +90,7 @@ fn add(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
     Ok(Response::json(200, json!({ "count": state.get() })).unwrap())
 }
 
-fn handle_req(state: &Arc<State>, req: &Request) -> Result<Response, Error> {
+fn handle_req(state: Arc<State>, req: Request) -> Result<Response, Error> {
     match (req.method(), req.url().path()) {
         ("GET", "/health") => Ok(Response::text(200, "ok")),
         ("GET", "/get") => Ok(get_count(state)),
@@ -105,7 +105,8 @@ pub fn main() {
     safina_timer::start_timer_thread();
     let executor = safina_executor::Executor::default();
     let state = Arc::new(State::new());
-    let request_handler = move |req: Request| log_response(&req, handle_req(&state, &req)).unwrap();
+    let request_handler =
+        move |req: Request| log_request_and_response(req, |req| handle_req(state, req));
     executor
         .block_on(
             HttpServerBuilder::new()

src/lib.rs

@@ -138,6 +138,7 @@ mod headers;
 mod http_conn;
 mod http_error;
 pub mod log;
+mod rand;
 mod request;
 mod request_body;
 mod response;

src/log/logger.rs

@@ -3,6 +3,7 @@ use crate::log::tag::Tag;
 use crate::log::tag_list::TagList;
 use crate::log::tag_value::TagValue;
 use crate::log::Level;
+use crate::Request;
 use std::cell::RefCell;
 use std::io::Write;
 use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
@@ -139,6 +140,12 @@ pub fn with_thread_local_log_tags<R, F: FnOnce(&[Tag]) -> R>(f: F) -> R {
     THREAD_LOCAL_TAGS.with(|cell| f(cell.borrow().as_slice()))
 }
 
+pub fn add_thread_local_log_tags_from_request(req: &Request) {
+    add_thread_local_log_tag("http_method", req.method());
+    add_thread_local_log_tag("path", req.url().path());
+    add_thread_local_log_tag("request_id", req.id);
+}
+
 #[allow(clippy::module_name_repetitions)]
 #[derive(Debug)]
 pub struct LoggerStoppedError {}

src/log/mod.rs

@@ -12,8 +12,8 @@ pub use log_file_writer::LogFileWriter;
 use logger::log;
 pub use logger::set_global_logger;
 pub use logger::{
-    add_thread_local_log_tag, clear_thread_local_log_tags, with_thread_local_log_tags,
-    LoggerStoppedError,
+    add_thread_local_log_tag, add_thread_local_log_tags_from_request, clear_thread_local_log_tags,
+    with_thread_local_log_tags, LoggerStoppedError,
 };
 use std::fmt::{Display, Formatter};
 use std::time::SystemTime;
@@ -83,17 +83,11 @@ pub fn debug(msg: impl Into<String>, tags: impl Into<TagList>) -> Result<(), Log
 ///
 /// # Errors
 /// Returns `Err` when the global logger has stopped.
-#[allow(clippy::needless_pass_by_value)]
 #[allow(clippy::module_name_repetitions)]
-pub fn log_response(
-    req: &Request,
-    result: Result<Response, Error>,
-) -> Result<Response, LoggerStoppedError> {
+pub fn log_response(result: Result<Response, Error>) -> Result<Response, LoggerStoppedError> {
     match result {
         Ok(response) => {
             let mut tags = Vec::new();
-            tags.push(Tag::new("http_method", req.method()));
-            tags.push(Tag::new("path", req.url().path()));
             tags.push(Tag::new("code", response.code));
             if let Some(body_len) = response.body.len() {
                 tags.push(Tag::new("body_len", body_len));
@@ -112,8 +106,6 @@ pub fn log_response(
             if let Some(backtrace) = e.backtrace {
                 tags.push(Tag::new("msg", format!("{backtrace:?}")));
             }
-            tags.push(Tag::new("http_method", req.method()));
-            tags.push(Tag::new("path", req.url().path()));
             tags.push(Tag::new("code", response.code));
             if let Some(body_len) = response.body.len() {
                 tags.push(Tag::new("body_len", body_len));
@@ -123,3 +115,29 @@ pub fn log_response(
         }
     }
 }
+
+/// Adds thread-local log tags from the request and then calls the handler `f`.
+/// When `f` does logging, the log messages will include the request id, HTTP method, and path.
+/// When `f` returns, this function makes a new log event for the result
+/// and sends it to the global logger.
+///
+/// When the result of `f` is an [`Error`] without a response,
+/// this function uses [`Response::internal_server_errror_500`] to make one.
+///
+/// Returns the response.
+///
+/// Clears thread-local log tags.
+///
+/// # Errors
+/// Returns `Err` when the global logger has stopped.
+#[allow(clippy::module_name_repetitions)]
+pub fn log_request_and_response<F: FnOnce(Request) -> Result<Response, Error>>(
+    req: Request,
+    f: F,
+) -> Response {
+    clear_thread_local_log_tags();
+    add_thread_local_log_tags_from_request(&req);
+    let response = log_response(f(req)).unwrap();
+    clear_thread_local_log_tags();
+    response
+}

src/rand.rs

@@ -0,0 +1,11 @@
+use rand::rngs::SmallRng;
+use rand::{RngCore, SeedableRng};
+use std::cell::RefCell;
+
+thread_local! {
+    pub static THREAD_LOCAL_SMALL_RNG: RefCell<SmallRng> = RefCell::new(SmallRng::from_entropy());
+}
+
+pub fn next_insecure_rand_u64() -> u64 {
+    THREAD_LOCAL_SMALL_RNG.with(|cell| cell.borrow_mut().next_u64())
+}

src/request.rs

@@ -1,5 +1,6 @@
 use crate::head::read_http_head;
 use crate::http_error::HttpError;
+use crate::rand::next_insecure_rand_u64;
 use crate::{AsciiString, ContentType, HeaderList, RequestBody, Response};
 use fixed_buffer::FixedBuf;
 use futures_io::AsyncRead;
@@ -10,6 +11,7 @@ use url::Url;
 
 #[derive(Clone, Eq, PartialEq)]
 pub struct Request {
+    pub id: u64,
     pub remote_addr: SocketAddr,
     pub method: String,
     pub url: Url,
@@ -240,6 +242,7 @@ pub async fn read_http_request<const BUF_SIZE: usize>(
         (false, None, _) => RequestBody::empty(),
     };
     Ok(Request {
+        id: next_insecure_rand_u64(),
         remote_addr,
         method: head.method,
         url: head.url,