chore(bors): merge pull request #1103

1103: feat(core): add OfflineRebuildReconciler skeleton (detection-only) r=Abhinandan-Purkait a=yugchaudhari

## Description

Starting on the offline volume rebuild (OEP 4208). This first PR adds the detection part, an `OfflineRebuildReconciler` that finds unpublished volumes in Degraded state and logs them. No rebuild happens yet.

Everything's behind `--offline-rebuild-enabled` (off by default), so this is a no-op unless you flip the flag.

## Motivation and Context

Unpublished volumes with degraded replicas currently stay degraded forever, the existing HotSpareReconciler only kicks in for published volumes (it gates on `target.is_some()`). This means if a node goes down while a volume is unpublished, replicas stay faulted until someone manually publishes and repairs.

OEP: https://github.com/openebs/openebs/blob/develop/designs/replicated-pv/mayastor/offline-volume-rebuild.md
Tracking: openebs/openebs#4208
Refs: openebs/mayastor#1977

## Testing

No BDD tests in this PR, there's no observable behavior to assert on yet (just a log line). Iteration 2 will add the actual rebuild and its tests will cover this detection path as a prerequisite.

Verified locally:
- `cargo check` passes in nix-shell
- Existing tests unaffected (feature is gated, reconciler is additive)

## Backward Compatibility

Fully backward compatible. The flag defaults to off, so existing deployments see zero change in behavior.

Co-authored-by: yugchaudhari <yug.chaudhari@kluisz.ai>

Author: mayastor-bors

control-plane/agents/src/bin/core/controller/reconciler/volume/mod.rs

@@ -1,12 +1,13 @@
 mod garbage_collector;
 mod hot_spare;
 mod nexus;
+mod offline_rebuild;
 
 use crate::controller::task_poller::{PollContext, PollPeriods, PollResult, PollTimer, TaskPoller};
 
 use crate::controller::reconciler::volume::{
     garbage_collector::GarbageCollector, hot_spare::HotSpareReconciler,
-    nexus::VolumeNexusReconciler,
+    nexus::VolumeNexusReconciler, offline_rebuild::OfflineRebuildReconciler,
 };
 
 /// Volume Reconciler loop which:
@@ -26,6 +27,7 @@ impl VolumeReconciler {
                 Box::new(HotSpareReconciler::new()),
                 Box::new(GarbageCollector::new()),
                 Box::new(VolumeNexusReconciler::new()),
+                Box::new(OfflineRebuildReconciler::new()),
             ],
         }
     }

control-plane/agents/src/bin/core/controller/reconciler/volume/offline_rebuild.rs

@@ -0,0 +1,72 @@
+use crate::controller::{
+    reconciler::{PollContext, TaskPoller},
+    resources::{operations_helper::OperationSequenceGuard, ResourceMutex},
+    task_poller::{PollResult, PollerState},
+};
+
+use stor_port::types::v0::{store::volume::VolumeSpec, transport::VolumeStatus};
+
+/// Offline Volume Rebuild reconciler.
+///
+/// Detects unpublished volumes whose replicas have become unhealthy and would
+/// benefit from a rebuild. Iteration 1 only logs eligible candidates; the
+/// actual rebuild action is added in a follow-up iteration.
+///
+/// See `designs/replicated-pv/mayastor/offline-volume-rebuild.md` for the
+/// full design.
+#[derive(Debug)]
+pub(super) struct OfflineRebuildReconciler {}
+
+impl OfflineRebuildReconciler {
+    pub(super) fn new() -> Self {
+        Self {}
+    }
+}
+
+#[async_trait::async_trait]
+impl TaskPoller for OfflineRebuildReconciler {
+    async fn poll(&mut self, context: &PollContext) -> PollResult {
+        if !context.registry().offline_rebuild_enabled() {
+            return PollResult::Ok(PollerState::Idle);
+        }
+
+        let mut results = vec![];
+        let volumes = context.specs().volumes_rsc();
+        for mut volume in volumes {
+            results.push(offline_rebuild_reconcile(&mut volume, context).await);
+        }
+        Self::squash_results(results)
+    }
+}
+
+#[tracing::instrument(
+    level = "debug",
+    skip(context, volume_spec),
+    fields(volume.uuid = %volume_spec.uuid(), request.reconcile = true)
+)]
+async fn offline_rebuild_reconcile(
+    volume_spec: &mut ResourceMutex<VolumeSpec>,
+    context: &PollContext,
+) -> PollResult {
+    let volume = match volume_spec.operation_guard() {
+        Ok(guard) => guard,
+        Err(_) => return PollResult::Ok(PollerState::Busy),
+    };
+
+    if !volume.as_ref().policy.self_heal
+        || !volume.as_ref().status.created()
+        || volume.as_ref().target().is_some()
+    {
+        return PollResult::Ok(PollerState::Idle);
+    }
+
+    let volume_state = context.registry().volume_state(volume.uuid()).await?;
+    if volume_state.status == VolumeStatus::Degraded {
+        tracing::debug!(
+            volume.uuid = %volume.uuid(),
+            "Unpublished volume is Degraded; eligible for offline rebuild"
+        );
+    }
+
+    PollResult::Ok(PollerState::Idle)
+}

control-plane/agents/src/bin/core/controller/registry.rs

@@ -125,6 +125,9 @@ pub(crate) struct RegistryInner<S: Store> {
     deprecated_access_mode: bool,
     /// Running in simulation mode.
     sim_args: Option<SimArgs>,
+    /// Enable the OfflineRebuildReconciler. Off by default until the
+    /// feature is fully tested and ready for general use.
+    offline_rebuild_enabled: bool,
 }
 
 impl Registry {
@@ -156,6 +159,7 @@ impl Registry {
         pool_cluster_size: Option<u32>,
         deprecated_access_mode: bool,
         sim_args: Option<SimArgs>,
+        offline_rebuild_enabled: bool,
     ) -> Result<Self, SvcError> {
         let store_endpoint = Self::format_store_endpoint(&store_url);
         tracing::info!("Connecting to persistent store at {}", store_endpoint);
@@ -220,6 +224,7 @@ impl Registry {
                 pool_cluster_size: pool_cluster_size.or(Some(POOL_BS_CLUSTER_SIZE_DEFAULT)),
                 deprecated_access_mode,
                 sim_args,
+                offline_rebuild_enabled,
             }),
         };
         registry.init().await?;
@@ -355,6 +360,10 @@ impl Registry {
     pub(crate) fn faulted_child_wait_period(&self) -> Option<std::time::Duration> {
         self.faulted_child_wait_period
     }
+    /// Whether the OfflineRebuildReconciler is enabled.
+    pub(crate) fn offline_rebuild_enabled(&self) -> bool {
+        self.offline_rebuild_enabled
+    }
     /// Allow for this given time before assuming failure and allowing the pool to get deleted.
     pub(crate) fn pool_async_creat_tmo(&self) -> std::time::Duration {
         self.pool_async_creat_tmo

control-plane/agents/src/bin/core/main.rs

@@ -47,6 +47,12 @@ pub(crate) struct CliArgs {
     #[clap(long)]
     pub(crate) faulted_child_wait_period: Option<humantime::Duration>,
 
+    /// Enable the OfflineRebuildReconciler, which detects unpublished
+    /// volumes with degraded replicas and logs them as rebuild candidates.
+    /// Disabled by default; rebuild action is not yet implemented.
+    #[clap(long, env = "OFFLINE_REBUILD_ENABLED")]
+    pub(crate) offline_rebuild_enabled: bool,
+
     /// When the pool creation gRPC times out, the actual call in the io-engine
     /// may still progress.
     /// We wait up to this period before considering the operation a failure and
@@ -288,6 +294,7 @@ async fn server(cli_args: CliArgs) -> anyhow::Result<()> {
         cli_args.pool_cluster_size,
         cli_args.deprecated_access_mode,
         sim_args,
+        cli_args.offline_rebuild_enabled,
     )
     .await?;