use crate::storage::engine::StorageEngine;
use crate::syslog::{LogSystemEvent, SystemEvent};
use async_trait::async_trait;
use bytes::Bytes;
use reduct_base::error::{ErrorCode, ReductError};
use reduct_base::msg::bucket_api::BucketSettings;
use reduct_base::Labels;
use std::sync::Arc;
pub(super) struct LocalSystemLogger {
bucket_name: &'static str,
bucket_settings: BucketSettings,
entry_prefix: Option<&'static str>,
storage: Arc<StorageEngine>,
}
impl LocalSystemLogger {
pub(super) fn new(
bucket_name: &'static str,
bucket_settings: BucketSettings,
entry_prefix: Option<&'static str>,
storage: Arc<StorageEngine>,
) -> Self {
Self {
bucket_name,
bucket_settings,
entry_prefix,
storage,
}
}
async fn log_local(&self, event: SystemEvent) -> Result<(), ReductError> {
let instance = if event.instance.is_empty() {
"unknown".to_string()
} else {
event.instance.clone()
};
let entry_name = match self.entry_prefix {
Some(prefix) => format!("{}/{}/{}", prefix, instance, event.entry_name),
None => format!("{}/{}", instance, event.entry_name),
};
let labels = Labels::from([("status".to_string(), event.status.to_string())]);
let payload = event.to_flat_json()?;
let mut writer = match self
.storage
.begin_write(
self.bucket_name,
&entry_name,
event.timestamp,
payload.len() as u64,
"application/json".to_string(),
labels.clone(),
)
.await
{
Ok(writer) => writer,
Err(err) if err.status == ErrorCode::NotFound => {
self.storage
.create_system_bucket(self.bucket_name, self.bucket_settings.clone())
.await?;
self.storage
.begin_write(
self.bucket_name,
&entry_name,
event.timestamp,
payload.len() as u64,
"application/json".to_string(),
labels,
)
.await?
}
Err(err) => return Err(err),
};
writer.send(Ok(Some(Bytes::from(payload)))).await?;
writer.send(Ok(None)).await?;
Ok(())
}
}
#[async_trait]
impl LogSystemEvent for LocalSystemLogger {
async fn log_event(&mut self, event: SystemEvent) -> Result<(), ReductError> {
self.log_local(event).await
}
}