Merge #1879

1879: feat(persistence): update replica health condition as atomic operation r=dsharma-dc a=dsharma-dc

Fail the transaction operation and shutdown the nexus to avoid succeeding IO via remaining replicas. This ensures data integrity in case the replica being marked here has been picked up as source of truth elsewhere for the volume.

Co-authored-by: Diwakar Sharma <[email protected]>

Author: mayastor-bors

io-engine/src/bdev/nexus/mod.rs

@@ -48,7 +48,7 @@ pub use nexus_iter::{
 pub(crate) use nexus_module::{NexusModule, NEXUS_MODULE_NAME};
 pub(crate) use nexus_nbd::{NbdDisk, NbdError};
 pub(crate) use nexus_persistence::PersistOp;
-pub use nexus_persistence::{ChildInfo, NexusInfo};
+pub use nexus_persistence::{ChildInfo, NexusInfo, NexusInfoTxn, PersistentNexusInfo};
 pub(crate) use nexus_share::NexusPtpl;
 
 pub use nexus_bdev_snapshot::{

io-engine/src/bdev/nexus/nexus_bdev_error.rs

@@ -176,7 +176,7 @@ pub enum Error {
     InvalidReservation { reservation: u8 },
     #[snafu(display("failed to update share properties {}", name))]
     UpdateShareProperties { source: CoreError, name: String },
-    #[snafu(display("failed to save nexus state {}", name))]
+    #[snafu(display("failed to save nexus state {name}, {source}"))]
     SaveStateFailed { source: StoreError, name: String },
 }
 

io-engine/src/bdev/nexus/nexus_persistence.rs

@@ -1,5 +1,10 @@
 use super::{IoMode, Nexus, NexusChild};
-use crate::{persistent_store::PersistentStore, sleep::mayastor_sleep};
+use crate::{
+    persistent_store::{to_json_byte_vec, PersistentStore},
+    sleep::mayastor_sleep,
+    store::store_defs::StoreError,
+};
+use etcd_client::Error as EtcdErr;
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 
@@ -8,10 +13,10 @@ use super::Error;
 /// Information associated with the persisted NexusInfo structure.
 pub struct PersistentNexusInfo {
     /// Structure that is written to the persistent store.
-    inner: NexusInfo,
+    pub inner: NexusInfo,
     /// Key to use to persist the NexusInfo structure.
     /// If `Some` the key has been supplied by the control plane.
-    key: Option<String>,
+    pub key: Option<String>,
 }
 
 impl PersistentNexusInfo {
@@ -31,13 +36,20 @@ impl PersistentNexusInfo {
 
 /// Definition of the nexus information that gets saved in the persistent
 /// store.
-#[derive(Serialize, Deserialize, Debug, Default)]
+#[derive(Clone, Serialize, Deserialize, Debug, Default)]
 pub struct NexusInfo {
     /// Nexus destroyed successfully.
     pub clean_shutdown: bool,
+    /// Nexus needs to be shutdown.
+    pub do_self_shutdown: bool,
     /// Information about children.
     pub children: Vec<ChildInfo>,
 }
+pub struct NexusInfoTxn<'a> {
+    key_info: &'a mut PersistentNexusInfo,
+    // Expected value for the key.
+    expected: NexusInfo,
+}
 
 /// Definition of the child information that gets saved in the persistent
 /// store.
@@ -96,6 +108,7 @@ impl<'n> Nexus<'n> {
                 // expect the NexusInfo structure to contain default values.
                 assert!(nexus_info.children.is_empty());
                 assert!(!nexus_info.clean_shutdown);
+                assert!(!nexus_info.do_self_shutdown);
                 self.children_iter().for_each(|c| {
                     let child_info = ChildInfo {
                         uuid: NexusChild::uuid(c.uri()).expect("Failed to get child UUID."),
@@ -164,11 +177,40 @@ impl<'n> Nexus<'n> {
 
                 let uuid = NexusChild::uuid(child_uri).expect("Failed to get child UUID.");
 
+                let expected_value = nexus_info.clone();
                 nexus_info.children.iter_mut().for_each(|c| {
                     if c.uuid == uuid {
                         c.healthy = *healthy;
                     }
                 });
+
+                let mut txn = NexusInfoTxn {
+                    key_info: &mut persistent_nexus_info,
+                    expected: expected_value,
+                };
+
+                // Try executing the transaction. If the nexus info key's value isn't what we
+                // expect, shutdown this nexus. This can only happen if do_self_shutdown is
+                // found true in etcd for this key, which can only be set by core-agent's
+                // republish path. In this situation, the newly published nexus could've
+                // picked up the replica that we are trying to mark unhealthy here. Shutting
+                // down this nexus here ensures we don't let this IO succeed to other replicas
+                // after marking this one as unhealthy.
+                match self.save_txn(&mut txn).await {
+                    Ok(_) => {
+                        self.set_nexus_io_mode(IoMode::Normal).await;
+                        return Ok(());
+                    }
+                    Err(e) => {
+                        error!(
+                            "{self:?}: failed to update persistent store txn, \
+                            will shutdown the nexus: {e}"
+                        );
+                        self.try_self_shutdown();
+
+                        return Err(e);
+                    }
+                }
             }
             PersistOp::Shutdown => {
                 // Only update the clean shutdown variable. Do not update the
@@ -243,4 +285,82 @@ impl<'n> Nexus<'n> {
             }
         }
     }
+
+    async fn save_txn(&self, info: &mut NexusInfoTxn<'_>) -> Result<(), Error> {
+        // If a key has been provided, use it to store the NexusInfo; use the
+        // nexus uuid as the key otherwise.
+        let key = match &info.key_info.key {
+            Some(k) => k.clone(),
+            None => self.uuid().to_string(),
+        };
+
+        let mut retry = PersistentStore::retries();
+
+        let new_value = to_json_byte_vec(&info.key_info.inner);
+        let expected_value = to_json_byte_vec(&info.expected);
+        let mut logged = false;
+
+        loop {
+            match PersistentStore::txn_create_execute(&key, &new_value, &expected_value).await {
+                Ok(txn_resp) => {
+                    if let Some(current_value) = txn_resp {
+                        let val = serde_json::from_slice::<NexusInfo>(&current_value).unwrap();
+
+                        // The server had likely received the transaction and executed it, but
+                        // client here saw a timeout. So if the current value is same as what we intended
+                        // to set, then consider success. Don't trust any other value and shutdown.
+                        if current_value == new_value {
+                            info!("value for key {key} already updated: {val:?}");
+                            return Ok(());
+                        }
+
+                        warn!("current state found: key - {key}, value - {val:?}");
+
+                        // This nexus won't be used again but let's still update this field with what's
+                        // found in etcd.
+                        info.key_info.inner_mut().do_self_shutdown = val.do_self_shutdown;
+
+                        return Err(Error::SaveStateFailed {
+                            source: StoreError::Txn {
+                                key: key.clone(),
+                                source: EtcdErr::IoError(std::io::Error::new(
+                                    std::io::ErrorKind::Other,
+                                    "Txn CompareOp failed",
+                                )),
+                            },
+                            name: self.name.clone(),
+                        });
+                    } else {
+                        // Don't need to check individual op responses.
+                        debug!(?key, "{self:?}: the state was saved successfully via txn");
+                        return Ok(());
+                    }
+                }
+
+                Err(err) => {
+                    retry -= 1;
+                    if retry == 0 {
+                        return Err(Error::SaveStateFailed {
+                            source: err,
+                            name: self.name.clone(),
+                        });
+                    }
+
+                    if !logged {
+                        error!(
+                            "{self:?}: failed to persist nexus info transaction: {err}\
+                        will silently retry ({retry} left): {err}"
+                        );
+                        logged = true;
+                    }
+
+                    // Allow some time for the connection to the persistent
+                    // store to be re-established before retrying the operation.
+                    if mayastor_sleep(Duration::from_secs(1)).await.is_err() {
+                        error!("{self:?}: failed to wait for sleep");
+                    }
+                }
+            };
+        }
+    }
 }

io-engine/src/persistent_store.rs

@@ -5,19 +5,18 @@
 //! the etcd-client crate. This crate has a dependency on the tokio async
 //! runtime.
 use crate::{
-    core,
-    core::Reactor,
+    core::{self, Reactor},
     store::{
         etcd::Etcd,
         store_defs::{
             DeleteWait, GetWait, PutWait, Store, StoreError, StoreKey, StoreValue, TxnWait,
         },
     },
 };
-use etcd_client::{Compare, TxnOp, TxnResponse};
 use futures::channel::oneshot;
 use once_cell::sync::OnceCell;
 use parking_lot::Mutex;
+use serde::Serialize;
 use serde_json::Value;
 use snafu::ResultExt;
 use std::{future::Future, time::Duration};
@@ -196,18 +195,27 @@ impl PersistentStore {
         })?
     }
 
-    /// Executes a transaction for the given key.
-    pub async fn txn(
+    /// Executes a transaction with Put op for the given key.
+    /// On failure, Get the key value.
+    pub async fn txn_create_execute(
         key: &impl StoreKey,
-        cmps: Vec<Compare>,
-        ops_success: Vec<TxnOp>,
-        ops_failure: Option<Vec<TxnOp>>,
-    ) -> Result<TxnResponse, StoreError> {
+        new_value: &[u8],
+        expected_value: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
         let key_string = key.to_string();
+
+        info!(
+            "Executing transaction for key {}, value {}, expected value {}.",
+            key_string,
+            String::from_utf8_lossy(new_value),
+            String::from_utf8_lossy(expected_value)
+        );
+
+        let new_value_owned = new_value.to_owned();
+        let expected_value_owned = expected_value.to_owned();
         let rx = Self::execute_store_op(async move {
-            info!("Executing transaction for key {}.", key_string);
             Self::backing_store()
-                .txn_kv(&key_string, cmps, ops_success, ops_failure)
+                .put_kv_cas(&key_string, new_value_owned, expected_value_owned)
                 .await
         });
 
@@ -324,3 +332,8 @@ impl PersistentStore {
         persistent_store.lock().store = backing_store;
     }
 }
+
+pub fn to_json_byte_vec<T: Serialize>(input: &T) -> Vec<u8> {
+    let val = serde_json::to_value(input).expect("Failed conversion to serde_json value");
+    serde_json::to_vec(&val).expect("Failed conversion to serde_json bytes")
+}

io-engine/src/store/etcd.rs

@@ -5,7 +5,7 @@ use crate::store::store_defs::{
     StoreError::MissingEntry, StoreKey, StoreValue, Txn as TxnErr, ValueString,
 };
 use async_trait::async_trait;
-use etcd_client::{Client, Compare, Txn, TxnOp, TxnResponse};
+use etcd_client::{Client, Compare, CompareOp, Error, Txn, TxnOp, TxnOpResponse};
 use serde_json::Value;
 use snafu::ResultExt;
 
@@ -49,23 +49,50 @@ impl Store for Etcd {
         Ok(())
     }
 
-    /// Executes a transaction for the given key. If the compares succeed, then
-    /// ops_success will be executed atomically, otherwise ops_failure will be
-    /// executed atomically.
-    async fn txn_kv<K: StoreKey>(
+    /// Executes a transaction for the given key. If the compare succeed, then
+    /// new_value `Put` will be executed atomically, otherwise the current value
+    /// will be `Get` and returned.
+    async fn put_kv_cas<K: StoreKey>(
         &mut self,
         key: &K,
-        cmps: Vec<Compare>,
-        ops_success: Vec<TxnOp>,
-        ops_failure: Option<Vec<TxnOp>>,
-    ) -> Result<TxnResponse, StoreError> {
-        let fops = ops_failure.map_or(vec![], |v| v);
-        self.0
-            .txn(Txn::new().when(cmps).and_then(ops_success).or_else(fops))
+        new_value: Vec<u8>,
+        expected_value: Vec<u8>,
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        let cmp = Compare::value(key.to_string(), CompareOp::Equal, expected_value);
+        let put_op = TxnOp::put(key.to_string(), new_value, None);
+        let or_else_op = TxnOp::get(key.to_string(), None);
+
+        let txn_resp = self
+            .0
+            .txn(
+                Txn::new()
+                    .when([cmp])
+                    .and_then([put_op])
+                    .or_else([or_else_op]),
+            )
             .await
             .context(TxnErr {
                 key: key.to_string(),
-            })
+            })?;
+
+        if !txn_resp.succeeded() {
+            if let Some(TxnOpResponse::Get(g)) = &txn_resp.op_responses().first() {
+                if let Some(kv) = g.kvs().first() {
+                    let current_value = kv.value();
+                    return Ok(Some(current_value.to_owned()));
+                }
+            } else {
+                return Err(StoreError::Txn {
+                    key: key.to_string(),
+                    source: Error::IoError(std::io::Error::new(
+                        std::io::ErrorKind::Other,
+                        "Requested TxnOpResponse::Get not found",
+                    )),
+                });
+            }
+        }
+
+        Ok(None)
     }
 
     /// 'Get' the value for the given key from etcd.

io-engine/src/store/store_defs.rs

@@ -1,7 +1,7 @@
 //! Definition of a trait for a key-value store together with its error codes.
 
 use async_trait::async_trait;
-use etcd_client::{Compare, Error, TxnOp, TxnResponse};
+use etcd_client::Error;
 use serde_json::{Error as SerdeError, Value};
 use snafu::Snafu;
 
@@ -105,13 +105,13 @@ pub trait Store: Sync + Send + Clone {
         value: &V,
     ) -> Result<(), StoreError>;
 
-    async fn txn_kv<K: StoreKey>(
+    /// Put an entry by doing compare-and-swap with expected value.
+    async fn put_kv_cas<K: StoreKey>(
         &mut self,
         key: &K,
-        cmps: Vec<Compare>,
-        ops_success: Vec<TxnOp>,
-        ops_failure: Option<Vec<TxnOp>>,
-    ) -> Result<TxnResponse, StoreError>;
+        new_value: Vec<u8>,
+        expected_value: Vec<u8>,
+    ) -> Result<Option<Vec<u8>>, StoreError>;
 
     /// Get an entry from the store.
     async fn get_kv<K: StoreKey>(&mut self, key: &K) -> Result<Value, StoreError>;