lit/commands/
datacenter.rs

1//! Datacenter deployment optimizations — cluster node management, object store
2//! sharding, replication factor control, health monitoring, Prometheus-style
3//! metrics, and connection pooling configuration.
4//!
5//! These features enable Lit to operate in distributed datacenter environments
6//! with high availability, fault tolerance, and observability requirements.
7
8use crate::core::find_repo_root;
9use crate::errors::LitError;
10use crate::response::DatacenterResponse;
11use chrono::Utc;
12use serde::{Deserialize, Serialize};
13use std::collections::HashMap;
14use std::fs;
15use std::path::Path;
16
17// ── Data types ──────────────────────────────────────────────────────────────
18
19/// Role a node plays in the cluster
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
21pub enum NodeRole {
22    /// Full read-write primary
23    Primary,
24    /// Read replica
25    Replica,
26    /// Relay / edge cache only
27    Relay,
28    /// Metrics & monitoring observer
29    Observer,
30}
31
32impl std::fmt::Display for NodeRole {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        match self {
35            NodeRole::Primary => write!(f, "primary"),
36            NodeRole::Replica => write!(f, "replica"),
37            NodeRole::Relay => write!(f, "relay"),
38            NodeRole::Observer => write!(f, "observer"),
39        }
40    }
41}
42
43/// Health status of a node
44#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
45pub enum HealthStatus {
46    Healthy,
47    Degraded,
48    Unreachable,
49    Draining,
50    Bootstrapping,
51}
52
53impl std::fmt::Display for HealthStatus {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        match self {
56            HealthStatus::Healthy => write!(f, "healthy"),
57            HealthStatus::Degraded => write!(f, "degraded"),
58            HealthStatus::Unreachable => write!(f, "unreachable"),
59            HealthStatus::Draining => write!(f, "draining"),
60            HealthStatus::Bootstrapping => write!(f, "bootstrapping"),
61        }
62    }
63}
64
65/// Sharding strategy for distributing objects across nodes
66#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
67pub enum ShardStrategy {
68    /// Consistent hash ring (default)
69    ConsistentHash,
70    /// Range-based on object hash prefix
71    RangePrefix,
72    /// Round-robin object distribution
73    RoundRobin,
74    /// Domain-aware — keep a content domain's objects co-located
75    DomainAffinity,
76}
77
78/// Replication mode
79#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
80pub enum ReplicationMode {
81    /// Synchronous — wait for all replicas to confirm
82    Synchronous,
83    /// Async — primary confirms immediately, replicas catch up
84    Asynchronous,
85    /// Semi-sync — wait for at least quorum to confirm
86    SemiSync,
87}
88
89/// A registered datacenter cluster node
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct ClusterNode {
92    /// Unique node identifier
93    pub node_id: String,
94    /// Display name
95    pub name: String,
96    /// Network endpoint (host:port or URL)
97    pub endpoint: String,
98    /// Region / availability zone
99    pub region: String,
100    /// Node role
101    pub role: NodeRole,
102    /// Current health
103    pub health: HealthStatus,
104    /// Shard assignments (hex prefixes this node owns)
105    pub shard_ranges: Vec<String>,
106    /// Last heartbeat timestamp
107    pub last_heartbeat: String,
108    /// Node capacity metrics
109    pub capacity: NodeCapacity,
110    /// ISO 8601 date the node was registered
111    pub registered_at: String,
112}
113
114/// Capacity and utilization metrics for a node
115#[derive(Debug, Clone, Serialize, Deserialize)]
116pub struct NodeCapacity {
117    /// Total storage in bytes
118    pub storage_total: u64,
119    /// Used storage in bytes
120    pub storage_used: u64,
121    /// Object count on this node
122    pub object_count: u64,
123    /// Maximum concurrent connections
124    pub max_connections: u32,
125    /// Current active connections
126    pub active_connections: u32,
127    /// CPU utilization (0.0 – 1.0)
128    pub cpu_utilization: f64,
129    /// Memory utilization (0.0 – 1.0)
130    pub memory_utilization: f64,
131}
132
133/// Cluster-level configuration
134#[derive(Debug, Clone, Serialize, Deserialize)]
135pub struct ClusterConfig {
136    /// Replication factor (how many copies per object)
137    pub replication_factor: u32,
138    /// Replication mode
139    pub replication_mode: ReplicationMode,
140    /// Sharding strategy
141    pub shard_strategy: ShardStrategy,
142    /// Number of virtual shards (powers of 16, e.g. 256 = 2 hex chars)
143    pub shard_count: u32,
144    /// Connection pool size per node
145    pub connection_pool_size: u32,
146    /// Heartbeat interval in seconds
147    pub heartbeat_interval_secs: u32,
148    /// Node timeout before marking unreachable
149    pub node_timeout_secs: u32,
150    /// Whether to enable Prometheus-style metrics endpoint
151    pub metrics_enabled: bool,
152    /// Metrics endpoint port (default: 9090)
153    pub metrics_port: u16,
154    /// Maximum object size before chunked transfer (bytes)
155    pub chunk_threshold: u64,
156    /// Chunk size for large object transfer (bytes)
157    pub chunk_size: u64,
158    /// Compression for inter-node transfer
159    pub transfer_compression: bool,
160    /// Enable read replicas for load balancing reads
161    pub read_load_balance: bool,
162    /// Write concern — how many nodes must confirm a write
163    pub write_concern: u32,
164}
165
166impl Default for ClusterConfig {
167    fn default() -> Self {
168        Self {
169            replication_factor: 3,
170            replication_mode: ReplicationMode::SemiSync,
171            shard_strategy: ShardStrategy::ConsistentHash,
172            shard_count: 256,
173            connection_pool_size: 32,
174            heartbeat_interval_secs: 10,
175            node_timeout_secs: 30,
176            metrics_enabled: true,
177            metrics_port: 9090,
178            chunk_threshold: 64 * 1024 * 1024,
179            chunk_size: 4 * 1024 * 1024,
180            transfer_compression: true,
181            read_load_balance: true,
182            write_concern: 2,
183        }
184    }
185}
186
187/// Prometheus-style metric representation
188#[derive(Debug, Clone, Serialize, Deserialize)]
189pub struct Metric {
190    pub name: String,
191    pub help: String,
192    pub metric_type: String,
193    pub value: f64,
194    pub labels: HashMap<String, String>,
195}
196
197// ── Helpers ─────────────────────────────────────────────────────────────────
198
199fn datacenter_dir(repo_root: &Path) -> std::path::PathBuf {
200    repo_root.join(".lit").join("datacenter")
201}
202
203fn nodes_dir(repo_root: &Path) -> std::path::PathBuf {
204    datacenter_dir(repo_root).join("nodes")
205}
206
207fn load_cluster_config(repo_root: &Path) -> Result<ClusterConfig, LitError> {
208    let path = datacenter_dir(repo_root).join("cluster.json");
209    if path.exists() {
210        let json = fs::read_to_string(&path).map_err(|e| LitError::io(e.to_string()))?;
211        serde_json::from_str(&json)
212            .map_err(|e| LitError::general(format!("Parse cluster config: {}", e)))
213    } else {
214        Ok(ClusterConfig::default())
215    }
216}
217
218fn save_cluster_config(repo_root: &Path, config: &ClusterConfig) -> Result<(), LitError> {
219    let dir = datacenter_dir(repo_root);
220    fs::create_dir_all(&dir).map_err(|e| LitError::io(e.to_string()))?;
221    let json = serde_json::to_string_pretty(config)
222        .map_err(|e| LitError::general(format!("Serialize cluster config: {}", e)))?;
223    fs::write(dir.join("cluster.json"), json).map_err(|e| LitError::io(e.to_string()))?;
224    Ok(())
225}
226
227fn save_node(repo_root: &Path, node: &ClusterNode) -> Result<(), LitError> {
228    let dir = nodes_dir(repo_root);
229    fs::create_dir_all(&dir).map_err(|e| LitError::io(e.to_string()))?;
230    let json = serde_json::to_string_pretty(node)
231        .map_err(|e| LitError::general(format!("Serialize node: {}", e)))?;
232    fs::write(dir.join(format!("{}.json", node.node_id)), json)
233        .map_err(|e| LitError::io(e.to_string()))?;
234    Ok(())
235}
236
237fn load_all_nodes(repo_root: &Path) -> Result<Vec<ClusterNode>, LitError> {
238    let dir = nodes_dir(repo_root);
239    let mut nodes = Vec::new();
240    if dir.exists() {
241        for entry in fs::read_dir(&dir).map_err(|e| LitError::io(e.to_string()))? {
242            let entry = entry.map_err(|e| LitError::io(e.to_string()))?;
243            if entry
244                .path()
245                .extension()
246                .map(|e| e == "json")
247                .unwrap_or(false)
248            {
249                let json =
250                    fs::read_to_string(entry.path()).map_err(|e| LitError::io(e.to_string()))?;
251                if let Ok(node) = serde_json::from_str::<ClusterNode>(&json) {
252                    nodes.push(node);
253                }
254            }
255        }
256    }
257    Ok(nodes)
258}
259
260/// Assign shard ranges to a node based on current cluster state
261fn compute_shard_ranges(node_id: &str, all_nodes: &[ClusterNode], shard_count: u32) -> Vec<String> {
262    let active_nodes: Vec<&ClusterNode> = all_nodes
263        .iter()
264        .filter(|n| n.health != HealthStatus::Unreachable && n.health != HealthStatus::Draining)
265        .collect();
266
267    if active_nodes.is_empty() {
268        return (0..shard_count).map(|i| format!("{:02x}", i)).collect();
269    }
270
271    let pos = active_nodes
272        .iter()
273        .position(|n| n.node_id == node_id)
274        .unwrap_or(active_nodes.len());
275
276    let total = active_nodes.len() as u32;
277    let shards_per_node = shard_count / total.max(1);
278    let start = pos as u32 * shards_per_node;
279    let end = if pos as u32 == total - 1 {
280        shard_count
281    } else {
282        start + shards_per_node
283    };
284
285    (start..end).map(|i| format!("{:02x}", i % 256)).collect()
286}
287
288/// Collect Prometheus-style metrics for the local node
289fn collect_metrics(repo_root: &Path) -> Vec<Metric> {
290    let objects_dir = repo_root.join(".lit").join("objects");
291    let mut object_count: u64 = 0;
292    let mut total_size: u64 = 0;
293
294    if let Ok(entries) = fs::read_dir(&objects_dir) {
295        for shard in entries.flatten() {
296            if shard.file_type().map(|t| t.is_dir()).unwrap_or(false) {
297                if let Ok(files) = fs::read_dir(shard.path()) {
298                    for file in files.flatten() {
299                        object_count += 1;
300                        total_size += file.metadata().map(|m| m.len()).unwrap_or(0);
301                    }
302                }
303            }
304        }
305    }
306
307    let refs_count = repo_root
308        .join(".lit")
309        .join("refs")
310        .read_dir()
311        .map(|e| e.count() as u64)
312        .unwrap_or(0);
313
314    vec![
315        Metric {
316            name: "lit_objects_total".into(),
317            help: "Total number of objects in the local store".into(),
318            metric_type: "gauge".into(),
319            value: object_count as f64,
320            labels: HashMap::new(),
321        },
322        Metric {
323            name: "lit_objects_size_bytes".into(),
324            help: "Total size of all objects in bytes".into(),
325            metric_type: "gauge".into(),
326            value: total_size as f64,
327            labels: HashMap::new(),
328        },
329        Metric {
330            name: "lit_refs_total".into(),
331            help: "Total number of refs".into(),
332            metric_type: "gauge".into(),
333            value: refs_count as f64,
334            labels: HashMap::new(),
335        },
336    ]
337}
338
339// ── Public API ──────────────────────────────────────────────────────────────
340
341/// Show cluster status — nodes, shard distribution, config
342pub fn execute_status() -> Result<DatacenterResponse, LitError> {
343    let repo_root = find_repo_root()?;
344    let config = load_cluster_config(&repo_root)?;
345    let nodes = load_all_nodes(&repo_root)?;
346
347    let healthy = nodes
348        .iter()
349        .filter(|n| n.health == HealthStatus::Healthy)
350        .count();
351    let total = nodes.len();
352
353    Ok(DatacenterResponse {
354        action: "status".into(),
355        message: format!(
356            "Cluster: {} node(s) ({} healthy), replication_factor={}, shards={}, strategy={}",
357            total,
358            healthy,
359            config.replication_factor,
360            config.shard_count,
361            match config.shard_strategy {
362                ShardStrategy::ConsistentHash => "consistent-hash",
363                ShardStrategy::RangePrefix => "range-prefix",
364                ShardStrategy::RoundRobin => "round-robin",
365                ShardStrategy::DomainAffinity => "domain-affinity",
366            }
367        ),
368        details: Some(serde_json::json!({
369            "config": config,
370            "nodes": nodes,
371            "summary": {
372                "total_nodes": total,
373                "healthy_nodes": healthy,
374                "total_storage": nodes.iter().map(|n| n.capacity.storage_total).sum::<u64>(),
375                "used_storage": nodes.iter().map(|n| n.capacity.storage_used).sum::<u64>(),
376                "total_objects": nodes.iter().map(|n| n.capacity.object_count).sum::<u64>(),
377            }
378        })),
379    })
380}
381
382/// Register a new cluster node
383pub fn execute_register_node(
384    node_id: String,
385    name: String,
386    endpoint: String,
387    region: String,
388    role: Option<String>,
389) -> Result<DatacenterResponse, LitError> {
390    let repo_root = find_repo_root()?;
391    let config = load_cluster_config(&repo_root)?;
392    let existing = load_all_nodes(&repo_root)?;
393
394    let role_enum = match role.as_deref() {
395        Some("primary") => NodeRole::Primary,
396        Some("replica") => NodeRole::Replica,
397        Some("relay") => NodeRole::Relay,
398        Some("observer") => NodeRole::Observer,
399        _ => NodeRole::Replica,
400    };
401
402    let node = ClusterNode {
403        node_id: node_id.clone(),
404        name: name.clone(),
405        endpoint,
406        region: region.clone(),
407        role: role_enum,
408        health: HealthStatus::Bootstrapping,
409        shard_ranges: compute_shard_ranges(&node_id, &existing, config.shard_count),
410        last_heartbeat: Utc::now().to_rfc3339(),
411        capacity: NodeCapacity {
412            storage_total: 0,
413            storage_used: 0,
414            object_count: 0,
415            max_connections: config.connection_pool_size,
416            active_connections: 0,
417            cpu_utilization: 0.0,
418            memory_utilization: 0.0,
419        },
420        registered_at: Utc::now().to_rfc3339(),
421    };
422
423    save_node(&repo_root, &node)?;
424
425    Ok(DatacenterResponse {
426        action: "register-node".into(),
427        message: format!(
428            "Node '{}' ({}) registered in region '{}' with {} shard(s)",
429            name,
430            node_id,
431            region,
432            node.shard_ranges.len()
433        ),
434        details: Some(serde_json::to_value(&node).unwrap_or_default()),
435    })
436}
437
438/// Configure cluster-level settings
439#[allow(clippy::too_many_arguments)]
440pub fn execute_configure(
441    replication_factor: Option<u32>,
442    shard_count: Option<u32>,
443    shard_strategy: Option<String>,
444    replication_mode: Option<String>,
445    connection_pool_size: Option<u32>,
446    metrics_enabled: Option<bool>,
447    metrics_port: Option<u16>,
448    write_concern: Option<u32>,
449) -> Result<DatacenterResponse, LitError> {
450    let repo_root = find_repo_root()?;
451    let mut config = load_cluster_config(&repo_root)?;
452
453    if let Some(rf) = replication_factor {
454        config.replication_factor = rf;
455    }
456    if let Some(sc) = shard_count {
457        config.shard_count = sc;
458    }
459    if let Some(ss) = shard_strategy {
460        config.shard_strategy = match ss.as_str() {
461            "consistent-hash" => ShardStrategy::ConsistentHash,
462            "range-prefix" => ShardStrategy::RangePrefix,
463            "round-robin" => ShardStrategy::RoundRobin,
464            "domain-affinity" => ShardStrategy::DomainAffinity,
465            _ => ShardStrategy::ConsistentHash,
466        };
467    }
468    if let Some(rm) = replication_mode {
469        config.replication_mode = match rm.as_str() {
470            "sync" | "synchronous" => ReplicationMode::Synchronous,
471            "async" | "asynchronous" => ReplicationMode::Asynchronous,
472            _ => ReplicationMode::SemiSync,
473        };
474    }
475    if let Some(cps) = connection_pool_size {
476        config.connection_pool_size = cps;
477    }
478    if let Some(me) = metrics_enabled {
479        config.metrics_enabled = me;
480    }
481    if let Some(mp) = metrics_port {
482        config.metrics_port = mp;
483    }
484    if let Some(wc) = write_concern {
485        config.write_concern = wc;
486    }
487
488    save_cluster_config(&repo_root, &config)?;
489
490    Ok(DatacenterResponse {
491        action: "configure".into(),
492        message: "Cluster configuration updated".into(),
493        details: Some(serde_json::to_value(&config).unwrap_or_default()),
494    })
495}
496
497/// Run health checks on all registered nodes
498pub fn execute_health() -> Result<DatacenterResponse, LitError> {
499    let repo_root = find_repo_root()?;
500    let config = load_cluster_config(&repo_root)?;
501    let nodes = load_all_nodes(&repo_root)?;
502
503    let mut health_report: Vec<serde_json::Value> = Vec::new();
504    let timeout_cutoff = Utc::now() - chrono::Duration::seconds(config.node_timeout_secs as i64);
505
506    for node in &nodes {
507        let last_hb = chrono::DateTime::parse_from_rfc3339(&node.last_heartbeat)
508            .map(|dt| dt.with_timezone(&Utc))
509            .unwrap_or_else(|_| Utc::now());
510
511        let effective_health = if last_hb < timeout_cutoff && node.health == HealthStatus::Healthy {
512            HealthStatus::Unreachable
513        } else {
514            node.health.clone()
515        };
516
517        let storage_pct = if node.capacity.storage_total > 0 {
518            (node.capacity.storage_used as f64 / node.capacity.storage_total as f64) * 100.0
519        } else {
520            0.0
521        };
522
523        health_report.push(serde_json::json!({
524            "node_id": node.node_id,
525            "name": node.name,
526            "role": node.role.to_string(),
527            "health": effective_health.to_string(),
528            "region": node.region,
529            "storage_pct": format!("{:.1}%", storage_pct),
530            "cpu": format!("{:.1}%", node.capacity.cpu_utilization * 100.0),
531            "memory": format!("{:.1}%", node.capacity.memory_utilization * 100.0),
532            "connections": format!("{}/{}", node.capacity.active_connections, node.capacity.max_connections),
533            "last_heartbeat": node.last_heartbeat,
534        }));
535    }
536
537    Ok(DatacenterResponse {
538        action: "health".into(),
539        message: format!("Health check for {} node(s)", nodes.len()),
540        details: Some(serde_json::to_value(&health_report).unwrap_or_default()),
541    })
542}
543
544/// Collect and return Prometheus-style metrics
545pub fn execute_metrics() -> Result<DatacenterResponse, LitError> {
546    let repo_root = find_repo_root()?;
547    let metrics = collect_metrics(&repo_root);
548
549    // Also include per-node metrics if cluster is configured
550    let nodes = load_all_nodes(&repo_root)?;
551    let mut all_metrics = metrics;
552
553    all_metrics.push(Metric {
554        name: "lit_cluster_nodes_total".into(),
555        help: "Total nodes in cluster".into(),
556        metric_type: "gauge".into(),
557        value: nodes.len() as f64,
558        labels: HashMap::new(),
559    });
560
561    let healthy = nodes
562        .iter()
563        .filter(|n| n.health == HealthStatus::Healthy)
564        .count();
565    all_metrics.push(Metric {
566        name: "lit_cluster_nodes_healthy".into(),
567        help: "Healthy nodes in cluster".into(),
568        metric_type: "gauge".into(),
569        value: healthy as f64,
570        labels: HashMap::new(),
571    });
572
573    // Render as Prometheus exposition format
574    let exposition: String = all_metrics
575        .iter()
576        .map(|m| {
577            let labels_str = if m.labels.is_empty() {
578                String::new()
579            } else {
580                let pairs: Vec<String> = m
581                    .labels
582                    .iter()
583                    .map(|(k, v)| format!("{}=\"{}\"", k, v))
584                    .collect();
585                format!("{{{}}}", pairs.join(","))
586            };
587            format!(
588                "# HELP {} {}\n# TYPE {} {}\n{}{} {}",
589                m.name, m.help, m.name, m.metric_type, m.name, labels_str, m.value
590            )
591        })
592        .collect::<Vec<_>>()
593        .join("\n\n");
594
595    Ok(DatacenterResponse {
596        action: "metrics".into(),
597        message: format!("{} metric(s) collected", all_metrics.len()),
598        details: Some(serde_json::json!({
599            "metrics": all_metrics,
600            "exposition": exposition,
601        })),
602    })
603}
604
605/// Remove a node from the cluster (drain first recommended)
606pub fn execute_remove_node(node_id: String) -> Result<DatacenterResponse, LitError> {
607    let repo_root = find_repo_root()?;
608    let dir = nodes_dir(&repo_root);
609    let path = dir.join(format!("{}.json", node_id));
610
611    if !path.exists() {
612        return Err(LitError::general(format!("Node not found: {}", node_id)));
613    }
614
615    fs::remove_file(&path).map_err(|e| LitError::io(e.to_string()))?;
616
617    Ok(DatacenterResponse {
618        action: "remove-node".into(),
619        message: format!("Node '{}' removed from cluster", node_id),
620        details: None,
621    })
622}
lit/commands/datacenter.rs

lit/commands/
datacenter.rs
