diff --git a/.gitignore b/.gitignore index a16bb8c..1bfcd50 100644 --- a/.gitignore +++ b/.gitignore @@ -14,5 +14,8 @@ bazel-* Thumbs.db ._* +# Rust +target/ + # Logs *.log diff --git a/kq/loader/mod.rs b/kq/loader/mod.rs index 3b3c88f..dbf3f53 100644 --- a/kq/loader/mod.rs +++ b/kq/loader/mod.rs @@ -33,6 +33,7 @@ pub enum LoadingPhase { ConvertingPods, ConvertingNamespaces, ConvertingDaemonSets, + ConvertingDeployments, Finalizing, } @@ -45,6 +46,7 @@ impl LoadingPhase { LoadingPhase::ConvertingPods => "Converting pods", LoadingPhase::ConvertingNamespaces => "Converting namespaces", LoadingPhase::ConvertingDaemonSets => "Converting daemonsets", + LoadingPhase::ConvertingDeployments => "Converting deployments", LoadingPhase::Finalizing => "Finalizing", } } @@ -492,6 +494,7 @@ impl SnapshotLoader { pods: None, namespaces: None, daemon_sets: None, + deployments: None, }, tables: final_tables, table_batches, diff --git a/kq/loader/progress_reporter.rs b/kq/loader/progress_reporter.rs index c983cad..1bff2d3 100644 --- a/kq/loader/progress_reporter.rs +++ b/kq/loader/progress_reporter.rs @@ -81,10 +81,11 @@ impl ProgressReporter for TerminalProgressReporter { let icon = match phase { LoadingPhase::ReadingFile => "📖", LoadingPhase::ParsingJSON => "🔍", - LoadingPhase::ConvertingNodes | + LoadingPhase::ConvertingNodes | LoadingPhase::ConvertingPods | LoadingPhase::ConvertingNamespaces | - LoadingPhase::ConvertingDaemonSets => "⚡", + LoadingPhase::ConvertingDaemonSets | + LoadingPhase::ConvertingDeployments => "⚡", LoadingPhase::Finalizing => "✨", }; bar.set_message(format!("{} {}", icon, message)); diff --git a/kq/loader/resource_table.rs b/kq/loader/resource_table.rs index 7f8f7c7..0faa248 100644 --- a/kq/loader/resource_table.rs +++ b/kq/loader/resource_table.rs @@ -50,6 +50,14 @@ pub(crate) const RESOURCE_TABLES: &[ResourceTable] = &[ parquet_file: "daemonsets.parquet", schema: NestedSchemas::daemon_sets_nested_schema, }, + ResourceTable { + table_name: "deployments", + json_key: "deployments", + ndjson_file: "deployments.ndjson.gz", + ipc_file: "deployments.arrow", + parquet_file: "deployments.parquet", + schema: NestedSchemas::deployments_nested_schema, + }, ]; pub(crate) fn resource_for_json_key(json_key: &str) -> Option { diff --git a/kq/query/mod.rs b/kq/query/mod.rs index 3b6ed78..bae6d8c 100644 --- a/kq/query/mod.rs +++ b/kq/query/mod.rs @@ -346,6 +346,7 @@ impl QueryEngine { "nodes" => "Kubernetes nodes with capacity, allocatable resources, and conditions".to_string(), "namespaces" => "Kubernetes namespaces with metadata and status".to_string(), "daemon_sets" => "Kubernetes DaemonSets with replica status information".to_string(), + "deployments" => "Kubernetes Deployments with replica status information".to_string(), _ => format!("Kubernetes resource table: {}", table_name), } } @@ -573,6 +574,7 @@ mod tests { namespaces: Some(vec![]), daemon_sets: Some(vec![]), pods: Some(vec![]), + deployments: Some(vec![]), }, tables: HashMap::new(), table_batches: HashMap::new(), @@ -673,6 +675,7 @@ mod tests { "status": { "phase": "Active" } }], "daemonSets": [], + "deployments": [], "pods": pods }); @@ -937,7 +940,7 @@ mod tests { .iter() .map(|table| table.name.clone()) .collect(); - for expected in ["pods", "nodes", "namespaces", "daemon_sets"] { + for expected in ["pods", "nodes", "namespaces", "daemon_sets", "deployments"] { assert!( names.contains(expected), "missing user-facing view '{expected}'; got {names:?}" diff --git a/kq/schema/BUILD.bazel b/kq/schema/BUILD.bazel index 1ca6657..52310cc 100644 --- a/kq/schema/BUILD.bazel +++ b/kq/schema/BUILD.bazel @@ -29,6 +29,7 @@ rust_test( "//kq/tests:golden_node.json", "//kq/tests:golden_namespace.json", "//kq/tests:golden_daemonset.json", + "//kq/tests:golden_deployment.json", ], deps = [ "@crates//:arrow-json", diff --git a/kq/schema/kubernetes.rs b/kq/schema/kubernetes.rs index 88d093b..f4244d5 100644 --- a/kq/schema/kubernetes.rs +++ b/kq/schema/kubernetes.rs @@ -78,6 +78,15 @@ pub use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ // LabelSelectorRequirement, }; +// Apps v1 - Deployment +pub use k8s_openapi::api::apps::v1::{ + Deployment, + DeploymentSpec, + DeploymentStatus, + DeploymentCondition, + DeploymentStrategy, +}; + // Note: ResourceList is now a BTreeMap // We'll keep a type alias for compatibility with our converters pub use k8s_openapi::apimachinery::pkg::api::resource::Quantity; @@ -97,6 +106,8 @@ pub struct ClusterSnapshot { #[serde(rename = "daemonSets", default)] pub daemon_sets: Option>, pub pods: Option>, + #[serde(default)] + pub deployments: Option>, } impl ClusterSnapshot { @@ -107,12 +118,14 @@ impl ClusterSnapshot { let namespace_count = self.namespaces.as_ref().map_or(0, |n| n.len()); let daemonset_count = self.daemon_sets.as_ref().map_or(0, |d| d.len()); let pod_count = self.pods.as_ref().map(|p| p.len()); + let deployment_count = self.deployments.as_ref().map_or(0, |d| d.len()); SnapshotSummary { timestamp: ts, node_count, namespace_count, daemonset_count, pod_count, + deployment_count, } } } @@ -124,6 +137,7 @@ pub struct SnapshotSummary { pub namespace_count: usize, pub daemonset_count: usize, pub pod_count: Option, + pub deployment_count: usize, } #[cfg(test)] @@ -138,7 +152,8 @@ mod tests { "nodes": [], "namespaces": [], "daemonSets": [], - "pods": [] + "pods": [], + "deployments": [] } "#; @@ -147,6 +162,7 @@ mod tests { assert_eq!(snapshot.namespaces.as_ref().map_or(0, |n| n.len()), 0); assert_eq!(snapshot.daemon_sets.as_ref().map_or(0, |d| d.len()), 0); assert!(snapshot.pods.is_some()); + assert_eq!(snapshot.deployments.as_ref().map_or(0, |d| d.len()), 0); } #[test] @@ -157,11 +173,13 @@ mod tests { namespaces: Some(vec![]), daemon_sets: Some(vec![]), pods: Some(vec![]), + deployments: Some(vec![]), }; let summary = snapshot.get_summary(); assert_eq!(summary.node_count, 0); assert_eq!(summary.pod_count, Some(0)); + assert_eq!(summary.deployment_count, 0); } #[test] @@ -171,6 +189,7 @@ mod tests { let _pod: Option = None; let _namespace: Option = None; let _daemonset: Option = None; + let _deployment: Option = None; // This test ensures the types compile assert!(true); diff --git a/kq/schema/mod.rs b/kq/schema/mod.rs index 37fb91f..67bdacb 100644 --- a/kq/schema/mod.rs +++ b/kq/schema/mod.rs @@ -39,6 +39,7 @@ impl SchemaRegistry { self.schemas.insert("nodes".to_string(), Self::nodes_schema()); self.schemas.insert("namespaces".to_string(), Self::namespaces_schema()); self.schemas.insert("daemon_sets".to_string(), Self::daemon_sets_schema()); + self.schemas.insert("deployments".to_string(), Self::deployments_schema()); } pub fn get_schema(&self, table_name: &str) -> Option> { @@ -209,6 +210,28 @@ impl SchemaRegistry { Arc::new(Schema::new(fields)) } + /// Schema for deployments table + pub fn deployments_schema() -> Arc { + let fields = vec![ + // All fields nullable for robustness + Field::new("name", DataType::Utf8, true), + Field::new("namespace", DataType::Utf8, true), + Field::new("uid", DataType::Utf8, true), + Field::new("creation_timestamp", DataType::Timestamp(TimeUnit::Millisecond, None), true), + Field::new("labels", DataType::Utf8, true), // JSON string for now + Field::new("replicas", DataType::Int32, true), + Field::new("ready_replicas", DataType::Int32, true), + Field::new("available_replicas", DataType::Int32, true), + Field::new("updated_replicas", DataType::Int32, true), + Field::new("unavailable_replicas", DataType::Int32, true), + + // Hidden JSON column - contains full object JSON + Field::new(".json", DataType::Utf8, true), + ]; + + Arc::new(Schema::new(fields)) + } + } impl Default for SchemaRegistry { @@ -226,11 +249,12 @@ mod tests { let registry = SchemaRegistry::new(); let tables = registry.list_tables(); - assert_eq!(tables.len(), 4); + assert_eq!(tables.len(), 5); assert!(tables.contains(&"pods".to_string())); assert!(tables.contains(&"nodes".to_string())); assert!(tables.contains(&"namespaces".to_string())); assert!(tables.contains(&"daemon_sets".to_string())); + assert!(tables.contains(&"deployments".to_string())); } #[test] @@ -283,6 +307,7 @@ mod tests { assert!(tables.contains(&"nodes".to_string())); assert!(tables.contains(&"namespaces".to_string())); assert!(tables.contains(&"daemon_sets".to_string())); + assert!(tables.contains(&"deployments".to_string())); } #[test] diff --git a/kq/schema/nested.rs b/kq/schema/nested.rs index d36d094..45175ac 100644 --- a/kq/schema/nested.rs +++ b/kq/schema/nested.rs @@ -563,6 +563,135 @@ impl NestedSchemas { Arc::new(Schema::new(fields)) } + + /// Enhanced deployments schema with nested structure + pub fn deployments_nested_schema() -> Arc { + let fields = vec![ + // Full metadata struct + Field::new("metadata", DataType::Struct( + vec![ + Field::new("name", DataType::Utf8, true), + Field::new("namespace", DataType::Utf8, true), + Field::new("uid", DataType::Utf8, true), + Field::new("creationTimestamp", DataType::Timestamp(TimeUnit::Millisecond, None), true), + Field::new("labels", DataType::Map( + Arc::new(Field::new("entries", DataType::Struct( + vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, true), + ].into() + ), false)), + false + ), true), + Field::new("annotations", DataType::Map( + Arc::new(Field::new("entries", DataType::Struct( + vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, true), + ].into() + ), false)), + false + ), true), + ].into() + ), true), + + // DeploymentSpec + Field::new("spec", DataType::Struct( + vec![ + Field::new("replicas", DataType::Int32, true), + Field::new("minReadySeconds", DataType::Int32, true), + Field::new("progressDeadlineSeconds", DataType::Int32, true), + Field::new("revisionHistoryLimit", DataType::Int32, true), + Field::new("paused", DataType::Boolean, true), + // Label selector + Field::new("selector", DataType::Struct(vec![ + Field::new("matchLabels", DataType::Map( + Arc::new(Field::new("entries", DataType::Struct( + vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, true), + ].into() + ), false)), + false + ), true), + Field::new("matchExpressions", DataType::List( + Arc::new(Field::new("item", DataType::Struct(vec![ + Field::new("key", DataType::Utf8, true), + Field::new("operator", DataType::Utf8, true), + Field::new("values", DataType::List( + Arc::new(Field::new("item", DataType::Utf8, false)) + ), true), + ].into()), true)) + ), true), + ].into()), true), + // Strategy + Field::new("strategy", DataType::Struct(vec![ + Field::new("type", DataType::Utf8, true), + Field::new("rollingUpdate", DataType::Struct(vec![ + Field::new("maxUnavailable", DataType::Utf8, true), + Field::new("maxSurge", DataType::Utf8, true), + ].into()), true), + ].into()), true), + // Pod template — just containers for image/resource queries; + // full pod details live on the pods table (join on ownerReferences). + Field::new("template", DataType::Struct(vec![ + Field::new("spec", DataType::Struct(vec![ + Field::new("containers", DataType::List( + Arc::new(Field::new("item", DataType::Struct(vec![ + Field::new("name", DataType::Utf8, true), + Field::new("image", DataType::Utf8, true), + Field::new("resources", DataType::Struct(vec![ + Field::new("requests", DataType::Map( + Arc::new(Field::new("entries", DataType::Struct(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, true), + ].into()), false)), + false, + ), true), + Field::new("limits", DataType::Map( + Arc::new(Field::new("entries", DataType::Struct(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, true), + ].into()), false)), + false, + ), true), + ].into()), true), + ].into()), true)) + ), true), + ].into()), true), + ].into()), true), + ].into() + ), true), + + // DeploymentStatus + Field::new("status", DataType::Struct( + vec![ + Field::new("observedGeneration", DataType::Int64, true), + Field::new("replicas", DataType::Int32, true), + Field::new("updatedReplicas", DataType::Int32, true), + Field::new("readyReplicas", DataType::Int32, true), + Field::new("availableReplicas", DataType::Int32, true), + Field::new("unavailableReplicas", DataType::Int32, true), + // Conditions + Field::new("conditions", DataType::List( + Arc::new(Field::new("item", DataType::Struct(vec![ + Field::new("type", DataType::Utf8, true), + Field::new("status", DataType::Utf8, true), + Field::new("reason", DataType::Utf8, true), + Field::new("message", DataType::Utf8, true), + Field::new("lastTransitionTime", DataType::Timestamp(TimeUnit::Millisecond, None), true), + Field::new("lastUpdateTime", DataType::Timestamp(TimeUnit::Millisecond, None), true), + ].into()), true)) + ), true), + ].into() + ), true), + + // Derived + Field::new("ready_percentage", DataType::Float32, true), + ]; + + Arc::new(Schema::new(fields)) + } } #[cfg(test)] @@ -686,4 +815,26 @@ mod tests { assert!(batch_count > 0, "Should have parsed at least one batch"); } + + /// Test parsing a synthetic deployment JSON. + #[test] + fn test_parse_golden_deployment() { + let schema = NestedSchemas::deployments_nested_schema(); + let golden_file = File::open("kq/tests/golden_deployment.json") + .expect("Golden file should exist"); + let buf_reader = BufReader::new(golden_file); + + let json_reader = ReaderBuilder::new(schema.clone()) + .build(buf_reader) + .expect("Should build JSON reader"); + + let mut batch_count = 0; + for batch_result in json_reader { + let batch = batch_result.expect("Should parse batch successfully"); + assert!(batch.num_rows() > 0, "Batch should have rows"); + batch_count += 1; + } + + assert!(batch_count > 0, "Should have parsed at least one batch"); + } } diff --git a/kq/synthetic/mod.rs b/kq/synthetic/mod.rs index 525fded..41a9731 100644 --- a/kq/synthetic/mod.rs +++ b/kq/synthetic/mod.rs @@ -30,6 +30,19 @@ const DAEMONSET_WORKLOADS: &[(&str, &str, &str)] = &[ ("dns-node-cache", "kube-system", "dns"), ]; +const DEPLOYMENT_WORKLOADS: &[(&str, &str, i32, &str)] = &[ + ("frontend", "production", 8, "web"), + ("api-gateway", "production", 5, "gateway"), + ("user-service", "production", 3, "services"), + ("payment-worker", "production", 4, "workers"), + ("search-indexer", "production", 2, "workers"), + ("notification-controller", "production", 3, "controllers"), + ("cache-warmer", "production", 2, "workers"), + ("data-processor", "production", 6, "workers"), + ("auth-proxy", "production", 3, "infra"), + ("config-controller", "production", 2, "controllers"), +]; + const POOLS: &[(&str, u32)] = &[ ("general", 58), ("cpu", 12), @@ -112,6 +125,7 @@ pub struct SyntheticSnapshotSummary { pub pod_count: usize, pub namespace_count: usize, pub daemonset_count: usize, + pub deployment_count: usize, pub min_pods_per_node: usize, pub max_pods_per_node: usize, pub running_pods: usize, @@ -237,6 +251,7 @@ pub fn generate_ndjson_snapshot(config: &SyntheticSnapshotConfig) -> Result Result Result<()> { finish_gzip_writer(writer) } +fn write_deployments(config: &SyntheticSnapshotConfig) -> Result<()> { + let path = config.output_dir.join("deployments.ndjson.gz"); + let mut writer = gzip_line_writer(&path)?; + for (idx, (name, namespace, replicas, component)) in DEPLOYMENT_WORKLOADS.iter().enumerate() { + let ready = if idx % 7 == 0 { replicas - 1 } else { *replicas }; + let available = ready; + let updated = replicas; + let deployment = json!({ + "metadata": { + "name": name, + "namespace": namespace, + "uid": stable_uid(&config.cluster_name, "deployment", idx), + "creationTimestamp": format_timestamp(config.timestamp - Duration::days(90 - (idx % 30) as i64)), + "labels": { + "app": name, + "app.kubernetes.io/name": name, + "app.kubernetes.io/component": component, + "app.kubernetes.io/part-of": "production-services", + "synthetic.kq.dev/cluster": config.cluster_name, + "workload.kq.dev/kind": "Deployment" + }, + "annotations": { + "deployment.kubernetes.io/revision": format!("{}", 10 + idx), + "prometheus.io/scrape": "true", + "prometheus.io/port": "9090" + } + }, + "spec": { + "replicas": replicas, + "revisionHistoryLimit": 10, + "selector": { + "matchLabels": { + "app": name + } + }, + "strategy": { + "type": "RollingUpdate", + "rollingUpdate": { + "maxUnavailable": "25%", + "maxSurge": "25%" + } + }, + "template": { + "spec": { + "containers": [{ + "name": name, + "image": format!("registry.kq.dev/production/{}:1.{}.{}", name, idx, replicas), + "resources": { + "requests": { + "cpu": "200m", + "memory": "256Mi" + }, + "limits": { + "cpu": "500m", + "memory": "512Mi" + } + } + }] + } + } + }, + "status": { + "observedGeneration": 1, + "replicas": replicas, + "updatedReplicas": updated, + "readyReplicas": ready, + "availableReplicas": available, + "unavailableReplicas": replicas - available, + "conditions": [ + { + "type": "Available", + "status": if available > 0 { "True" } else { "False" }, + "reason": "MinimumReplicasAvailable", + "message": "Deployment has minimum availability.", + "lastTransitionTime": format_timestamp(config.timestamp - Duration::hours((idx % 24) as i64)), + "lastUpdateTime": format_timestamp(config.timestamp - Duration::minutes((idx % 60) as i64)) + }, + { + "type": "Progressing", + "status": if updated == replicas { "True" } else { "False" }, + "reason": "NewReplicaSetAvailable", + "message": "ReplicaSet has been successfully progressed.", + "lastTransitionTime": format_timestamp(config.timestamp - Duration::hours((idx % 12) as i64)), + "lastUpdateTime": format_timestamp(config.timestamp - Duration::minutes((idx % 30) as i64)) + } + ] + }, + "ready_percentage": if *replicas > 0 { (ready as f32 / *replicas as f32) * 100.0 } else { 0.0 } + }); + write_json_line(&mut writer, &deployment)?; + } + finish_gzip_writer(writer) +} + fn make_node_context(config: &SyntheticSnapshotConfig, rng: &mut SimpleRng, index: usize) -> NodeContext { let pool = rng.weighted(POOLS).to_string(); let zone = rng.pick(ZONES).to_string(); @@ -1367,6 +1478,7 @@ mod tests { assert!(config.output_dir.join("pods.ndjson.gz").exists()); assert!(config.output_dir.join("namespaces.ndjson.gz").exists()); assert!(config.output_dir.join("daemonsets.ndjson.gz").exists()); + assert!(config.output_dir.join("deployments.ndjson.gz").exists()); } #[test] @@ -1429,6 +1541,7 @@ mod tests { for file in [ "daemonsets.ndjson.gz", + "deployments.ndjson.gz", "namespaces.ndjson.gz", "nodes.ndjson.gz", "pods.ndjson.gz", diff --git a/kq/tests/BUILD.bazel b/kq/tests/BUILD.bazel index a88b79f..045679b 100644 --- a/kq/tests/BUILD.bazel +++ b/kq/tests/BUILD.bazel @@ -6,6 +6,7 @@ exports_files([ "golden_node.json", "golden_namespace.json", "golden_daemonset.json", + "golden_deployment.json", ]) # Test targets diff --git a/kq/tests/golden_deployment.json b/kq/tests/golden_deployment.json new file mode 100644 index 0000000..f02fc84 --- /dev/null +++ b/kq/tests/golden_deployment.json @@ -0,0 +1,79 @@ +{ + "metadata": { + "name": "frontend", + "namespace": "production", + "uid": "deployment-uid-1", + "creationTimestamp": "2024-01-01T12:00:00Z", + "labels": { + "app": "frontend", + "app.kubernetes.io/name": "frontend", + "app.kubernetes.io/component": "web" + }, + "annotations": { + "deployment.kubernetes.io/revision": "10" + } + }, + "spec": { + "replicas": 3, + "revisionHistoryLimit": 10, + "selector": { + "matchLabels": { + "app": "frontend" + } + }, + "strategy": { + "type": "RollingUpdate", + "rollingUpdate": { + "maxUnavailable": "25%", + "maxSurge": "25%" + } + }, + "template": { + "spec": { + "containers": [ + { + "name": "frontend", + "image": "registry.kq.dev/production/frontend:1.0.3", + "resources": { + "requests": { + "cpu": "200m", + "memory": "256Mi" + }, + "limits": { + "cpu": "500m", + "memory": "512Mi" + } + } + } + ] + } + } + }, + "status": { + "observedGeneration": 1, + "replicas": 3, + "updatedReplicas": 3, + "readyReplicas": 3, + "availableReplicas": 3, + "unavailableReplicas": 0, + "conditions": [ + { + "type": "Available", + "status": "True", + "reason": "MinimumReplicasAvailable", + "message": "Deployment has minimum availability.", + "lastTransitionTime": "2024-01-01T12:00:00Z", + "lastUpdateTime": "2024-01-01T12:00:00Z" + }, + { + "type": "Progressing", + "status": "True", + "reason": "NewReplicaSetAvailable", + "message": "ReplicaSet has been successfully progressed.", + "lastTransitionTime": "2024-01-01T12:00:00Z", + "lastUpdateTime": "2024-01-01T12:00:00Z" + } + ] + }, + "ready_percentage": 100.0 +}