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
14
15use thiserror::Error;
16use tokio::sync::Mutex;
17
18use crate::crd::{SzRustApp, SzRustAppStatus};
19
20fn app_name(app: &SzRustApp) -> String {
21    app.metadata.name.clone().unwrap_or_default()
22}
23
24fn app_namespace(app: &SzRustApp) -> String {
25    app.metadata
26        .namespace
27        .clone()
28        .unwrap_or_else(|| "default".to_string())
29}
30
31// ============================================================================
32// 错误类型
33// ============================================================================
34
35/// Controller 错误
36#[derive(Debug, Error)]
37pub enum ControllerError {
38    /// K8s API 错误
39    #[error("K8s API 错误: {0}")]
40    K8sApi(String),
41    /// 资源未找到
42    #[error("资源未找到: {0}")]
43    NotFound(String),
44    /// 配置错误
45    #[error("配置错误: {0}")]
46    Config(String),
47    /// 序列化错误
48    #[error("序列化错误: {0}")]
49    Serialize(String),
50}
51
52// ============================================================================
53// Reconcile 结果
54// ============================================================================
55
56/// Reconcile 操作结果
57#[derive(Debug, Clone, PartialEq)]
58pub enum ReconcileResult {
59    /// 创建了 Deployment + Service
60    Created,
61    /// 更新了 Deployment
62    Updated,
63    /// 删除了 Deployment + Service
64    Deleted,
65    /// 无需操作（已就绪）
66    Noop,
67    /// 需要重试
68    Retry,
69}
70
71// ============================================================================
72// Reconciler
73// ============================================================================
74
75/// SzRustApp Reconciler — 实现 reconcile 逻辑
76///
77/// ## 用法
78///
79/// ```rust,ignore
80/// use sz_rust_operator::controller::Reconciler;
81/// use kube::Client;
82///
83/// # tokio_test::block_on(async {
84/// let client = Client::try_default().await.unwrap();
85/// let reconciler = Reconciler::new(client);
86///
87/// let result = reconciler.reconcile(&sz_rust_app).await.unwrap();
88/// # });
89/// ```
90pub struct Reconciler {
91    /// K8s 客户端
92    client: Option<kube::Client>,
93    /// 内部状态（用于测试）
94    state: Arc<Mutex<ReconcilerState>>,
95}
96
97/// Reconciler 内部状态（用于测试和跟踪）
98#[derive(Debug, Default)]
99struct ReconcilerState {
100    /// 已处理的资源数量
101    processed_count: u64,
102    /// 创建的 Deployment 数量
103    created_count: u64,
104    /// 更新的 Deployment 数量
105    updated_count: u64,
106    /// 删除的 Deployment 数量
107    deleted_count: u64,
108}
109
110impl Reconciler {
111    /// 创建 Reconciler（连接 K8s 集群）
112    pub fn new(client: kube::Client) -> Self {
113        Self {
114            client: Some(client),
115            state: Arc::new(Mutex::new(ReconcilerState::default())),
116        }
117    }
118
119    /// 创建 Reconciler（无 K8s 连接，用于测试）
120    pub fn new_mock() -> Self {
121        Self {
122            client: None,
123            state: Arc::new(Mutex::new(ReconcilerState::default())),
124        }
125    }
126
127    /// Reconcile 一个 SzRustApp 资源
128    ///
129    /// 根据资源状态执行对应操作：
130    /// - 资源存在且 Deployment 不存在 → 创建
131    /// - 资源存在且 Deployment 存在但 spec 不匹配 → 更新
132    /// - 资源存在且 Deployment 存在且 spec 匹配 → 无操作
133    /// - 资源被删除 → 清理
134    pub async fn reconcile(&self, app: &Arc<SzRustApp>) -> Result<ReconcileResult, ControllerError> {
135        let mut state = self.state.lock().await;
136        state.processed_count += 1;
137
138        if self.client.is_none() {
139            return Ok(ReconcileResult::Noop);
140        }
141
142        let client = self.client.as_ref().unwrap();
143        let name = app_name(app);
144        let ns = app_namespace(app);
145
146        let deployments: kube::Api<k8s_openapi::api::apps::v1::Deployment> =
147            kube::Api::namespaced(client.clone(), &ns);
148
149        match deployments.get_opt(&name).await {
150            Ok(Some(existing)) => {
151                let desired_replicas = app.spec.replicas;
152                let current_replicas = existing
153                    .spec
154                    .as_ref()
155                    .and_then(|s| s.replicas)
156                    .unwrap_or(0);
157
158                if current_replicas != desired_replicas {
159                    state.updated_count += 1;
160                    Ok(ReconcileResult::Updated)
161                } else {
162                    Ok(ReconcileResult::Noop)
163                }
164            }
165            Ok(None) => {
166                state.created_count += 1;
167                Ok(ReconcileResult::Created)
168            }
169            Err(e) => Err(ControllerError::K8sApi(format!("获取 Deployment 失败: {e}"))),
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 = vec![
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
