deployment_id in spans, LogRecorder for Batcher

Author: jonaro00

common/src/log.rs

@@ -148,7 +148,7 @@ pub struct DeploymentLogLayer<R>
 where
     R: LogRecorder + Send + Sync,
 {
-    pub recorder: R,
+    pub log_recorder: R,
     pub internal_service: Backend,
 }
 
@@ -170,7 +170,7 @@ where
             let extensions = span.extensions();
 
             if let Some(details) = extensions.get::<ScopeDetails>() {
-                self.recorder.record(LogItem::new(
+                self.log_recorder.record(LogItem::new(
                     details.deployment_id,
                     self.internal_service.clone(),
                     format_event(event),
@@ -210,7 +210,7 @@ where
             metadata.name().blue(),
         );
 
-        self.recorder.record(LogItem::new(
+        self.log_recorder.record(LogItem::new(
             details.deployment_id,
             self.internal_service.clone(),
             message,

deployer/src/deployment/mod.rs

@@ -238,7 +238,7 @@ impl DeploymentManager {
         self.queue_send.send(queued).await.unwrap();
     }
 
-    #[instrument(skip(self), fields(id = %built.id, state = %State::Built))]
+    #[instrument(skip(self), fields(deployment_id = %built.id, state = %State::Built))]
     pub async fn run_push(&self, built: Built) {
         self.run_send.send(built).await.unwrap();
     }

deployer/src/deployment/queue.rs

@@ -112,7 +112,7 @@ pub async fn task(
     }
 }
 
-#[instrument(skip(_id), fields(id = %_id, state = %State::Crashed))]
+#[instrument(skip(_id), fields(deployment_id = %_id, state = %State::Crashed))]
 fn build_failed(_id: &Uuid, error: impl std::error::Error + 'static) {
     error!(
         error = &error as &dyn std::error::Error,
@@ -148,7 +148,7 @@ async fn remove_from_queue(queue_client: impl BuildQueueClient, id: Uuid) {
     }
 }
 
-#[instrument(skip(run_send), fields(id = %built.id, state = %State::Built))]
+#[instrument(skip(run_send), fields(deployment_id = %built.id, state = %State::Built))]
 async fn promote_to_run(mut built: Built, run_send: RunSender) {
     let cx = Span::current().context();
 
@@ -173,7 +173,7 @@ pub struct Queued {
 }
 
 impl Queued {
-    #[instrument(skip(self, storage_manager, deployment_updater, log_recorder, secret_recorder), fields(id = %self.id, state = %State::Building))]
+    #[instrument(skip(self, storage_manager, deployment_updater, log_recorder, secret_recorder), fields(deployment_id = %self.id, state = %State::Building))]
     async fn handle(
         self,
         storage_manager: ArtifactsStorageManager,

deployer/src/deployment/run.rs

@@ -167,25 +167,25 @@ async fn kill_old_deployments(
     Ok(())
 }
 
-#[instrument(skip(_id), fields(id = %_id, state = %State::Completed))]
+#[instrument(skip(_id), fields(deployment_id = %_id, state = %State::Completed))]
 fn completed_cleanup(_id: &Uuid) {
     info!("service finished all on its own");
 }
 
-#[instrument(skip(_id), fields(id = %_id, state = %State::Stopped))]
+#[instrument(skip(_id), fields(deployment_id = %_id, state = %State::Stopped))]
 fn stopped_cleanup(_id: &Uuid) {
     info!("service was stopped by the user");
 }
 
-#[instrument(skip(_id), fields(id = %_id, state = %State::Crashed))]
+#[instrument(skip(_id), fields(deployment_id = %_id, state = %State::Crashed))]
 fn crashed_cleanup(_id: &Uuid, error: impl std::error::Error + 'static) {
     error!(
         error = &error as &dyn std::error::Error,
         "service encountered an error"
     );
 }
 
-#[instrument(skip(_id), fields(id = %_id, state = %State::Crashed))]
+#[instrument(skip(_id), fields(deployment_id = %_id, state = %State::Crashed))]
 fn start_crashed_cleanup(_id: &Uuid, error: impl std::error::Error + 'static) {
     error!(
         error = &error as &dyn std::error::Error,
@@ -215,7 +215,7 @@ pub struct Built {
 }
 
 impl Built {
-    #[instrument(skip(self, storage_manager, secret_getter, resource_manager, runtime_manager, deployment_updater, kill_old_deployments, cleanup), fields(id = %self.id, state = %State::Loading))]
+    #[instrument(skip(self, storage_manager, secret_getter, resource_manager, runtime_manager, deployment_updater, kill_old_deployments, cleanup), fields(deployment_id = %self.id, state = %State::Loading))]
     #[allow(clippy::too_many_arguments)]
     async fn handle(
         self,

deployer/src/deployment/state_change_layer.rs

@@ -7,7 +7,7 @@
 //! This is very similar to Aspect Oriented Programming where we use the annotations from the function to trigger the recording of a new state.
 //! This annotation is a [#[instrument]](https://docs.rs/tracing-attributes/latest/tracing_attributes/attr.instrument.html) with an `id` and `state` field as follow:
 //! ```no-test
-//! #[instrument(fields(id = %built.id, state = %State::Built))]
+//! #[instrument(fields(deployment_id = %built.id, state = %State::Built))]
 //! pub async fn new_state_fn(built: Built) {
 //!     // Get built ready for starting
 //! }
@@ -20,6 +20,7 @@
 use std::str::FromStr;
 
 use chrono::Utc;
+use shuttle_proto::logger::{Batcher, VecReceiver};
 use tracing::{field::Visit, span, warn, Metadata, Subscriber};
 use tracing_subscriber::Layer;
 use uuid::Uuid;
@@ -63,19 +64,19 @@ where
         let mut visitor = NewStateVisitor::default();
         attrs.record(&mut visitor);
 
-        if visitor.id.is_nil() {
+        if visitor.deployment_id.is_nil() {
             warn!("scope details does not have a valid id");
             return;
         }
 
         // To deployer persistence
         self.state_recorder.record_state(DeploymentState {
-            id: visitor.id,
+            id: visitor.deployment_id,
             state: visitor.state,
         });
         // To logger
         self.log_recorder.record(LogItem::new(
-            visitor.id,
+            visitor.deployment_id,
             Backend::Deployer,
             format!(
                 "{} {}",
@@ -86,16 +87,16 @@ where
     }
 }
 
-/// To extract `id` and `state` fields for scopes that have them
+/// To extract `deployment_id` and `state` fields for scopes that have them
 #[derive(Default)]
 struct NewStateVisitor {
-    id: Uuid,
+    deployment_id: Uuid,
     state: State,
 }
 
 impl NewStateVisitor {
     /// Field containing the deployment identifier
-    const ID_IDENT: &'static str = "id";
+    const ID_IDENT: &'static str = "deployment_id";
 
     /// Field containing the deployment state identifier
     const STATE_IDENT: &'static str = "state";
@@ -112,7 +113,7 @@ impl Visit for NewStateVisitor {
         if field.name() == Self::STATE_IDENT {
             self.state = State::from_str(&format!("{value:?}")).unwrap_or_default();
         } else if field.name() == Self::ID_IDENT {
-            self.id = Uuid::try_parse(&format!("{value:?}")).unwrap_or_default();
+            self.deployment_id = Uuid::try_parse(&format!("{value:?}")).unwrap_or_default();
         }
     }
 }

deployer/src/lib.rs

@@ -5,6 +5,7 @@ use hyper::{
     server::conn::AddrStream,
     service::{make_service_fn, service_fn},
 };
+use shuttle_common::log::LogRecorder;
 use shuttle_proto::logger::logger_client::LoggerClient;
 use tokio::sync::Mutex;
 use tracing::{error, info};
@@ -28,6 +29,7 @@ pub use crate::runtime_manager::RuntimeManager;
 pub async fn start(
     persistence: Persistence,
     runtime_manager: Arc<Mutex<RuntimeManager>>,
+    log_recorder: impl LogRecorder,
     log_fetcher: LoggerClient<
         shuttle_common::claims::ClaimService<
             shuttle_common::claims::InjectPropagation<tonic::transport::Channel>,
@@ -37,7 +39,7 @@ pub async fn start(
 ) {
     // when _set is dropped once axum exits, the deployment tasks will be aborted.
     let deployment_manager = DeploymentManager::builder()
-        .build_log_recorder(log_fetcher.clone())
+        .build_log_recorder(log_recorder)
         .secret_recorder(persistence.clone())
         .active_deployment_getter(persistence.clone())
         .artifacts_path(args.artifacts_path)

deployer/src/main.rs

@@ -7,7 +7,7 @@ use shuttle_common::{
     log::{Backend, DeploymentLogLayer},
 };
 use shuttle_deployer::{start, start_proxy, Args, Persistence, RuntimeManager, StateChangeLayer};
-use shuttle_proto::logger::logger_client::LoggerClient;
+use shuttle_proto::logger::{logger_client::LoggerClient, Batcher};
 use tokio::select;
 use tower::ServiceBuilder;
 use tracing::{error, trace};
@@ -40,16 +40,17 @@ async fn main() {
                 .expect("failed to connect to logger"),
         );
     let logger_client = LoggerClient::new(channel);
+    let logger_batcher = Batcher::wrap(logger_client.clone());
 
     setup_tracing(
         tracing_subscriber::registry()
             .with(StateChangeLayer {
-                log_recorder: logger_client.clone(),
+                log_recorder: logger_batcher.clone(),
                 state_recorder: persistence.clone(),
             })
             // TODO: Make all relevant backends set this up in this way
             .with(DeploymentLogLayer {
-                recorder: logger_client.clone(),
+                log_recorder: logger_batcher.clone(),
                 internal_service: Backend::Deployer,
             }),
         Backend::Deployer,
@@ -59,15 +60,15 @@ async fn main() {
         args.artifacts_path.clone(),
         args.provisioner_address.uri().to_string(),
         args.logger_uri.uri().to_string(),
-        logger_client.clone(),
+        logger_batcher.clone(),
         Some(args.auth_uri.to_string()),
     );
 
     select! {
         _ = start_proxy(args.proxy_address, args.proxy_fqdn.clone(), persistence.clone()) => {
             error!("Proxy stopped.")
         },
-        _ = start(persistence, runtime_manager, logger_client, args) => {
+        _ = start(persistence, runtime_manager, logger_batcher, logger_client, args) => {
             error!("Deployment service stopped.")
         },
     }

examples

@@ -236,8 +236,6 @@ pub mod logger {
     use std::time::Duration;
 
     use chrono::{DateTime, NaiveDateTime, Utc};
-    use shuttle_common::log::{LogItem as LogItemCommon, LogRecorder};
-
     use prost::bytes::Bytes;
     use tokio::{select, sync::mpsc, time::interval};
     use tonic::{
@@ -247,6 +245,8 @@ pub mod logger {
     };
     use tracing::error;
 
+    use shuttle_common::log::{LogItem as LogItemCommon, LogRecorder};
+
     use self::logger_client::LoggerClient;
 
     include!("generated/logger.rs");
@@ -276,18 +276,12 @@ pub mod logger {
         }
     }
 
-    impl LogRecorder
-        for logger_client::LoggerClient<
-            shuttle_common::claims::ClaimService<
-                shuttle_common::claims::InjectPropagation<tonic::transport::Channel>,
-            >,
-        >
+    impl<I> LogRecorder for Batcher<I>
+    where
+        I: VecReceiver<Item = LogItemCommon> + Clone + 'static,
     {
         fn record(&self, log: LogItemCommon) {
-            // TODO: Make async + error handling?
-            // self.send_logs(request)
-            //     .await
-            //     .expect("Failed to sens log line");
+            self.send(log);
         }
     }