updated journald

Signed-off-by: wasup-yash <[email protected]>

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "rust-analyzer.linkedProjects": [
+        "./conmon-rs/server/Cargo.toml",
+        "./conmon-rs/server/Cargo.toml"
+    ]
+}

conmon-rs/common/proto/conmon.capnp

@@ -60,6 +60,7 @@ interface Conmon {
             containerRuntimeInterface @0;
             # The JSON logger, requires `path` to be set.
             json @1;
+            journald @2;
         }
     }
 

conmon-rs/server/Cargo.toml

@@ -46,11 +46,10 @@ tracing-opentelemetry = "0.22.0"
 tracing-subscriber = "0.3.18"
 tz-rs = "0.6.14"
 uuid = { version = "1.6.1", features = ["v4", "fast-rng", "macro-diagnostics"] }
-
+systemd = "0.10"
 [build-dependencies]
 shadow-rs = "0.25.0"
 dashmap = "5.5.3"
-
 [dev-dependencies]
 mockall = "0.11.4"
 time = { version = "0.3.30", features = ["parsing"] }

conmon-rs/server/src/container_io.rs

@@ -1,6 +1,6 @@
 use crate::{
-    attach::SharedContainerAttach, container_log::SharedContainerLog, streams::Streams,
-    terminal::Terminal,
+    attach::SharedContainerAttach, container_log::container_log::SharedContainerLog,
+    streams::Streams, terminal::Terminal,
 };
 use anyhow::{bail, Context, Result};
 use getset::{Getters, MutGetters};

conmon-rs/server/src/container_log.rs → conmon-rs/server/src/container_log/container_log.rs

@@ -1,4 +1,7 @@
-use crate::{container_io::Pipe, cri_logger::CriLogger, json_logger::JsonLogger};
+use crate::{
+    container_io::Pipe, container_log::cri_logger::CriLogger,
+    container_log::journald::JournaldLogger, container_log::json_logger::JsonLogger,
+};
 use anyhow::Result;
 use capnp::struct_list::Reader;
 use conmon_common::conmon_capnp::conmon::log_driver::{Owned, Type};
@@ -12,45 +15,53 @@ pub type SharedContainerLog = Arc<RwLock<ContainerLog>>;
 pub struct ContainerLog {
     drivers: Vec<LogDriver>,
 }
-
 #[derive(Debug)]
 enum LogDriver {
     ContainerRuntimeInterface(CriLogger),
+    Journald(JournaldLogger),
     Json(JsonLogger),
 }
 
 impl ContainerLog {
-    /// Create a new default SharedContainerLog.
+    ///Create a new default SharedContainerLog
     pub fn new() -> SharedContainerLog {
         Arc::new(RwLock::new(Self::default()))
     }
-
+    /// Create a new SharedContainerLog from an capnp owned reader.
     pub fn from(reader: Reader<Owned>) -> Result<SharedContainerLog> {
         let drivers = reader
             .iter()
             .map(|x| -> Result<_> {
                 match x.get_type()? {
                     Type::ContainerRuntimeInterface => {
-                        Ok(LogDriver::ContainerRuntimeInterface(CriLogger::new(
+                        Ok(Some(LogDriver::ContainerRuntimeInterface(CriLogger::new(
                             x.get_path()?,
                             if x.get_max_size() > 0 {
                                 Some(x.get_max_size() as usize)
                             } else {
                                 None
                             },
-                        )?))
+                        )?)))
                     }
-                    Type::Json => Ok(LogDriver::Json(JsonLogger::new(
+                    Type::Json => Ok(Some(LogDriver::Json(JsonLogger::new(
                         x.get_path()?,
                         if x.get_max_size() > 0 {
                             Some(x.get_max_size() as usize)
                         } else {
                             None
                         },
-                    )?)),
+                    )?))),
+                    Type::Journald => Ok(Some(LogDriver::Journald(JournaldLogger::new(
+                        if x.get_max_size() > 0 {
+                            Some(x.get_max_size() as usize)
+                        } else {
+                            None
+                        },
+                    )?))),
                 }
             })
-            .collect::<Result<Vec<_>>>()?;
+            .filter_map(Result::transpose)
+            .collect::<Result<Vec<_>, _>>()?;
         Ok(Arc::new(RwLock::new(Self { drivers })))
     }
 
@@ -64,6 +75,7 @@ impl ContainerLog {
                         cri_logger.init().boxed()
                     }
                     LogDriver::Json(ref mut json_logger) => json_logger.init().boxed(),
+                    LogDriver::Journald(ref mut journald_logger) => journald_logger.init().boxed(),
                 })
                 .collect::<Vec<_>>(),
         )
@@ -72,7 +84,6 @@ impl ContainerLog {
         .collect::<Result<Vec<_>>>()?;
         Ok(())
     }
-
     /// Reopen the container logs.
     pub async fn reopen(&mut self) -> Result<()> {
         join_all(
@@ -83,6 +94,9 @@ impl ContainerLog {
                         cri_logger.reopen().boxed()
                     }
                     LogDriver::Json(ref mut json_logger) => json_logger.reopen().boxed(),
+                    LogDriver::Journald(ref mut journald_logger) => {
+                        journald_logger.reopen().boxed()
+                    }
                 })
                 .collect::<Vec<_>>(),
         )
@@ -91,11 +105,10 @@ impl ContainerLog {
         .collect::<Result<Vec<_>>>()?;
         Ok(())
     }
-
     /// Write the contents of the provided reader into all loggers.
     pub async fn write<T>(&mut self, pipe: Pipe, bytes: T) -> Result<()>
     where
-        T: AsyncBufRead + Unpin + Clone,
+        T: AsyncBufRead + Unpin + Clone, // Using Clone to satisfy both Clone and Copy requirements
     {
         let futures = self
             .drivers
@@ -108,16 +121,19 @@ impl ContainerLog {
                 ) -> Result<()> {
                     match logger {
                         LogDriver::ContainerRuntimeInterface(cri_logger) => {
-                            cri_logger.write(pipe, bytes).await
+                            cri_logger.write(pipe, bytes.clone()).await
+                        }
+                        LogDriver::Journald(journald_logger) => {
+                            journald_logger.write(pipe, bytes.clone()).await
+                        }
+                        LogDriver::Json(json_logger) => {
+                            json_logger.write(pipe, bytes.clone()).await
                         }
-                        LogDriver::Json(json_logger) => json_logger.write(pipe, bytes).await,
                     }
                 }
-
                 box_future(x, pipe, bytes.clone())
             })
             .collect::<Vec<_>>();
-
         join_all(futures)
             .await
             .into_iter()