feat(logs): Improve logs with spans (#194)

Author: iambriccardo

api/src/main.rs

@@ -7,14 +7,8 @@ use telemetry::init_tracing;
 use tracing::{error, info};
 
 fn main() -> anyhow::Result<()> {
-    let app_name = env!("CARGO_BIN_NAME");
-
-    // We pass emit_on_span_close = true to emit logs on span close
-    // for the api because it is a web server, and we need to emit logs
-    // for every closing request. This is a bit of a hack, but it works
-    // for now. Ideally the tracing middleware should emit a log on
-    // request end, but it doesn't do that yet.
-    let _log_flusher = init_tracing(app_name, true)?;
+    // Initialize tracing from the binary name
+    let _log_flusher = init_tracing(env!("CARGO_BIN_NAME"))?;
 
     // Initialize Sentry before the async runtime starts
     let _sentry_guard = init_sentry()?;

api/src/span_builder.rs

@@ -3,34 +3,60 @@ use actix_web::{
     body::MessageBody,
     dev::{ServiceRequest, ServiceResponse},
 };
-use tracing::Span;
+use tracing::{Span, info};
 use tracing_actix_web::{DefaultRootSpanBuilder, RootSpanBuilder};
 
 /// The `RootSpanBuilder` implementation for the API service.
-/// It extracts the project ref from the `tenant_id` header
-/// and sets it as a field in the root span. If the header is not
-/// present, it sets the field to `Empty` to allow handlers
-/// to set it later.
+///
+/// It extracts the project ref from the `tenant_id` header and sets it as a field in the root span.
+/// It also emits info logs for request start and completion.
+#[derive(Debug)]
 pub struct ApiRootSpanBuilder;
 
 impl RootSpanBuilder for ApiRootSpanBuilder {
     fn on_request_start(request: &ServiceRequest) -> Span {
         let project = request.headers().get("tenant_id");
-        match project {
+        let span = match project {
             Some(project) => {
-                // We convert lossily to a string to be able to read at least part of the
-                // project ref in case of invalid UTF-8. This is useful for debugging.
-                // This is anyways an extreme edge case, as the project ref is
-                // generated by the system and should be valid UTF-8.
+                // We convert lossily to a string to be able to read at least part of the project
+                // ref in case of invalid UTF-8. This is useful for debugging.
+                //
+                // However, this is an edge case, as the project ref is generated by the system and
+                // should be valid UTF-8.
                 let project = String::from_utf8_lossy(project.as_bytes());
                 let project = project.as_ref();
                 tracing_actix_web::root_span!(request, project = project)
             }
             None => tracing_actix_web::root_span!(request, project = tracing::field::Empty),
+        };
+
+        // We enter the span temporarily to log the request received. The span will be entered
+        // automatically before the request is handled.
+        {
+            let _enter = span.enter();
+            info!(
+                method = %request.method(),
+                uri = %request.uri(),
+                path = %request.path(),
+                query_string = %request.query_string(),
+                version = ?request.version(),
+                "HTTP request received"
+            );
         }
+
+        span
     }
 
     fn on_request_end<B: MessageBody>(span: Span, outcome: &Result<ServiceResponse<B>, Error>) {
+        // We call the default builder to add more details to the current span.
         DefaultRootSpanBuilder::on_request_end(span, outcome);
+
+        // In case we have a positive response, we want to log the success. In case of error, it will
+        // be logged automatically since we have the `emit_event_on_error` feature enabled.
+        if let Ok(response) = outcome {
+            if response.response().error().is_none() {
+                info!("HTTP request completed successfully");
+            }
+        }
     }
 }

api/src/startup.rs

@@ -233,7 +233,7 @@ pub async fn run(
     let openapi = ApiDoc::openapi();
 
     let server = HttpServer::new(move || {
-        let tracing_middleware = TracingLogger::<ApiRootSpanBuilder>::new();
+        let tracing_logger = TracingLogger::<ApiRootSpanBuilder>::new();
         let authentication = HttpAuthentication::bearer(auth_validator);
         let app = App::new()
             .wrap(
@@ -242,7 +242,7 @@ pub async fn run(
                     .start_transaction(true)
                     .finish(),
             )
-            .wrap(tracing_middleware)
+            .wrap(tracing_logger)
             .service(health_check)
             .service(
                 SwaggerUi::new("/swagger-ui/{_:.*}").url("/api-docs/openapi.json", openapi.clone()),

etl/src/clients/bigquery.rs

@@ -177,7 +177,7 @@ impl BigQueryClient {
             "".to_string()
         };
 
-        info!("Creating table {full_table_name} in BigQuery");
+        info!("creating table {full_table_name} in BigQuery");
 
         let query = format!("create table {full_table_name} {columns_spec} {max_staleness_option}");
 

etl/src/pipeline.rs

@@ -86,6 +86,10 @@ where
         }
     }
 
+    pub fn id(&self) -> PipelineId {
+        self.id
+    }
+
     pub fn shutdown_tx(&self) -> ShutdownTx {
         self.shutdown_tx.clone()
     }

etl/src/replication/client.rs

@@ -37,6 +37,7 @@ where
         info!("postgres connection terminated successfully")
     }
     .instrument(span);
+
     tokio::spawn(task);
 }
 

etl/src/workers/apply.rs

@@ -122,7 +122,11 @@ where
     async fn start(self) -> Result<ApplyWorkerHandle, Self::Error> {
         info!("starting apply worker");
 
-        let apply_worker_span = tracing::info_span!("apply_worker");
+        let apply_worker_span = tracing::info_span!(
+            "apply_worker",
+            pipeline_id = self.pipeline_id,
+            publication_name = self.config.publication_name
+        );
         let apply_worker = async move {
             let start_lsn = get_start_lsn(self.pipeline_id, &self.replication_client).await?;
 
@@ -151,7 +155,7 @@ where
 
             Ok(())
         }
-        .instrument(apply_worker_span);
+        .instrument(apply_worker_span.or_current());
 
         let handle = tokio::spawn(apply_worker);
 

etl/src/workers/table_sync.rs

@@ -314,8 +314,12 @@ where
         let state = TableSyncWorkerState::new(self.table_id, table_replication_phase);
 
         let state_clone = state.clone();
-        let table_sync_worker_span =
-            tracing::info_span!("table_sync_worker", table_id = self.table_id);
+        let table_sync_worker_span = tracing::info_span!(
+            "table_sync_worker",
+            pipeline_id = self.pipeline_id,
+            publication_name = self.config.publication_name,
+            table_id = self.table_id,
+        );
         let table_sync_worker = async move {
             debug!(
                 "waiting to acquire a running permit for table sync worker for table {}",

replicator/src/core.rs

@@ -13,7 +13,7 @@ use etl::state::store::postgres::PostgresStateStore;
 use etl::{destination::base::Destination, pipeline::PipelineId};
 use secrecy::ExposeSecret;
 use std::fmt;
-use tracing::{info, instrument, warn};
+use tracing::{info, warn};
 
 pub async fn start_replicator() -> anyhow::Result<()> {
     info!("starting replicator service");
@@ -145,7 +145,7 @@ async fn init_state_store(
     Ok(PostgresStateStore::new(pipeline_id, pg_connection_config))
 }
 
-#[instrument(skip(pipeline))]
+#[tracing::instrument(skip(pipeline), fields(pipeline_id = pipeline.id()))]
 async fn start_pipeline<S, D>(mut pipeline: Pipeline<S, D>) -> anyhow::Result<()>
 where
     S: StateStore + Clone + Send + Sync + 'static,

replicator/src/main.rs

@@ -12,12 +12,8 @@ mod core;
 mod migrations;
 
 fn main() -> anyhow::Result<()> {
-    let app_name = env!("CARGO_BIN_NAME");
-
-    // We pass `emit_on_span_close = false` to avoid emitting logs on span close
-    // for replicator because it is not a web server, and we don't need to emit logs
-    // for every closing span.
-    let _log_flusher = init_tracing(app_name, false)?;
+    // Initialize tracing from the binary name
+    let _log_flusher = init_tracing(env!("CARGO_BIN_NAME"))?;
 
     // Initialize Sentry before the async runtime starts
     let _sentry_guard = init_sentry()?;