reductstore 1.20.2

ReductStore is a time series database designed specifically for storing and managing large amounts of blob data.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

// Copyright 2021-2026 ReductSoftware UG
// Licensed under the Apache License, Version 2.0

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
    }
}