678 lines
18 KiB
Rust
Raw Normal View History

//! Configuration loading and management for the orchestrator
//!
//! This module handles loading configuration from Nickel files with support
//! for:
//! - Nickel-based configuration (schema-driven)
//! - Environment-specific overrides
//! - CLI argument overrides
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use platform_config::ConfigLoader;
use serde::{Deserialize, Serialize};
/// Complete orchestrator configuration
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct OrchestratorConfig {
pub orchestrator: OrchestratorSettings,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrchestratorSettings {
pub workspace: WorkspaceConfig,
pub server: ServerConfig,
pub storage: StorageConfig,
pub queue: QueueConfig,
pub batch: BatchConfig,
#[serde(default)]
pub extensions: ExtensionsConfig,
#[serde(default)]
pub monitoring: Option<MonitoringConfig>,
#[serde(default)]
pub logging: Option<LoggingConfig>,
#[serde(default)]
pub security: Option<SecurityConfig>,
#[serde(default)]
pub performance: Option<PerformanceConfig>,
#[serde(default)]
pub build: Option<DockerBuildConfig>,
}
/// Workspace configuration (from workspace.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkspaceConfig {
pub name: String,
pub path: String,
#[serde(default = "default_true")]
pub enabled: bool,
#[serde(default)]
pub multi_workspace: bool,
}
/// Server configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServerConfig {
pub host: String,
pub port: u16,
#[serde(default)]
pub workers: usize,
#[serde(default)]
pub keep_alive: u64,
#[serde(default)]
pub max_connections: usize,
}
impl Default for ServerConfig {
fn default() -> Self {
Self {
host: "127.0.0.1".to_string(),
port: 9090,
workers: 4,
keep_alive: 75,
max_connections: 1000,
}
}
}
/// Storage configuration (from storage.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StorageConfig {
pub backend: String,
pub path: Option<String>,
#[serde(default)]
pub cache: Option<CacheStorageConfig>,
#[serde(default)]
pub compression: Option<CompressionConfig>,
#[serde(default)]
pub backup: Option<BackupConfig>,
#[serde(default)]
pub replication: Option<ReplicationConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheStorageConfig {
#[serde(default = "default_true")]
pub enabled: bool,
pub cache_type: Option<String>,
pub size: Option<u64>,
pub eviction_policy: Option<String>,
pub ttl: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CompressionConfig {
#[serde(default)]
pub enabled: bool,
pub algorithm: Option<String>,
pub level: Option<u32>,
pub min_size: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BackupConfig {
#[serde(default)]
pub enabled: bool,
pub interval: Option<u64>,
pub path: Option<String>,
pub max_backups: Option<usize>,
#[serde(default)]
pub incremental: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicationConfig {
#[serde(default)]
pub enabled: bool,
pub replicas: Option<Vec<String>>,
pub timeout: Option<u64>,
}
/// Queue configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueConfig {
pub max_concurrent_tasks: usize,
#[serde(default = "default_retry_attempts")]
pub retry_attempts: u32,
#[serde(default = "default_retry_delay")]
pub retry_delay: u64, // milliseconds
#[serde(default = "default_task_timeout")]
pub task_timeout: u64, // milliseconds
#[serde(default = "default_persist")]
pub persist: bool,
#[serde(default)]
pub dead_letter_queue: Option<DeadLetterQueueConfig>,
#[serde(default)]
pub priority_queue: bool,
#[serde(default)]
pub metrics: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeadLetterQueueConfig {
pub enabled: bool,
pub max_size: Option<usize>,
}
impl Default for QueueConfig {
fn default() -> Self {
Self {
max_concurrent_tasks: 10,
retry_attempts: 3,
retry_delay: 5000,
task_timeout: 3600000,
persist: true,
dead_letter_queue: None,
priority_queue: false,
metrics: false,
}
}
}
/// Batch workflow configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchConfig {
#[serde(default = "default_parallel_limit")]
pub parallel_limit: usize,
#[serde(default = "default_operation_timeout")]
pub operation_timeout_minutes: u64,
#[serde(default)]
pub checkpointing: Option<CheckpointingConfig>,
#[serde(default)]
pub rollback: Option<RollbackConfig>,
#[serde(default)]
pub metrics: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckpointingConfig {
pub enabled: bool,
pub interval: Option<u64>,
pub max_checkpoints: Option<usize>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RollbackConfig {
pub enabled: bool,
pub strategy: Option<String>,
pub max_rollback_depth: Option<usize>,
}
impl Default for BatchConfig {
fn default() -> Self {
Self {
parallel_limit: 5,
operation_timeout_minutes: 2,
checkpointing: None,
rollback: None,
metrics: false,
}
}
}
/// Extensions configuration
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ExtensionsConfig {
#[serde(default)]
pub auto_load: bool,
pub oci_registry_url: Option<String>,
pub oci_namespace: Option<String>,
pub discovery_interval: Option<u64>,
pub max_concurrent: Option<usize>,
pub timeout: Option<u64>,
#[serde(default = "default_true")]
pub sandbox: bool,
}
/// Monitoring configuration (from monitoring.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoringConfig {
pub enabled: bool,
#[serde(default)]
pub metrics: Option<MetricsConfig>,
#[serde(default)]
pub health_check: Option<HealthCheckConfig>,
#[serde(default)]
pub tracing: Option<TracingConfig>,
#[serde(default)]
pub alerting: Option<AlertingConfig>,
#[serde(default)]
pub resources: Option<ResourcesMonitoringConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetricsConfig {
#[serde(default)]
pub enabled: bool,
pub interval: Option<u64>,
pub exporters: Option<Vec<String>>,
pub prometheus_path: Option<String>,
pub retention_days: Option<u64>,
pub buffer_size: Option<usize>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthCheckConfig {
#[serde(default)]
pub enabled: bool,
pub check_type: Option<String>,
pub interval: Option<u64>,
pub timeout: Option<u64>,
pub unhealthy_threshold: Option<usize>,
pub healthy_threshold: Option<usize>,
pub endpoint: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TracingConfig {
#[serde(default)]
pub enabled: bool,
pub sample_rate: Option<f32>,
pub backend: Option<String>,
pub endpoint: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertingConfig {
#[serde(default)]
pub enabled: bool,
pub rules_path: Option<String>,
pub channels: Option<Vec<String>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourcesMonitoringConfig {
#[serde(default)]
pub cpu: bool,
#[serde(default)]
pub memory: bool,
#[serde(default)]
pub disk: bool,
#[serde(default)]
pub network: bool,
pub alert_threshold: Option<f64>,
}
impl Default for MonitoringConfig {
fn default() -> Self {
Self {
enabled: true,
metrics: Some(MetricsConfig {
enabled: true,
interval: Some(60),
exporters: None,
prometheus_path: None,
retention_days: None,
buffer_size: None,
}),
health_check: Some(HealthCheckConfig {
enabled: true,
check_type: None,
interval: Some(30),
timeout: None,
unhealthy_threshold: None,
healthy_threshold: None,
endpoint: None,
}),
tracing: None,
alerting: None,
resources: Some(ResourcesMonitoringConfig {
cpu: false,
memory: false,
disk: false,
network: false,
alert_threshold: None,
}),
}
}
}
/// Logging configuration (from logging.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LoggingConfig {
pub level: String,
pub format: String,
#[serde(default)]
pub outputs: Option<Vec<String>>,
#[serde(default)]
pub file: Option<FileLoggingConfig>,
#[serde(default)]
pub syslog: Option<SyslogConfig>,
#[serde(default)]
pub fields: Option<FieldsConfig>,
#[serde(default)]
pub sampling: Option<SamplingConfig>,
#[serde(default)]
pub modules: Option<serde_json::Value>,
#[serde(default)]
pub performance: Option<PerformanceLoggingConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileLoggingConfig {
pub path: Option<String>,
pub max_size: Option<u64>,
pub max_backups: Option<usize>,
pub max_age: Option<u64>,
#[serde(default)]
pub compress: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SyslogConfig {
pub address: Option<String>,
pub facility: Option<String>,
pub protocol: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldsConfig {
#[serde(default = "default_true")]
pub service_name: bool,
#[serde(default = "default_true")]
pub hostname: bool,
#[serde(default = "default_true")]
pub pid: bool,
#[serde(default = "default_true")]
pub timestamp: bool,
#[serde(default)]
pub caller: bool,
#[serde(default)]
pub stack_trace: bool,
pub custom: Option<Vec<String>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SamplingConfig {
#[serde(default)]
pub enabled: bool,
pub initial: Option<u64>,
pub thereafter: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceLoggingConfig {
#[serde(default)]
pub enabled: bool,
pub slow_threshold: Option<u64>,
#[serde(default)]
pub memory_info: bool,
}
impl Default for LoggingConfig {
fn default() -> Self {
Self {
level: "info".to_string(),
format: "text".to_string(),
outputs: None,
file: None,
syslog: None,
fields: None,
sampling: None,
modules: None,
performance: None,
}
}
}
/// Security configuration (from security.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecurityConfig {
#[serde(default)]
pub tls: Option<TlsConfig>,
#[serde(default)]
pub rbac: Option<RbacConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TlsConfig {
pub enabled: bool,
pub cert_path: Option<String>,
pub key_path: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RbacConfig {
pub enabled: bool,
pub policy_file: Option<String>,
}
/// Performance tuning configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceConfig {
#[serde(default)]
pub profiling: bool,
#[serde(default)]
pub cpu_affinity: bool,
#[serde(default)]
pub memory_limits: Option<MemoryLimitsConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MemoryLimitsConfig {
pub max_heap_mb: Option<u64>,
pub gc_threshold: Option<u64>,
}
/// Docker build configuration (from docker-build.ncl schema)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DockerBuildConfig {
pub base_image: String,
#[serde(default)]
pub build_args: HashMap<String, String>,
}
// Serde default helpers
fn default_retry_attempts() -> u32 {
3
}
fn default_retry_delay() -> u64 {
5000
}
fn default_task_timeout() -> u64 {
3600000
}
fn default_persist() -> bool {
true
}
fn default_parallel_limit() -> usize {
5
}
fn default_operation_timeout() -> u64 {
2
}
fn default_true() -> bool {
true
}
impl Default for OrchestratorSettings {
fn default() -> Self {
Self {
workspace: WorkspaceConfig {
name: "default".to_string(),
path: "/tmp".to_string(),
enabled: true,
multi_workspace: false,
},
server: ServerConfig::default(),
storage: StorageConfig {
backend: "filesystem".to_string(),
path: Some("/tmp/orchestrator".to_string()),
cache: None,
compression: None,
backup: None,
replication: None,
},
queue: QueueConfig::default(),
batch: BatchConfig::default(),
extensions: ExtensionsConfig::default(),
monitoring: Some(MonitoringConfig::default()),
logging: Some(LoggingConfig::default()),
security: None,
performance: None,
build: None,
}
}
}
impl OrchestratorConfig {
/// Load configuration from Nickel orchestrator.ncl
pub fn load() -> Result<Self> {
let config_json = platform_config::load_service_config_from_ncl("orchestrator")
.context("Failed to load orchestrator configuration from Nickel")?;
let config: OrchestratorConfig = serde_json::from_value(config_json)
.context("Failed to deserialize orchestrator configuration")?;
let mut config = config;
Self::apply_env_overrides(&mut config)?;
Ok(config)
}
/// Apply environment variable overrides to configuration
/// Environment variables use format: ORCHESTRATOR_{SECTION}_{KEY}=value
fn apply_env_overrides(config: &mut Self) -> Result<()> {
// Server overrides
if let Ok(host) = std::env::var("ORCHESTRATOR_SERVER_HOST") {
config.orchestrator.server.host = host;
}
if let Ok(port) = std::env::var("ORCHESTRATOR_SERVER_PORT") {
config.orchestrator.server.port = port
.parse()
.context("ORCHESTRATOR_SERVER_PORT must be a valid port number")?;
}
if let Ok(workers) = std::env::var("ORCHESTRATOR_SERVER_WORKERS") {
config.orchestrator.server.workers = workers
.parse()
.context("ORCHESTRATOR_SERVER_WORKERS must be a valid number")?;
}
// Logging overrides
if let Ok(level) = std::env::var("ORCHESTRATOR_LOG_LEVEL") {
if let Some(ref mut logging) = config.orchestrator.logging {
logging.level = level;
}
}
Ok(())
}
/// Apply CLI argument overrides to the configuration
pub fn apply_cli_overrides(&mut self, args: &crate::Args) {
if let Some(port) = args.port {
self.orchestrator.server.port = port;
}
}
/// Get server configuration
pub fn server(&self) -> &ServerConfig {
&self.orchestrator.server
}
/// Get storage configuration
pub fn storage(&self) -> &StorageConfig {
&self.orchestrator.storage
}
/// Get queue configuration
pub fn queue(&self) -> &QueueConfig {
&self.orchestrator.queue
}
/// Get batch configuration
pub fn batch(&self) -> &BatchConfig {
&self.orchestrator.batch
}
/// Get extensions configuration
pub fn extensions(&self) -> &ExtensionsConfig {
&self.orchestrator.extensions
}
/// Get logging configuration
pub fn logging(&self) -> Option<&LoggingConfig> {
self.orchestrator.logging.as_ref()
}
/// Get monitoring configuration
pub fn monitoring(&self) -> Option<&MonitoringConfig> {
self.orchestrator.monitoring.as_ref()
}
}
impl ConfigLoader for OrchestratorConfig {
fn service_name() -> &'static str {
"orchestrator"
}
fn load_from_hierarchy() -> std::result::Result<Self, Box<dyn std::error::Error + Send + Sync>>
{
let service = Self::service_name();
if let Some(path) = platform_config::resolve_config_path(service) {
return Self::from_path(&path);
}
Ok(Self::default())
}
fn apply_env_overrides(
&mut self,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
Self::apply_env_overrides(self).map_err(|e| {
Box::new(std::io::Error::other(e.to_string()))
as Box<dyn std::error::Error + Send + Sync>
})
}
fn from_path<P: AsRef<Path>>(
path: P,
) -> std::result::Result<Self, Box<dyn std::error::Error + Send + Sync>> {
let path = path.as_ref();
let json_value = platform_config::format::load_config(path).map_err(|e| {
let err: Box<dyn std::error::Error + Send + Sync> = Box::new(e);
err
})?;
serde_json::from_value(json_value).map_err(|e| {
let err_msg = format!(
"Failed to deserialize orchestrator config from {:?}: {}",
path, e
);
Box::new(std::io::Error::new(
std::io::ErrorKind::InvalidData,
err_msg,
)) as Box<dyn std::error::Error + Send + Sync>
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_default_server_config() {
let server = ServerConfig::default();
assert_eq!(server.host, "127.0.0.1");
assert_eq!(server.port, 9090);
assert_eq!(server.workers, 4);
}
#[test]
fn test_default_queue_config() {
let queue = QueueConfig::default();
assert_eq!(queue.max_concurrent_tasks, 10);
assert_eq!(queue.retry_attempts, 3);
assert_eq!(queue.retry_delay, 5000);
}
#[test]
fn test_default_batch_config() {
let batch = BatchConfig::default();
assert_eq!(batch.parallel_limit, 5);
assert_eq!(batch.operation_timeout_minutes, 2);
}
}