use crate::cfg::{CfgParser, ExtCfgBounds, ProvisionedLifecycle};
use crate::core::env::{Env, GetEnv};
use crate::lifecycle::{LifecycleRepoBuilder, ManageLifecycles, SystemEventSink};
use crate::storage::engine::StorageEngine;
use log::{debug, error, info};
use reduct_base::error::{ErrorCode, ReductError};
use reduct_base::msg::lifecycle_api::{LifecycleMode, LifecycleSettings, LifecycleType};
use std::collections::HashMap;
use std::sync::Arc;
impl<EnvGetter: GetEnv, ExtCfg: ExtCfgBounds> CfgParser<EnvGetter, ExtCfg> {
pub(in crate::cfg) async fn provision_lifecycle_repo(
&self,
storage: Arc<StorageEngine>,
system_event_sink: SystemEventSink,
) -> Result<Box<dyn ManageLifecycles + Send + Sync>, ReductError> {
let mut repo = LifecycleRepoBuilder::new(self.cfg.clone())
.with_system_event_sink(system_event_sink)
.build(Arc::clone(&storage))
.await;
for (name, lifecycle) in &self.cfg.lifecycles {
let mut settings = lifecycle.settings.clone();
if lifecycle.mode_override.is_none() {
if let Ok(info) = repo.get_info(name).await {
settings.mode = info.info.mode;
}
}
if let Err(err) = repo.create_lifecycle(name, settings.clone()).await {
if err.status() == ErrorCode::Conflict {
repo.update_lifecycle(name, settings).await?;
} else {
debug!(
"Skip provisioning lifecycle '{}' because create failed: {}",
name, err
);
continue;
}
}
repo.set_lifecycle_provisioned(name, true).await?;
let info_data = repo.get_info(name).await?;
info!(
"Provisioned lifecycle '{}' with {:?}",
name, info_data.settings
);
}
Ok(repo)
}
pub(in crate::cfg) fn parse_lifecycles(
env: &mut Env<EnvGetter>,
) -> HashMap<String, ProvisionedLifecycle> {
let mut lifecycles = HashMap::<String, (String, ProvisionedLifecycle)>::new();
for (id, name) in env.matches("RS_LIFECYCLE_(.*)_NAME") {
lifecycles.insert(
id,
(
name,
ProvisionedLifecycle {
settings: LifecycleSettings::default(),
mode_override: None,
},
),
);
}
let mut unfinished_lifecycles = vec![];
for (id, (name, lifecycle)) in &mut lifecycles {
if let Some(lifecycle_type) =
env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_TYPE", id))
{
match lifecycle_type.to_lowercase().as_str() {
"delete" => lifecycle.settings.lifecycle_type = LifecycleType::Delete,
"compress" => lifecycle.settings.lifecycle_type = LifecycleType::Compress,
_ => {
error!(
"Lifecycle '{}' has invalid type '{}'. Drop it.",
name, lifecycle_type
);
unfinished_lifecycles.push(id.clone());
continue;
}
}
} else {
error!("Lifecycle '{}' has no type. Drop it.", name);
unfinished_lifecycles.push(id.clone());
continue;
}
if let Some(bucket) = env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_BUCKET", id))
{
lifecycle.settings.bucket = bucket;
} else {
error!("Lifecycle '{}' has no bucket. Drop it.", name);
unfinished_lifecycles.push(id.clone());
continue;
}
if let Some(older_than) =
env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_OLDER_THAN", id))
{
lifecycle.settings.older_than = older_than;
} else {
error!("Lifecycle '{}' has no older_than value. Drop it.", name);
unfinished_lifecycles.push(id.clone());
continue;
}
if let Some(interval) =
env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_INTERVAL", id))
{
lifecycle.settings.interval = interval;
}
if let Some(entries) =
env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_ENTRIES", id))
{
lifecycle.settings.entries = entries
.split(',')
.map(|entry| entry.trim())
.filter(|entry| !entry.is_empty())
.map(|entry| entry.to_string())
.collect();
}
if let Some(when) = env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_WHEN", id)) {
match serde_json::from_str::<serde_json::Value>(&when) {
Ok(when) => lifecycle.settings.when = Some(when),
Err(err) => {
error!(
"Lifecycle '{}' has invalid when condition: {}. Drop it.",
name, err
);
unfinished_lifecycles.push(id.clone());
}
}
}
if let Some(mode) = env.get_optional::<String>(&format!("RS_LIFECYCLE_{}_MODE", id)) {
match mode.to_lowercase().as_str() {
"enabled" => {
lifecycle.settings.mode = LifecycleMode::Enabled;
lifecycle.mode_override = Some(mode);
}
"disabled" => {
lifecycle.settings.mode = LifecycleMode::Disabled;
lifecycle.mode_override = Some(mode);
}
"dry_run" => {
lifecycle.settings.mode = LifecycleMode::DryRun;
lifecycle.mode_override = Some(mode);
}
_ => {
debug!("Lifecycle '{}' has invalid mode '{}'. Drop it.", name, mode);
unfinished_lifecycles.push(id.clone());
}
}
}
}
lifecycles
.into_iter()
.filter(|(id, _)| !unfinished_lifecycles.contains(id))
.map(|(_, (name, lifecycle))| (name, lifecycle))
.collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cfg::Cfg;
use crate::storage::engine::StorageEngine;
use reduct_base::msg::lifecycle_api::{LifecycleInfo, LifecycleMode};
use rstest::{fixture, rstest};
use std::collections::BTreeMap;
use std::env::VarError;
use std::path::PathBuf;
use std::sync::Arc;
#[derive(Clone)]
struct TestEnvGetter {
values: BTreeMap<String, String>,
}
impl TestEnvGetter {
fn new(values: &[(&str, &str)]) -> Self {
Self {
values: values
.iter()
.map(|(key, value)| (key.to_string(), value.to_string()))
.collect(),
}
}
}
impl GetEnv for TestEnvGetter {
fn get(&self, key: &str) -> Result<String, VarError> {
self.values.get(key).cloned().ok_or(VarError::NotPresent)
}
fn all(&self) -> BTreeMap<String, String> {
self.values.clone()
}
}
#[fixture]
fn path() -> PathBuf {
tempfile::tempdir().unwrap().keep()
}
fn lifecycle_env(path: PathBuf, overrides: &[(&str, &str)]) -> TestEnvGetter {
let mut values = BTreeMap::from([
(
"RS_DATA_PATH".to_string(),
path.to_str().unwrap().to_string(),
),
("RS_BUCKET_1_NAME".to_string(), "telemetry".to_string()),
(
"RS_LIFECYCLE_A_NAME".to_string(),
"purge-sensors-30d".to_string(),
),
("RS_LIFECYCLE_A_TYPE".to_string(), "delete".to_string()),
("RS_LIFECYCLE_A_BUCKET".to_string(), "telemetry".to_string()),
(
"RS_LIFECYCLE_A_ENTRIES".to_string(),
"sensors/*, env/temp ,,env/humidity".to_string(),
),
("RS_LIFECYCLE_A_OLDER_THAN".to_string(), "30d".to_string()),
("RS_LIFECYCLE_A_INTERVAL".to_string(), "10m".to_string()),
(
"RS_LIFECYCLE_A_WHEN".to_string(),
r#"{"$eq":["&label","true"]}"#.to_string(),
),
("RS_LIFECYCLE_A_MODE".to_string(), "enabled".to_string()),
]);
for (key, value) in overrides {
values.insert(key.to_string(), value.to_string());
}
TestEnvGetter { values }
}
async fn lifecycle_infos(
env_getter: TestEnvGetter,
) -> (Vec<LifecycleInfo>, Option<LifecycleSettings>, bool) {
let components = CfgParser::from_env(env_getter, "0.0.0")
.await
.build()
.await
.unwrap();
let system_events_enabled = components.cfg.system_events_conf.enabled;
let repo = components.lifecycle_repo.read().await.unwrap();
let lifecycles = repo.lifecycles().await.unwrap();
let settings = match lifecycles.first() {
Some(info) => Some(repo.get_lifecycle_settings(&info.name).await.unwrap()),
None => None,
};
(lifecycles, settings, system_events_enabled)
}
#[rstest]
#[tokio::test]
async fn parses_lifecycle_settings_from_env(path: PathBuf) {
let (_, settings, system_events_enabled) = lifecycle_infos(lifecycle_env(path, &[])).await;
let settings = settings.unwrap();
assert!(system_events_enabled);
assert_eq!(settings.lifecycle_type, LifecycleType::Delete);
assert_eq!(settings.bucket, "telemetry");
assert_eq!(
settings.entries,
vec!["sensors/*", "env/temp", "env/humidity"]
);
assert_eq!(settings.older_than, "30d");
assert_eq!(settings.interval, "10m");
assert_eq!(settings.mode, LifecycleMode::Enabled);
assert_eq!(
settings.when,
Some(serde_json::json!({"$eq": ["&label", "true"]}))
);
}
#[rstest]
#[tokio::test]
async fn drops_lifecycle_without_type(path: PathBuf) {
let mut env_getter = lifecycle_env(path, &[]);
env_getter.values.remove("RS_LIFECYCLE_A_TYPE");
let (lifecycles, settings, _) = lifecycle_infos(env_getter).await;
assert!(lifecycles.is_empty());
assert!(settings.is_none());
}
#[rstest]
#[tokio::test]
async fn drops_lifecycle_without_older_than(path: PathBuf) {
let mut env_getter = lifecycle_env(path, &[]);
env_getter.values.remove("RS_LIFECYCLE_A_OLDER_THAN");
let (lifecycles, settings, _) = lifecycle_infos(env_getter).await;
assert!(lifecycles.is_empty());
assert!(settings.is_none());
}
#[rstest]
fn parse_lifecycles_parses_compress_type() {
let getter = TestEnvGetter::new(&[
("RS_LIFECYCLE_A_NAME", "compress-sensors-30d"),
("RS_LIFECYCLE_A_TYPE", "compress"),
("RS_LIFECYCLE_A_BUCKET", "telemetry"),
("RS_LIFECYCLE_A_OLDER_THAN", "30d"),
]);
let mut env = Env::new(getter);
let lifecycles = CfgParser::<TestEnvGetter>::parse_lifecycles(&mut env);
let lifecycle = lifecycles.get("compress-sensors-30d").unwrap();
assert_eq!(lifecycle.settings.lifecycle_type, LifecycleType::Compress);
}
#[rstest]
#[case("RS_LIFECYCLE_A_TYPE", "archive")]
#[case("RS_LIFECYCLE_A_WHEN", r#"{"$eq":["&label","true"]"#)]
#[case("RS_LIFECYCLE_A_MODE", "paused")]
#[tokio::test]
async fn drops_lifecycle_with_invalid_cfg(
path: PathBuf,
#[case] key: &str,
#[case] value: &str,
) {
let (lifecycles, settings, _) = lifecycle_infos(lifecycle_env(path, &[(key, value)])).await;
assert!(lifecycles.is_empty());
assert!(settings.is_none());
}
#[rstest]
fn parse_lifecycles_trims_entries_and_parses_when() {
let getter = TestEnvGetter::new(&[
("RS_LIFECYCLE_A_NAME", "purge-sensors-30d"),
("RS_LIFECYCLE_A_TYPE", "delete"),
("RS_LIFECYCLE_A_BUCKET", "telemetry"),
(
"RS_LIFECYCLE_A_ENTRIES",
"sensors/*, env/temp ,,env/humidity",
),
("RS_LIFECYCLE_A_OLDER_THAN", "30d"),
("RS_LIFECYCLE_A_INTERVAL", "10m"),
("RS_LIFECYCLE_A_WHEN", r#"{"$eq":["&label","true"]}"#),
]);
let mut env = Env::new(getter);
let lifecycles = CfgParser::<TestEnvGetter>::parse_lifecycles(&mut env);
let lifecycle = lifecycles.get("purge-sensors-30d").unwrap();
assert_eq!(lifecycle.settings.lifecycle_type, LifecycleType::Delete);
assert_eq!(lifecycle.settings.bucket, "telemetry");
assert_eq!(
lifecycle.settings.entries,
vec!["sensors/*", "env/temp", "env/humidity"]
);
assert_eq!(lifecycle.settings.older_than, "30d");
assert_eq!(lifecycle.settings.interval, "10m");
assert_eq!(lifecycle.settings.mode, LifecycleMode::Enabled);
assert_eq!(
lifecycle.settings.when,
Some(serde_json::json!({"$eq": ["&label", "true"]}))
);
}
#[rstest]
fn parse_lifecycles_parses_mode() {
let getter = TestEnvGetter::new(&[
("RS_LIFECYCLE_A_NAME", "purge-sensors-30d"),
("RS_LIFECYCLE_A_TYPE", "delete"),
("RS_LIFECYCLE_A_BUCKET", "telemetry"),
("RS_LIFECYCLE_A_OLDER_THAN", "30d"),
("RS_LIFECYCLE_A_MODE", "disabled"),
]);
let mut env = Env::new(getter);
let lifecycles = CfgParser::<TestEnvGetter>::parse_lifecycles(&mut env);
let lifecycle = lifecycles.get("purge-sensors-30d").unwrap();
assert_eq!(lifecycle.settings.mode, LifecycleMode::Disabled);
assert_eq!(lifecycle.mode_override, Some("disabled".to_string()));
}
#[rstest]
fn parse_lifecycles_parses_dry_run_mode() {
let getter = TestEnvGetter::new(&[
("RS_LIFECYCLE_A_NAME", "purge-sensors-30d"),
("RS_LIFECYCLE_A_TYPE", "delete"),
("RS_LIFECYCLE_A_BUCKET", "telemetry"),
("RS_LIFECYCLE_A_OLDER_THAN", "30d"),
("RS_LIFECYCLE_A_MODE", "dry_run"),
]);
let mut env = Env::new(getter);
let lifecycles = CfgParser::<TestEnvGetter>::parse_lifecycles(&mut env);
let lifecycle = lifecycles.get("purge-sensors-30d").unwrap();
assert_eq!(lifecycle.settings.mode, LifecycleMode::DryRun);
assert_eq!(lifecycle.mode_override, Some("dry_run".to_string()));
}
#[rstest]
fn parse_lifecycles_defaults_mode_when_not_set() {
let getter = TestEnvGetter::new(&[
("RS_LIFECYCLE_A_NAME", "purge-sensors-30d"),
("RS_LIFECYCLE_A_TYPE", "delete"),
("RS_LIFECYCLE_A_BUCKET", "telemetry"),
("RS_LIFECYCLE_A_OLDER_THAN", "30d"),
]);
let mut env = Env::new(getter);
let lifecycles = CfgParser::<TestEnvGetter>::parse_lifecycles(&mut env);
let lifecycle = lifecycles.get("purge-sensors-30d").unwrap();
assert_eq!(lifecycle.settings.mode, LifecycleMode::Enabled);
assert_eq!(lifecycle.mode_override, None);
}
#[rstest]
#[tokio::test]
async fn provision_preserves_existing_mode(path: PathBuf) {
let cfg = Cfg {
data_path: path.clone(),
..Cfg::default()
};
let storage = Arc::new(
StorageEngine::builder()
.with_data_path(path.clone())
.with_cfg(cfg.clone())
.build()
.await,
);
storage
.create_bucket("telemetry", Default::default())
.await
.unwrap();
let mut repo = LifecycleRepoBuilder::new(cfg).build(storage).await;
repo.create_lifecycle(
"purge-sensors-30d",
LifecycleSettings {
lifecycle_type: LifecycleType::Delete,
bucket: "telemetry".to_string(),
entries: vec![],
older_than: "1d".to_string(),
interval: "1h".to_string(),
when: None,
mode: LifecycleMode::Enabled,
},
)
.await
.unwrap();
repo.set_mode("purge-sensors-30d", LifecycleMode::Disabled)
.await
.unwrap();
let mut env_getter = lifecycle_env(path, &[]);
env_getter.values.remove("RS_LIFECYCLE_A_MODE");
let (lifecycles, _, _) = lifecycle_infos(env_getter).await;
assert_eq!(lifecycles.len(), 1);
assert_eq!(lifecycles[0].mode, LifecycleMode::Disabled);
}
#[rstest]
#[tokio::test]
async fn provision_overrides_mode_when_env_set(path: PathBuf) {
let cfg = Cfg {
data_path: path.clone(),
..Cfg::default()
};
let storage = Arc::new(
StorageEngine::builder()
.with_data_path(path.clone())
.with_cfg(cfg.clone())
.build()
.await,
);
storage
.create_bucket("telemetry", Default::default())
.await
.unwrap();
let mut repo = LifecycleRepoBuilder::new(cfg).build(storage).await;
repo.create_lifecycle(
"purge-sensors-30d",
LifecycleSettings {
lifecycle_type: LifecycleType::Delete,
bucket: "telemetry".to_string(),
entries: vec![],
older_than: "1d".to_string(),
interval: "1h".to_string(),
when: None,
mode: LifecycleMode::Enabled,
},
)
.await
.unwrap();
repo.set_mode("purge-sensors-30d", LifecycleMode::Disabled)
.await
.unwrap();
let (lifecycles, _, _) = lifecycle_infos(lifecycle_env(path, &[])).await;
assert_eq!(lifecycles.len(), 1);
assert_eq!(lifecycles[0].mode, LifecycleMode::Enabled);
}
}