#![recursion_limit = "512"]
use async_std::task;
use flume::Sender;
use memory_backend::MemoryBackend;
use std::collections::HashMap;
use std::convert::TryFrom;
use std::sync::Arc;
use std::sync::Mutex;
use storages_mgt::StorageMessage;
use zenoh::plugins::{RunningPluginTrait, ZenohPlugin};
use zenoh::prelude::sync::*;
use zenoh::runtime::Runtime;
use zenoh::Session;
use zenoh_backend_traits::config::ConfigDiff;
use zenoh_backend_traits::config::PluginConfig;
use zenoh_backend_traits::config::StorageConfig;
use zenoh_backend_traits::config::VolumeConfig;
use zenoh_backend_traits::VolumeInstance;
use zenoh_core::zlock;
use zenoh_plugin_trait::plugin_long_version;
use zenoh_plugin_trait::plugin_version;
use zenoh_plugin_trait::Plugin;
use zenoh_plugin_trait::PluginControl;
use zenoh_plugin_trait::PluginReport;
use zenoh_plugin_trait::PluginStatusRec;
use zenoh_result::ZResult;
use zenoh_util::LibLoader;
mod backends_mgt;
use backends_mgt::*;
mod memory_backend;
mod replica;
mod storages_mgt;
#[cfg(feature = "dynamic_plugin")]
zenoh_plugin_trait::declare_plugin!(StoragesPlugin);
pub struct StoragesPlugin {}
impl ZenohPlugin for StoragesPlugin {}
impl Plugin for StoragesPlugin {
const DEFAULT_NAME: &'static str = "storage_manager";
const PLUGIN_VERSION: &'static str = plugin_version!();
const PLUGIN_LONG_VERSION: &'static str = plugin_long_version!();
type StartArgs = Runtime;
type Instance = zenoh::plugins::RunningPlugin;
fn start(name: &str, runtime: &Self::StartArgs) -> ZResult<Self::Instance> {
zenoh_util::try_init_log_from_env();
tracing::debug!("StorageManager plugin {}", Self::PLUGIN_VERSION);
let config =
{ PluginConfig::try_from((name, runtime.config().lock().plugin(name).unwrap())) }?;
Ok(Box::new(StorageRuntime::from(StorageRuntimeInner::new(
runtime.clone(),
config,
)?)))
}
}
type PluginsManager = zenoh_plugin_trait::PluginsManager<VolumeConfig, VolumeInstance>;
struct StorageRuntime(Arc<Mutex<StorageRuntimeInner>>);
struct StorageRuntimeInner {
name: String,
runtime: Runtime,
session: Arc<Session>,
storages: HashMap<String, HashMap<String, Sender<StorageMessage>>>,
plugins_manager: PluginsManager,
}
impl StorageRuntimeInner {
fn status_key(&self) -> String {
format!(
"@/router/{}/status/plugins/{}",
&self.runtime.zid(),
&self.name
)
}
fn new(runtime: Runtime, config: PluginConfig) -> ZResult<Self> {
zenoh_util::try_init_log_from_env();
let PluginConfig {
name,
backend_search_dirs,
volumes,
storages,
..
} = config;
let lib_loader = backend_search_dirs
.map(|search_dirs| LibLoader::new(&search_dirs, false))
.unwrap_or_default();
let plugins_manager =
PluginsManager::dynamic(lib_loader.clone(), BACKEND_LIB_PREFIX)
.declare_static_plugin::<MemoryBackend, &str>(MEMORY_BACKEND_NAME, true);
let session = Arc::new(zenoh::init(runtime.clone()).res_sync()?);
let mut new_self = StorageRuntimeInner {
name,
runtime,
session,
storages: Default::default(),
plugins_manager,
};
new_self
.spawn_volume(&VolumeConfig {
name: MEMORY_BACKEND_NAME.into(),
backend: None,
paths: None,
required: false,
rest: Default::default(),
})
.map_or_else(
|e| {
tracing::error!(
"Cannot spawn static volume '{}': {}",
MEMORY_BACKEND_NAME,
e
)
},
|_| (),
);
for volume in &volumes {
new_self.spawn_volume(volume).map_or_else(
|e| tracing::error!("Cannot spawn volume '{}': {}", volume.name(), e),
|_| (),
);
}
for storage in &storages {
new_self.spawn_storage(storage).map_or_else(
|e| tracing::error!("Cannot spawn storage '{}': {}", storage.name(), e),
|_| (),
);
}
Ok(new_self)
}
fn update<I: IntoIterator<Item = ConfigDiff>>(&mut self, diffs: I) -> ZResult<()> {
for ref diff in diffs {
match diff {
ConfigDiff::DeleteVolume(volume) => self.kill_volume(&volume.name)?,
ConfigDiff::AddVolume(volume) => {
self.spawn_volume(volume)?;
}
ConfigDiff::DeleteStorage(config) => self.kill_storage(config),
ConfigDiff::AddStorage(config) => self.spawn_storage(config)?,
}
}
Ok(())
}
fn kill_volume<T: AsRef<str>>(&mut self, name: T) -> ZResult<()> {
let name = name.as_ref();
tracing::info!("Killing volume '{}'", name);
if let Some(storages) = self.storages.remove(name) {
async_std::task::block_on(futures::future::join_all(
storages
.into_values()
.map(|s| async move { s.send(StorageMessage::Stop) }),
));
}
self.plugins_manager
.started_plugin_mut(name)
.ok_or(format!("Cannot find volume '{}' to stop it", name))?
.stop();
Ok(())
}
fn spawn_volume(&mut self, config: &VolumeConfig) -> ZResult<()> {
let volume_id = config.name();
let backend_name = config.backend();
tracing::info!(
"Spawning volume '{}' with backend '{}'",
volume_id,
backend_name
);
let declared = if let Some(declared) = self.plugins_manager.plugin_mut(backend_name) {
declared
} else if let Some(paths) = config.paths() {
self.plugins_manager.declare_dynamic_plugin_by_paths(
backend_name,
volume_id,
paths,
true,
)?
} else {
self.plugins_manager
.declare_dynamic_plugin_by_name(volume_id, backend_name, true)?
};
let loaded = declared.load()?;
loaded.start(config)?;
Ok(())
}
fn kill_storage(&mut self, config: &StorageConfig) {
let volume = &config.volume_id;
tracing::info!("Killing storage '{}' from volume '{}'", config.name, volume);
if let Some(storages) = self.storages.get_mut(volume) {
if let Some(storage) = storages.get_mut(&config.name) {
tracing::debug!(
"Closing storage '{}' from volume '{}'",
config.name,
config.volume_id
);
let _ = storage.send(StorageMessage::Stop); }
}
}
fn spawn_storage(&mut self, storage: &StorageConfig) -> ZResult<()> {
let admin_key = self.status_key() + "/storages/" + &storage.name;
let volume_id = storage.volume_id.clone();
let backend = self
.plugins_manager
.started_plugin(&volume_id)
.ok_or(format!(
"Cannot find volume '{}' to spawn storage '{}'",
volume_id, storage.name
))?;
let storage_name = storage.name.clone();
tracing::info!(
"Spawning storage '{}' from volume '{}' with backend '{}'",
storage_name,
volume_id,
backend.name()
);
let in_interceptor = backend.instance().incoming_data_interceptor();
let out_interceptor = backend.instance().outgoing_data_interceptor();
let stopper = async_std::task::block_on(create_and_start_storage(
admin_key,
storage.clone(),
backend.instance(),
in_interceptor,
out_interceptor,
self.session.clone(),
))?;
self.storages
.entry(volume_id)
.or_default()
.insert(storage_name, stopper);
Ok(())
}
}
impl From<StorageRuntimeInner> for StorageRuntime {
fn from(inner: StorageRuntimeInner) -> Self {
StorageRuntime(Arc::new(Mutex::new(inner)))
}
}
impl PluginControl for StorageRuntime {
fn report(&self) -> PluginReport {
PluginReport::default()
}
fn plugins_status(&self, names: &keyexpr) -> Vec<PluginStatusRec> {
let guard = self.0.lock().unwrap();
guard
.plugins_manager
.plugins_status(names)
.into_iter()
.map(PluginStatusRec::into_owned)
.collect()
}
}
impl RunningPluginTrait for StorageRuntime {
fn config_checker(
&self,
_: &str,
old: &serde_json::Map<String, serde_json::Value>,
new: &serde_json::Map<String, serde_json::Value>,
) -> ZResult<Option<serde_json::Map<String, serde_json::Value>>> {
let name = { zlock!(self.0).name.clone() };
let old = PluginConfig::try_from((&name, old))?;
let new = PluginConfig::try_from((&name, new))?;
tracing::debug!("config change requested for plugin '{}'", name);
tracing::debug!("old config: {:?}", &old);
tracing::debug!("new config: {:?}", &new);
let diffs = ConfigDiff::diffs(old, new);
tracing::debug!("applying diff: {:?}", &diffs);
{ zlock!(self.0).update(diffs) }?;
tracing::debug!("applying diff done");
Ok(None)
}
fn adminspace_getter<'a>(
&'a self,
selector: &'a Selector<'a>,
plugin_status_key: &str,
) -> ZResult<Vec<zenoh::plugins::Response>> {
let mut responses = Vec::new();
let mut key = String::from(plugin_status_key);
with_extended_string(&mut key, &["/version"], |key| {
if keyexpr::new(key.as_str())
.unwrap()
.intersects(&selector.key_expr)
{
responses.push(zenoh::plugins::Response::new(
key.clone(),
StoragesPlugin::PLUGIN_VERSION.into(),
))
}
});
let guard = self.0.lock().unwrap();
with_extended_string(&mut key, &["/volumes/"], |key| {
for plugin in guard.plugins_manager.started_plugins_iter() {
with_extended_string(key, &[plugin.id()], |key| {
with_extended_string(key, &["/__path__"], |key| {
if keyexpr::new(key.as_str())
.unwrap()
.intersects(&selector.key_expr)
{
responses.push(zenoh::plugins::Response::new(
key.clone(),
plugin.path().into(),
))
}
});
if keyexpr::new(key.as_str())
.unwrap()
.intersects(&selector.key_expr)
{
responses.push(zenoh::plugins::Response::new(
key.clone(),
plugin.instance().get_admin_status(),
))
}
});
}
});
with_extended_string(&mut key, &["/storages/"], |key| {
for storages in guard.storages.values() {
for (storage, handle) in storages {
with_extended_string(key, &[storage], |key| {
if keyexpr::new(key.as_str())
.unwrap()
.intersects(&selector.key_expr)
{
if let Ok(value) = task::block_on(async {
let (tx, rx) = async_std::channel::bounded(1);
let _ = handle.send(StorageMessage::GetStatus(tx));
rx.recv().await
}) {
responses.push(zenoh::plugins::Response::new(key.clone(), value))
}
}
})
}
}
});
Ok(responses)
}
}
const BACKEND_LIB_PREFIX: &str = "zenoh_backend_";
const MEMORY_BACKEND_NAME: &str = "memory";
fn with_extended_string<R, F: FnMut(&mut String) -> R>(
prefix: &mut String,
suffixes: &[&str],
mut closure: F,
) -> R {
let prefix_len = prefix.len();
for suffix in suffixes {
prefix.push_str(suffix);
}
let result = closure(prefix);
prefix.truncate(prefix_len);
result
}