use crate::config::Settings; use crate::meta_plugin::{MetaPlugin, MetaPluginType}; use crate::modes::common::settings_meta_plugin_types; use clap::Command; use log::debug; use rusqlite::Connection; use std::collections::HashMap; pub struct MetaService; impl MetaService { pub fn new() -> Self { Self } pub fn get_plugins(&self, cmd: &mut Command, settings: &Settings) -> Vec> { debug!("META_SERVICE: get_plugins called"); let meta_plugin_types: Vec = settings_meta_plugin_types(cmd, settings); debug!("META_SERVICE: Meta plugin types from settings: {:?}", meta_plugin_types); // Create plugins with their configuration let meta_plugins: Vec> = meta_plugin_types .iter() .map(|meta_plugin_type| { debug!("META_SERVICE: Creating plugin: {:?}", meta_plugin_type); // Get the plugin name using strum's Display implementation let plugin_name = meta_plugin_type.to_string(); // Get options and outputs from settings let (options, outputs) = if let Some(meta_plugin_configs) = &settings.meta_plugins { if let Some(config) = meta_plugin_configs.iter().find(|c| c.name == plugin_name) { // Convert options and outputs to the appropriate types let options: std::collections::HashMap = config .options .iter() .map(|(k, v)| (k.clone(), v.clone())) .collect(); let outputs: std::collections::HashMap = config .outputs .iter() .map(|(k, v)| (k.clone(), serde_yaml::Value::String(v.clone()))) .collect(); (Some(options), Some(outputs)) } else { (None, None) } } else { (None, None) }; crate::meta_plugin::get_meta_plugin(meta_plugin_type.clone(), options, outputs) }) .collect(); meta_plugins } pub fn initialize_plugins( &self, plugins: &mut [Box], conn: &Connection, item_id: i64, ) { // Check for duplicate output names before initializing plugins let mut output_names: std::collections::HashMap> = std::collections::HashMap::new(); for plugin in plugins.iter() { let plugin_name = plugin.meta_type().to_string(); // For each plugin, collect all the output names it might write to for (internal_name, output_config) in plugin.outputs() { let output_name = match output_config { serde_yaml::Value::String(remapped_name) => remapped_name.clone(), serde_yaml::Value::Bool(true) => internal_name.clone(), serde_yaml::Value::Bool(false) => continue, // This output is disabled _ => internal_name.clone(), // Default to internal name for other types }; // Only track outputs that will actually be written if !matches!(output_config, serde_yaml::Value::Bool(false)) { output_names.entry(output_name) .or_insert_with(Vec::new) .push(plugin_name.clone()); } } } // Print warnings for duplicate output names for (output_name, plugin_names) in &output_names { if plugin_names.len() > 1 { log::warn!("META_SERVICE: Output name '{}' is provided by multiple plugins: {}", output_name, plugin_names.join(", ")); } } for meta_plugin in plugins.iter_mut() { let response = meta_plugin.initialize(); self.process_plugin_response(conn, item_id, meta_plugin, response); } } pub fn process_chunk( &self, plugins: &mut [Box], chunk: &[u8], conn: &Connection, item_id: i64, ) { for meta_plugin in plugins.iter_mut() { // Skip plugins that are already finalized if meta_plugin.is_finalized() { continue; } let response = meta_plugin.update(chunk); self.process_plugin_response(conn, item_id, meta_plugin, response.clone()); // Set finalized flag if response indicates finalization if response.is_finalized { meta_plugin.set_finalized(true); } } } pub fn finalize_plugins(&self, plugins: &mut [Box], conn: &Connection, item_id: i64) { for meta_plugin in plugins.iter_mut() { // Skip plugins that are already finalized if meta_plugin.is_finalized() { continue; } let response = meta_plugin.finalize(); self.process_plugin_response(conn, item_id, meta_plugin, response.clone()); // Set finalized flag if response indicates finalization if response.is_finalized { meta_plugin.set_finalized(true); } } } fn process_plugin_response( &self, conn: &Connection, item_id: i64, _plugin: &Box, response: crate::meta_plugin::MetaPluginResponse, ) { for meta_data in response.metadata { // The metadata has already been processed by the plugin, so we can use it directly // Save to database let db_meta = crate::db::Meta { id: item_id, name: meta_data.name, value: meta_data.value, }; if let Err(e) = crate::db::store_meta(conn, db_meta) { log::warn!("META_SERVICE: Failed to store metadata: {}", e); } } } pub fn collect_initial_meta(&self) -> HashMap { let mut item_meta: HashMap = crate::modes::common::get_meta_from_env(); if let Ok(hostname) = gethostname::gethostname().into_string() { if !item_meta.contains_key("hostname") { item_meta.insert("hostname".to_string(), hostname); } } item_meta } } impl Default for MetaService { fn default() -> Self { Self::new() } }