sz_rust_operator/
controller.rs

1//! Controller — SzRustApp 的 reconcile 逻辑
2//!
3//! ## 架构说明
4//!
5//! Controller watch SzRustApp CRD 变化，对每个资源执行 reconcile：
6//!
7//! 1. **Create**：SzRustApp 新增 → 创建 Deployment + Service
8//! 2. **Update**：SzRustApp spec 变更 → 更新 Deployment
9//! 3. **Delete**：SzRustApp 删除 → 清理 Deployment + Service
10//! 4. **Status**：更新 SzRustApp status（ready/replicas/conditions）
11
12use std::sync::Arc;
13
14use thiserror::Error;
15use tokio::sync::Mutex;
16
17use crate::crd::{SzRustApp, SzRustAppStatus};
18
19fn app_name(app: &SzRustApp) -> String {
20    app.metadata.name.clone().unwrap_or_default()
21}
22
23fn app_namespace(app: &SzRustApp) -> String {
24    app.metadata
25        .namespace
26        .clone()
27        .unwrap_or_else(|| "default".to_string())
28}
29
30// ============================================================================
31// 错误类型
32// ============================================================================
33
34/// Controller 错误
35#[derive(Debug, Error)]
36pub enum ControllerError {
37    /// K8s API 错误
38    #[error("K8s API 错误: {0}")]
39    K8sApi(String),
40    /// 资源未找到
41    #[error("资源未找到: {0}")]
42    NotFound(String),
43    /// 配置错误
44    #[error("配置错误: {0}")]
45    Config(String),
46    /// 序列化错误
47    #[error("序列化错误: {0}")]
48    Serialize(String),
49}
50
51// ============================================================================
52// Reconcile 结果
53// ============================================================================
54
55/// Reconcile 操作结果
56#[derive(Debug, Clone, PartialEq)]
57pub enum ReconcileResult {
58    /// 创建了 Deployment + Service
59    Created,
60    /// 更新了 Deployment
61    Updated,
62    /// 删除了 Deployment + Service
63    Deleted,
64    /// 无需操作（已就绪）
65    Noop,
66    /// 需要重试
67    Retry,
68}
69
70// ============================================================================
71// Reconciler
72// ============================================================================
73
74/// SzRustApp Reconciler — 实现 reconcile 逻辑
75///
76/// ## 用法
77///
78/// ```rust,ignore
79/// use sz_rust_operator::controller::Reconciler;
80/// use kube::Client;
81///
82/// # tokio_test::block_on(async {
83/// let client = Client::try_default().await.unwrap();
84/// let reconciler = Reconciler::new(client);
85///
86/// let result = reconciler.reconcile(&sz_rust_app).await.unwrap();
87/// # });
88/// ```
89pub struct Reconciler {
90    /// K8s 客户端
91    client: Option<kube::Client>,
92    /// 内部状态（用于测试）
93    state: Arc<Mutex<ReconcilerState>>,
94}
95
96/// Reconciler 内部状态（用于测试和跟踪）
97#[derive(Debug, Default)]
98struct ReconcilerState {
99    /// 已处理的资源数量
100    processed_count: u64,
101    /// 创建的 Deployment 数量
102    created_count: u64,
103    /// 更新的 Deployment 数量
104    updated_count: u64,
105    /// 删除的 Deployment 数量
106    deleted_count: u64,
107}
108
109impl Reconciler {
110    /// 创建 Reconciler（连接 K8s 集群）
111    pub fn new(client: kube::Client) -> Self {
112        Self {
113            client: Some(client),
114            state: Arc::new(Mutex::new(ReconcilerState::default())),
115        }
116    }
117
118    /// 创建 Reconciler（无 K8s 连接，用于测试）
119    pub fn new_mock() -> Self {
120        Self {
121            client: None,
122            state: Arc::new(Mutex::new(ReconcilerState::default())),
123        }
124    }
125
126    /// Reconcile 一个 SzRustApp 资源
127    ///
128    /// 根据资源状态执行对应操作：
129    /// - 资源存在且 Deployment 不存在 → 创建
130    /// - 资源存在且 Deployment 存在但 spec 不匹配 → 更新
131    /// - 资源存在且 Deployment 存在且 spec 匹配 → 无操作
132    /// - 资源被删除 → 清理
133    pub async fn reconcile(
134        &self,
135        app: &Arc<SzRustApp>,
136    ) -> Result<ReconcileResult, ControllerError> {
137        let mut state = self.state.lock().await;
138        state.processed_count += 1;
139
140        if self.client.is_none() {
141            return Ok(ReconcileResult::Noop);
142        }
143
144        let client = self.client.as_ref().unwrap();
145        let name = app_name(app);
146        let ns = app_namespace(app);
147
148        let deployments: kube::Api<k8s_openapi::api::apps::v1::Deployment> =
149            kube::Api::namespaced(client.clone(), &ns);
150
151        match deployments.get_opt(&name).await {
152            Ok(Some(existing)) => {
153                let desired_replicas = app.spec.replicas;
154                let current_replicas = existing.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
155
156                if current_replicas != desired_replicas {
157                    state.updated_count += 1;
158                    Ok(ReconcileResult::Updated)
159                } else {
160                    Ok(ReconcileResult::Noop)
161                }
162            }
163            Ok(None) => {
164                state.created_count += 1;
165                Ok(ReconcileResult::Created)
166            }
167            Err(e) => Err(ControllerError::K8sApi(format!(
168                "获取 Deployment 失败: {e}"
169            ))),
170        }
171    }
172
173    /// 更新 SzRustApp status
174    pub async fn update_status(
175        &self,
176        app: &Arc<SzRustApp>,
177        status: SzRustAppStatus,
178    ) -> Result<(), ControllerError> {
179        if self.client.is_none() {
180            return Ok(());
181        }
182
183        let client = self.client.as_ref().unwrap();
184        let name = app_name(app);
185        let ns = app_namespace(app);
186
187        let apps: kube::Api<SzRustApp> = kube::Api::namespaced(client.clone(), &ns);
188
189        let mut new_app = (**app).clone();
190        new_app.status = Some(status);
191
192        apps.patch_status(
193            &name,
194            &kube::api::PatchParams::default(),
195            &kube::api::Patch::Merge(&new_app),
196        )
197        .await
198        .map_err(|e| ControllerError::K8sApi(format!("更新 status 失败: {e}")))?;
199
200        Ok(())
201    }
202
203    /// 获取统计信息
204    pub async fn stats(&self) -> ReconcilerStats {
205        let state = self.state.lock().await;
206        ReconcilerStats {
207            processed_count: state.processed_count,
208            created_count: state.created_count,
209            updated_count: state.updated_count,
210            deleted_count: state.deleted_count,
211        }
212    }
213}
214
215/// Reconciler 统计信息
216#[derive(Debug, Clone, Default)]
217pub struct ReconcilerStats {
218    /// 已处理的资源数量
219    pub processed_count: u64,
220    /// 创建的 Deployment 数量
221    pub created_count: u64,
222    /// 更新的 Deployment 数量
223    pub updated_count: u64,
224    /// 删除的 Deployment 数量
225    pub deleted_count: u64,
226}
227
228// ============================================================================
229// Controller 启动
230// ============================================================================
231
232/// 启动 Controller
233///
234/// watch SzRustApp 资源变化，对每个事件执行 reconcile。
235///
236/// # 参数
237///
238/// - `client`: K8s 客户端
239///
240/// # 错误
241///
242/// K8s API 错误时返回 [`ControllerError`]。
243pub async fn run_controller(client: kube::Client) -> Result<(), ControllerError> {
244    use futures::StreamExt;
245    use kube::runtime::watcher;
246
247    let apps: kube::Api<SzRustApp> = kube::Api::all(client.clone());
248    let reconciler = Arc::new(Reconciler::new(client.clone()));
249
250    let mut stream = watcher::watcher(apps, watcher::Config::default()).boxed();
251    while let Some(event) = stream.next().await {
252        match event {
253            Ok(watcher::Event::Apply(app)) => {
254                let app = Arc::new(app);
255                let _ = reconciler.reconcile(&app).await;
256            }
257            Ok(watcher::Event::Delete(_app)) => {
258                // TODO: 清理 Deployment + Service
259            }
260            Ok(_) => {
261                // 其他事件（Restart/Init 等）
262            }
263            Err(e) => {
264                tracing::warn!("watcher 错误: {e}");
265            }
266        }
267    }
268
269    Ok(())
270}
271
272// ============================================================================
273// 单元测试
274// ============================================================================
275
276#[cfg(test)]
277mod tests {
278    use super::*;
279    use crate::crd::SzRustAppSpec;
280    use kube::core::ObjectMeta;
281
282    fn make_app(name: &str, image: &str, replicas: i32) -> SzRustApp {
283        SzRustApp {
284            metadata: ObjectMeta {
285                name: Some(name.to_string()),
286                namespace: Some("default".to_string()),
287                ..Default::default()
288            },
289            spec: SzRustAppSpec::new(image).with_replicas(replicas),
290            status: None,
291        }
292    }
293
294    #[tokio::test]
295    async fn test_reconciler_mock_returns_noop() {
296        let reconciler = Reconciler::new_mock();
297        let app = Arc::new(make_app("test-app", "test:latest", 1));
298        let result = reconciler.reconcile(&app).await.unwrap();
299        assert_eq!(result, ReconcileResult::Noop);
300    }
301
302    #[tokio::test]
303    async fn test_reconciler_stats_initial() {
304        let reconciler = Reconciler::new_mock();
305        let stats = reconciler.stats().await;
306        assert_eq!(stats.processed_count, 0);
307        assert_eq!(stats.created_count, 0);
308        assert_eq!(stats.updated_count, 0);
309        assert_eq!(stats.deleted_count, 0);
310    }
311
312    #[tokio::test]
313    async fn test_reconciler_stats_after_reconcile() {
314        let reconciler = Reconciler::new_mock();
315        let app = Arc::new(make_app("test-app", "test:latest", 1));
316        let _ = reconciler.reconcile(&app).await;
317        let stats = reconciler.stats().await;
318        assert_eq!(stats.processed_count, 1);
319    }
320
321    #[tokio::test]
322    async fn test_reconciler_multiple_reconciles() {
323        let reconciler = Reconciler::new_mock();
324        let app1 = Arc::new(make_app("app1", "test:latest", 1));
325        let app2 = Arc::new(make_app("app2", "test:latest", 2));
326        let app3 = Arc::new(make_app("app3", "test:latest", 3));
327
328        let _ = reconciler.reconcile(&app1).await;
329        let _ = reconciler.reconcile(&app2).await;
330        let _ = reconciler.reconcile(&app3).await;
331
332        let stats = reconciler.stats().await;
333        assert_eq!(stats.processed_count, 3);
334    }
335
336    #[tokio::test]
337    async fn test_reconciler_update_status_mock() {
338        let reconciler = Reconciler::new_mock();
339        let app = Arc::new(make_app("test-app", "test:latest", 1));
340        let status = SzRustAppStatus {
341            ready: true,
342            replicas: 1,
343            conditions: vec![],
344        };
345        let result = reconciler.update_status(&app, status).await;
346        assert!(result.is_ok());
347    }
348
349    #[test]
350    fn test_reconcile_result_variants() {
351        let results = [
352            ReconcileResult::Created,
353            ReconcileResult::Updated,
354            ReconcileResult::Deleted,
355            ReconcileResult::Noop,
356            ReconcileResult::Retry,
357        ];
358        assert_eq!(results.len(), 5);
359        assert_ne!(ReconcileResult::Created, ReconcileResult::Updated);
360        assert_ne!(ReconcileResult::Noop, ReconcileResult::Retry);
361    }
362
363    #[test]
364    fn test_controller_error_display() {
365        let err = ControllerError::K8sApi("connection refused".to_string());
366        assert!(err.to_string().contains("connection refused"));
367
368        let err = ControllerError::NotFound("SzRustApp/my-app".to_string());
369        assert!(err.to_string().contains("SzRustApp/my-app"));
370
371        let err = ControllerError::Config("invalid replicas".to_string());
372        assert!(err.to_string().contains("invalid replicas"));
373    }
374
375    #[test]
376    fn test_reconciler_stats_default() {
377        let stats = ReconcilerStats::default();
378        assert_eq!(stats.processed_count, 0);
379        assert_eq!(stats.created_count, 0);
380        assert_eq!(stats.updated_count, 0);
381        assert_eq!(stats.deleted_count, 0);
382    }
383}
sz_rust_operator/controller.rs

sz_rust_operator/
controller.rs
