logger/proto: remove dedup protobuf defs

shuttle-hq · iulianbarbu · Aug 24, 2023 · Aug 22, 2023 · Aug 22, 2023 · Aug 22, 2023
commit 53e73695a650dd0421fcec990e0e3dc3d34a3965
diff --git a/logger/src/dal.rs b/logger/src/dal.rs
@@ -129,11 +129,12 @@ pub struct Log {
 }
 
 impl Log {
-    pub(crate) fn from_stored(log: LogItem) -> Self {
-        let timestamp = log.tx_timestamp.clone().unwrap_or_default();
-        Log {
+    pub(crate) fn from_log_item(log: LogItem) -> Option<Self> {
+        let log_line = log.log_line?;
+        let timestamp = log_line.tx_timestamp.clone().unwrap_or_default();
+        Some(Log {
             deployment_id: log.deployment_id,
-            shuttle_service_name: log.service_name,
+            shuttle_service_name: log_line.service_name,
             tx_timestamp: DateTime::from_utc(
                 NaiveDateTime::from_timestamp_opt(
                     timestamp.seconds,
@@ -142,18 +143,20 @@ impl Log {
                 .unwrap_or_default(),
                 Utc,
             ),
-            data: log.data,
-        }
+            data: log_line.data,
+        })
     }
 }
 
 impl From<Log> for LogItem {
     fn from(log: Log) -> Self {
         LogItem {
-            service_name: log.shuttle_service_name,
             deployment_id: log.deployment_id,
-            tx_timestamp: Some(Timestamp::from(SystemTime::from(log.tx_timestamp))),
-            data: log.data,
+            log_line: Some(LogLine {
+                service_name: log.shuttle_service_name,
+                tx_timestamp: Some(Timestamp::from(SystemTime::from(log.tx_timestamp))),
+                data: log.data,
+            }),
         }
     }
 }

diff --git a/logger/src/lib.rs b/logger/src/lib.rs
@@ -65,7 +65,7 @@ where
         if !logs.is_empty() {
             _ = self
                 .logs_tx
-                .broadcast(logs.into_iter().map(Log::from_stored).collect())
+                .broadcast(logs.into_iter().filter_map(Log::from_log_item).collect())
                 .await
                 .map_err(|err| {
                     Status::internal(format!(

diff --git a/logger/tests/integration_tests.rs b/logger/tests/integration_tests.rs
@@ -32,19 +32,23 @@ async fn store_and_get_logs() {
         let expected_stored_logs = vec![
             LogItem {
                 deployment_id: deployment_id.to_string(),
-                service_name: SHUTTLE_SERVICE.to_string(),
-                tx_timestamp: Some(Timestamp::from(SystemTime::UNIX_EPOCH)),
-                data: "log 1 example".as_bytes().to_vec(),
+                log_line: Some(LogLine {
+                    service_name: SHUTTLE_SERVICE.to_string(),
+                    tx_timestamp: Some(Timestamp::from(SystemTime::UNIX_EPOCH)),
+                    data: "log 1 example".as_bytes().to_vec(),
+                }),
             },
             LogItem {
                 deployment_id: deployment_id.to_string(),
-                service_name: SHUTTLE_SERVICE.to_string(),
-                tx_timestamp: Some(Timestamp::from(
-                    SystemTime::UNIX_EPOCH
-                        .checked_add(Duration::from_secs(10))
-                        .unwrap(),
-                )),
-                data: "log 2 example".as_bytes().to_vec(),
+                log_line: Some(LogLine {
+                    service_name: SHUTTLE_SERVICE.to_string(),
+                    tx_timestamp: Some(Timestamp::from(
+                        SystemTime::UNIX_EPOCH
+                            .checked_add(Duration::from_secs(10))
+                            .unwrap(),
+                    )),
+                    data: "log 2 example".as_bytes().to_vec(),
+                }),
             },
         ];
         let response = client
@@ -69,7 +73,7 @@ async fn store_and_get_logs() {
             logs,
             expected_stored_logs
                 .into_iter()
-                .map(Into::into)
+                .map(|log| log.log_line.unwrap())
                 .collect::<Vec<LogLine>>()
         );
     });
@@ -95,19 +99,23 @@ async fn get_stream_logs() {
         let expected_stored_logs = vec![
             LogItem {
                 deployment_id: deployment_id.to_string(),
-                service_name: SHUTTLE_SERVICE.to_string(),
-                tx_timestamp: Some(Timestamp::from(SystemTime::UNIX_EPOCH)),
-                data: "log 1 example".as_bytes().to_vec(),
+                log_line: Some(LogLine {
+                    service_name: SHUTTLE_SERVICE.to_string(),
+                    tx_timestamp: Some(Timestamp::from(SystemTime::UNIX_EPOCH)),
+                    data: "log 1 example".as_bytes().to_vec(),
+                }),
             },
             LogItem {
                 deployment_id: deployment_id.to_string(),
-                service_name: SHUTTLE_SERVICE.to_string(),
-                tx_timestamp: Some(Timestamp::from(
-                    SystemTime::UNIX_EPOCH
-                        .checked_add(Duration::from_secs(10))
-                        .unwrap(),
-                )),
-                data: "log 2 example".as_bytes().to_vec(),
+                log_line: Some(LogLine {
+                    service_name: SHUTTLE_SERVICE.to_string(),
+                    tx_timestamp: Some(Timestamp::from(
+                        SystemTime::UNIX_EPOCH
+                            .checked_add(Duration::from_secs(10))
+                            .unwrap(),
+                    )),
+                    data: "log 2 example".as_bytes().to_vec(),
+                }),
             },
         ];
 
@@ -134,14 +142,14 @@ async fn get_stream_logs() {
             .unwrap()
             .unwrap()
             .unwrap();
-        assert_eq!(LogLine::from(expected_stored_logs[0].clone()), log);
+        assert_eq!(expected_stored_logs[0].clone().log_line.unwrap(), log);
 
         let log = timeout(std::time::Duration::from_millis(500), response.message())
             .await
             .unwrap()
             .unwrap()
             .unwrap();
-        assert_eq!(LogLine::from(expected_stored_logs[1].clone()), log);
+        assert_eq!(expected_stored_logs[1].clone().log_line.unwrap(), log);
     });
 
     tokio::select! {

diff --git a/proto/logger.proto b/proto/logger.proto
@@ -31,10 +31,8 @@ message LogsResponse {
 }
 
 message LogItem {
-  string service_name = 1;
-  string deployment_id = 2;
-  google.protobuf.Timestamp tx_timestamp = 3;
-  bytes data = 4;
+  string deployment_id = 1;
+  LogLine log_line = 2;
 }
 
 message LogLine {

diff --git a/proto/src/generated/logger.rs b/proto/src/generated/logger.rs
@@ -26,13 +26,9 @@ pub struct LogsResponse {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct LogItem {
     #[prost(string, tag = "1")]
-    pub service_name: ::prost::alloc::string::String,
-    #[prost(string, tag = "2")]
     pub deployment_id: ::prost::alloc::string::String,
-    #[prost(message, optional, tag = "3")]
-    pub tx_timestamp: ::core::option::Option<::prost_types::Timestamp>,
-    #[prost(bytes = "vec", tag = "4")]
-    pub data: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, optional, tag = "2")]
+    pub log_line: ::core::option::Option<LogLine>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

diff --git a/proto/src/lib.rs b/proto/src/lib.rs
@@ -232,14 +232,4 @@ pub mod resource_recorder {
 
 pub mod logger {
     include!("generated/logger.rs");
-
-    impl From<LogItem> for LogLine {
-        fn from(value: LogItem) -> Self {
-            LogLine {
-                service_name: value.service_name,
-                tx_timestamp: value.tx_timestamp,
-                data: value.data,
-            }
-        }
-    }
 }