feat: StatefulSet reconciliation improvements and status monitoring (#43) (#51)

* feat: Phase 1 - StatefulSet diff detection and validation (#43)

Implement intelligent StatefulSet update detection and immutable field
validation to improve reconciliation efficiency and safety.

Changes:
- Add statefulset_needs_update() method for semantic diff detection
- Add validate_statefulset_update() method for immutable field checks
- Refactor reconciliation loop to check/validate before updating
- Add new error types: InternalError, ImmutableFieldModified, SerdeJson
- Extend error policy with 60s requeue for immutable field errors
- Add 9 comprehensive unit tests (35 tests total, all passing)
- Update CHANGELOG.md with detailed changes

Benefits:
- Reduces unnecessary API calls and reconciliation overhead
- Prevents invalid updates that would cause API rejections
- Provides clear error messages for users
- Foundation for rollout monitoring in Phase 2

Related: #43

* feat: Phase 2 - Status types and StatefulSet helpers (#43)

Extend status structures and add StatefulSet status helper methods
for rollout monitoring support.

Changes:
- Add Condition struct for Kubernetes standard conditions (Ready, Progressing, Degraded)
- Extend Status struct with observed_generation and conditions fields
- Extend Pool status with replica tracking and revision fields
- Add new PoolState variants: Updating, RolloutComplete, RolloutFailed, Degraded
- Add context methods:
  - get_statefulset_status() - Fetch StatefulSet status
  - is_rollout_complete() - Check if rollout is complete
  - get_statefulset_revisions() - Get current and update revisions

Benefits:
- Foundation for comprehensive rollout monitoring
- Kubernetes-standard status conditions
- Per-pool rollout status tracking
- All existing tests continue to pass

Related: #43

* fix: Enable RUST_LOG environment variable support (#43)

The tracing subscriber was not respecting the RUST_LOG environment
variable, making it impossible to see debug logs from the operator.

Changes:
- Enable 'env-filter' feature for tracing-subscriber in Cargo.toml
- Add .with_env_filter() to tracing subscriber initialization
- Allows operators to see debug logs with: RUST_LOG=operator=debug

This fixes the "no logs" issue reported when running the operator.

Testing:
- Verified operator runs successfully with RUST_LOG=info
- Verified debug logs appear with RUST_LOG=operator=debug
- Confirmed diff detection and validation logs are visible

Related: #43

* fix: Block pool name changes to prevent orphaned StatefulSets (#43)

Changing a pool name creates a new StatefulSet but leaves the old one
orphaned. This is invalid because pool names are part of the StatefulSet
selector (immutable field).

Changes:
- Add validation before StatefulSet reconciliation loop
- List all StatefulSets owned by the Tenant
- Check for StatefulSets whose pool names don't match current spec
- Return ImmutableFieldModified error with clear guidance

Error message guides users to delete and recreate Tenant if rename needed.

Testing:
- All 35 tests passing
- Operator detects orphaned StatefulSets and returns error
- 60-second requeue for user to fix

Related: #43

* feat: Phase 3 - Status updates and rollout monitoring (#43)

Implement comprehensive status updates and rollout monitoring to track
StatefulSet reconciliation progress in real-time.

Changes:
- Add build_pool_status() helper method to Tenant to extract status from StatefulSets
- Collect pool statuses during StatefulSet reconciliation loop
- Aggregate pool statuses into overall Tenant conditions (Ready, Progressing, Degraded)
- Update Tenant status with replica counts, pool states, and conditions
- Requeue faster (10s) when pools are updating for responsive monitoring
- Refactor Context.update_status() to accept full Status struct
- Add chrono dependency for timestamp generation

Status Updates:
- Pool status includes: replicas, ready_replicas, current_replicas, updated_replicas,
  current_revision, update_revision, last_update_time, and state
- Pool states: NotCreated, Initialized, Updating, RolloutComplete, RolloutFailed, Degraded
- Tenant conditions: Ready (True/False), Progressing (True during rollout), Degraded (True when degraded)
- Tenant overall state: Ready, NotReady, Updating, Degraded

Rollout Monitoring:
- Tracks individual pool rollout status based on StatefulSet replicas and revisions
- Sets Progressing condition during rollouts
- Sets Ready condition when all replicas are ready and updated
- Sets Degraded condition when pools are unhealthy
- Requeues every 10 seconds during active rollouts for responsive updates

This completes Phase 3 of issue #43. Status information is now properly populated
and visible in kubectl/k9s.

* fix: Correct status update to send complete Tenant object (#43)

The replace_status API requires a complete Kubernetes object with
apiVersion, kind, metadata, spec, and status - not just the status field.

Changes:
- Clone the Tenant resource and set its status field
- Serialize the complete Tenant object
- Fixes 'Object Kind is missing' error during status updates

This fixes the status update errors seen in the operator logs.

* fix: Remove redundant Health column from kubectl output

The Health column was referencing a non-existent .status.healthStatus field.
Removed it since STATE column already shows the current state.

* fix: Use patch_status with JSON merge patch for status updates (#43)

Changed from replace_status with raw bytes to patch_status with JSON
merge patch, which is the proper kube-rs API for updating status
subresources.

This fixes the 'Object Kind is missing' error.

* fix: Resolve clippy lint warnings

- Use chained if-let conditions to collapse nested if statements
- Use as_deref() instead of as_ref().map(|s| s.as_str())
- Allow unwrap/expect in test modules (acceptable in tests)

All clippy checks now pass with -D warnings.

* fix: Resolve clippy lint warnings

Replace unwrap() and expect() calls with explicit pattern matching
in RBAC test code to satisfy clippy lints without using allow
annotations. All test behavior is preserved with explicit panic
messages on None values.

Author: shahab96

CHANGELOG.md

@@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+#### **StatefulSet Reconciliation Improvements** (2025-12-03, Issue #43)
+
+Implemented intelligent StatefulSet update detection and validation to improve reconciliation efficiency and safety:
+
+- **Diff Detection**: Added `statefulset_needs_update()` method to detect actual changes
+  - Compares existing vs desired StatefulSet specs semantically
+  - Avoids unnecessary API calls when no changes are needed
+  - Checks: replicas, image, env vars, resources, scheduling, pod management policy, etc.
+
+- **Immutable Field Validation**: Added `validate_statefulset_update()` method
+  - Prevents modifications to immutable StatefulSet fields (selector, volumeClaimTemplates, serviceName)
+  - Provides clear error messages for invalid updates (e.g., changing volumesPerServer)
+  - Protects against API rejections during reconciliation
+
+- **Enhanced Reconciliation Logic**: Refactored StatefulSet reconciliation loop
+  - Checks if StatefulSet exists before attempting update
+  - Validates update safety before applying changes
+  - Only applies updates when actual changes are detected
+  - Records Kubernetes events for update lifecycle (Created, UpdateStarted, UpdateValidationFailed)
+
+- **Error Handling**: Extended error policy
+  - Added 60-second requeue for immutable field modification errors (user-fixable)
+  - Consistent error handling across credential and validation failures
+
+- **New Error Types**: Added to `types::error::Error`
+  - `InternalError` - For unexpected internal conditions
+  - `ImmutableFieldModified` - For attempted modifications to immutable fields
+  - `SerdeJson` - For JSON serialization errors during comparisons
+
+- **Comprehensive Test Coverage**: Added 9 new unit tests (35 tests total)
+  - Tests for diff detection (no changes, image, replicas, env vars, resources)
+  - Tests for validation (selector, serviceName, volumesPerServer changes rejected)
+  - Test for safe updates (image changes allowed)
+
+**Benefits**:
+- Reduces unnecessary API calls and reconciliation overhead
+- Prevents reconciliation failures from invalid updates
+- Provides better error messages for users
+- Foundation for rollout monitoring (Phase 2)
+
 ### Changed
 
 #### **Code Refactoring**: Credential Validation Simplification (2025-11-15)

Cargo.lock

@@ -8,12 +8,13 @@ homepage = "https://rustfs.com"
 
 
 [dependencies]
+chrono = "0.4"
 const-str = "0.7.0"
 serde = { version = "1.0.228", features = ["derive"] }
 tokio = { version = "1.48.0", features = ["rt", "rt-multi-thread", "macros", "fs", "io-std", "io-util"] }
 futures = "0.3.31"
 tracing = "0.1.41"
-tracing-subscriber = { version = "0.3.20" }
+tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
 serde_json = "1.0.145"
 serde_yaml_ng = "0.10.0"
 strum = { version = "0.27.2", features = ["derive"] }

Cargo.toml

@@ -17,9 +17,6 @@ spec:
     - jsonPath: .status.currentState
       name: State
       type: string
-    - jsonPath: .status.healthStatus
-      name: Health
-      type: string
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
@@ -1157,15 +1154,87 @@ spec:
               availableReplicas:
                 format: int32
                 type: integer
+              conditions:
+                description: Kubernetes standard conditions
+                items:
+                  description: Kubernetes standard condition for Tenant resources
+                  properties:
+                    lastTransitionTime:
+                      description: Last time the condition transitioned from one status to another
+                      nullable: true
+                      type: string
+                    message:
+                      description: Human-readable message indicating details about the transition
+                      type: string
+                    observedGeneration:
+                      description: The generation of the Tenant resource that this condition reflects
+                      format: int64
+                      nullable: true
+                      type: integer
+                    reason:
+                      description: One-word CamelCase reason for the condition's last transition
+                      type: string
+                    status:
+                      description: Status of the condition (True, False, Unknown)
+                      type: string
+                    type:
+                      description: Type of condition (Ready, Progressing, Degraded)
+                      type: string
+                  required:
+                  - message
+                  - reason
+                  - status
+                  - type
+                  type: object
+                type: array
               currentState:
                 type: string
+              observedGeneration:
+                description: The generation observed by the operator
+                format: int64
+                nullable: true
+                type: integer
               pools:
                 items:
                   properties:
+                    currentReplicas:
+                      description: Number of pods with current revision
+                      format: int32
+                      nullable: true
+                      type: integer
+                    currentRevision:
+                      description: Current revision hash of the StatefulSet
+                      nullable: true
+                      type: string
+                    lastUpdateTime:
+                      description: Last time the pool status was updated
+                      nullable: true
+                      type: string
+                    readyReplicas:
+                      description: Number of pods with Ready condition
+                      format: int32
+                      nullable: true
+                      type: integer
+                    replicas:
+                      description: Total number of non-terminated pods targeted by this pool's StatefulSet
+                      format: int32
+                      nullable: true
+                      type: integer
                     ssName:
+                      description: Name of the StatefulSet for this pool
                       type: string
                     state:
+                      description: Current state of the pool
+                      type: string
+                    updateRevision:
+                      description: Update revision hash of the StatefulSet (different from current during rollout)
+                      nullable: true
                       type: string
+                    updatedReplicas:
+                      description: Number of pods with updated revision
+                      format: int32
+                      nullable: true
+                      type: integer
                   required:
                   - ssName
                   - state

deploy/rustfs-operator/crds/tenant.yaml

@@ -106,38 +106,41 @@ impl Context {
             .await
     }
 
-    pub async fn update_status<S>(
+    pub async fn update_status(
         &self,
         resource: &Tenant,
-        current_status: S,
-        replica: i32,
-    ) -> Result<Tenant, Error>
-    where
-        S: ToString,
-    {
+        status: crate::types::v1alpha1::status::Status,
+    ) -> Result<Tenant, Error> {
+        use kube::api::{Patch, PatchParams};
+
         let api: Api<Tenant> = Api::namespaced(self.client.clone(), &resource.namespace()?);
-        let name = &resource.name();
+        let name = resource.name();
 
-        let update_func = async |tenant: &Tenant| {
-            let mut status = tenant.status.clone().unwrap_or_default();
-            status.available_replicas = replica;
-            status.current_state = current_status.to_string();
-            let status_body = serde_json::to_vec(&status)?;
+        // Create a JSON merge patch for the status
+        let status_patch = serde_json::json!({
+            "status": status
+        });
 
-            api.replace_status(name, &PostParams::default(), status_body)
-                .context(KubeSnafu)
-                .await
-        };
-
-        match update_func(resource).await {
+        // Try to patch the status
+        match api
+            .patch_status(
+                &name,
+                &PatchParams::default(),
+                &Patch::Merge(status_patch.clone()),
+            )
+            .context(KubeSnafu)
+            .await
+        {
             Ok(t) => return Ok(t),
             _ => {}
         }
 
         info!("status update failed due to conflict, retrieve the latest resource and retry.");
 
-        let new_one = api.get(name).context(KubeSnafu).await?;
-        update_func(&new_one).await
+        // Retry with the same patch
+        api.patch_status(&name, &PatchParams::default(), &Patch::Merge(status_patch))
+            .context(KubeSnafu)
+            .await
     }
 
     pub async fn delete<T>(&self, name: &str, namespace: &str) -> Result<(), Error>
@@ -287,4 +290,85 @@ impl Context {
 
         Ok(())
     }
+
+    /// Gets the status of a StatefulSet including rollout progress
+    ///
+    /// # Returns
+    /// The StatefulSet status with replica counts and revision information
+    pub async fn get_statefulset_status(
+        &self,
+        name: &str,
+        namespace: &str,
+    ) -> Result<k8s_openapi::api::apps::v1::StatefulSetStatus, Error> {
+        let ss: k8s_openapi::api::apps::v1::StatefulSet = self.get(name, namespace).await?;
+
+        ss.status.ok_or_else(|| Error::Types {
+            source: types::error::Error::InternalError {
+                msg: format!("StatefulSet {} has no status", name),
+            },
+        })
+    }
+
+    /// Checks if a StatefulSet rollout is complete
+    ///
+    /// A rollout is considered complete when:
+    /// - observedGeneration matches metadata.generation (controller has seen latest spec)
+    /// - replicas == readyReplicas (all pods are ready)
+    /// - currentRevision == updateRevision (all pods are on the new revision)
+    /// - updatedReplicas == replicas (all pods have been updated)
+    ///
+    /// # Returns
+    /// - `Ok(true)` if rollout is complete
+    /// - `Ok(false)` if rollout is still in progress
+    /// - `Err` if there's an error fetching the StatefulSet
+    pub async fn is_rollout_complete(&self, name: &str, namespace: &str) -> Result<bool, Error> {
+        let ss: k8s_openapi::api::apps::v1::StatefulSet = self.get(name, namespace).await?;
+
+        let metadata = &ss.metadata;
+        let spec = ss.spec.as_ref().ok_or_else(|| Error::Types {
+            source: types::error::Error::InternalError {
+                msg: format!("StatefulSet {} missing spec", name),
+            },
+        })?;
+
+        let status = ss.status.as_ref().ok_or_else(|| Error::Types {
+            source: types::error::Error::InternalError {
+                msg: format!("StatefulSet {} missing status", name),
+            },
+        })?;
+
+        let desired_replicas = spec.replicas.unwrap_or(1);
+
+        // Check if controller has observed the latest generation
+        let generation_current = metadata.generation.is_some()
+            && status.observed_generation.is_some()
+            && metadata.generation == status.observed_generation;
+
+        // Check if all replicas are ready
+        let replicas_ready = status.replicas == desired_replicas
+            && status.ready_replicas.unwrap_or(0) == desired_replicas
+            && status.updated_replicas.unwrap_or(0) == desired_replicas;
+
+        // Check if all pods are on the same revision
+        let revisions_match = status.current_revision.is_some()
+            && status.update_revision.is_some()
+            && status.current_revision == status.update_revision;
+
+        Ok(generation_current && replicas_ready && revisions_match)
+    }
+
+    /// Gets the current and update revision of a StatefulSet
+    ///
+    /// # Returns
+    /// A tuple of (current_revision, update_revision)
+    /// Returns None for either value if not available
+    pub async fn get_statefulset_revisions(
+        &self,
+        name: &str,
+        namespace: &str,
+    ) -> Result<(Option<String>, Option<String>), Error> {
+        let status = self.get_statefulset_status(name, namespace).await?;
+
+        Ok((status.current_revision, status.update_revision))
+    }
 }

src/context.rs

@@ -38,6 +38,7 @@ pub mod tests;
 
 pub async fn run() -> Result<(), Box<dyn std::error::Error>> {
     tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .with_level(true)
         .with_file(true)
         .with_line_number(true)