Add additional debugging context when json deserialization fails (ope…

…nzfs#378)

Wrap `serde_json::from_slice()` with a helper function that will add
additional context when it fails, including the type we were trying to
decode it as, and the actual JSON string.

cmd/zfs_object_agent/util/Cargo.toml

@@ -34,6 +34,7 @@ rand = "0.8.3"
 rusoto_credential = "0.47.0"
 safer-ffi = { version = "0.0.7", features = ["proc_macros"] }
 serde = { version = "1.0.125", features = ["derive"] }
+serde_json = "1.0.64"
 serial_test = "0.6.0"
 signal-hook = "0.3.13"
 tokio = { version = "1.4", features = ["full"] }

cmd/zfs_object_agent/util/src/lib.rs

@@ -20,6 +20,7 @@ pub mod message;
 mod mutex_ext;
 mod nicenum;
 mod range_tree;
+pub mod serde;
 pub mod tunable;
 pub mod unordered;
 mod vec_ext;

cmd/zfs_object_agent/util/src/serde.rs

@@ -0,0 +1,13 @@
+use std::any::type_name;
+
+use anyhow::Context;
+use anyhow::Result;
+use serde::de::DeserializeOwned;
+
+pub fn from_json_slice<T: DeserializeOwned>(slice: &[u8]) -> Result<T> {
+    let t = type_name::<T>();
+    serde_json::from_slice(slice).with_context(|| match std::str::from_utf8(slice) {
+        Ok(utf) => format!("could not decode JSON as {t}: {utf}"),
+        Err(e) => format!("could not decode JSON as {t}: [non-UTF slice: {e}]",),
+    })
+}

cmd/zfs_object_agent/zettacache/src/block_access.rs

@@ -30,6 +30,7 @@ use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use util::iter_wrapping;
 use util::measure;
+use util::serde::from_json_slice;
 use util::tunable;
 use util::with_alloctag;
 use util::zettacache_stats::*;
@@ -686,7 +687,7 @@ impl BlockAccess {
         // Note, the NUL byte is not included in either slice
         let (header_slice, post_header_slice) = split2(buf, |&c| c == b'\0')
             .ok_or_else(|| anyhow!("nul byte not found in {}-byte buf", buf.len()))?;
-        let header: BlockHeader = serde_json::from_slice(header_slice)
+        let header: BlockHeader = from_json_slice(header_slice)
             .with_context(|| format!("{}-byte BlockHeader", header_slice.len()))?;
 
         if header.payload_size > post_header_slice.len() {
@@ -721,7 +722,7 @@ impl BlockAccess {
         };
 
         let struct_obj: T = match header.encoding {
-            EncodeType::Json => serde_json::from_slice(serde_slice)?,
+            EncodeType::Json => from_json_slice(serde_slice)?,
             EncodeType::Bincode => Self::bincode_options().deserialize(serde_slice)?,
             EncodeType::BincodeFixint => Self::bincode_fixint_options().deserialize(serde_slice)?,
         };

cmd/zfs_object_agent/zettacache/src/slab_allocator.rs

@@ -293,12 +293,15 @@ impl SlabAllocator {
         target_free_slabs.saturating_sub(current_free_slabs)
     }
 
-    /// This should be called after the checkpoint has been persisted to disk, at which point
-    /// it's safe to re-allocate freed slabs.
-    pub fn checkpoint_done(&self) {
-        let mut inner = self.inner.lock().unwrap();
-        let inner_ref = &mut *inner; // avoid 2 exclusive borrows from the MutexGuard
-        inner_ref.allocatable.append(&mut inner_ref.freeing);
+    /// Release the space held by freed slabs, allowing them to be re-allocated.  This is safe to
+    /// call after the checkpoint has been persisted to disk.
+    pub fn release_frees(&self) {
+        let mut guard = self.inner.lock().unwrap();
+        let inner = &mut *guard;
+        // By using the `&mut Inner` directly, the borrow checker can understand the
+        // `inner.allocatable` and `inner.freeing` below as "split borrows", allowing two
+        // exclusive references to different fields of the same struct.
+        inner.allocatable.append(&mut inner.freeing);
         inner.allocatable.shuffle(&mut thread_rng());
     }
 

cmd/zfs_object_agent/zettacache/src/zettacache/mod.rs

@@ -2152,7 +2152,7 @@ impl ZettaCacheState {
                 old_index_size + (self.pending_changes_cap * size_of::<IndexEntry>()) as u64,
             ),
         );
-        self.slab_allocator.checkpoint_done();
+        self.slab_allocator.release_frees();
 
         info!(
             "completed {:?} in {}ms; flushed {} operations ({}) to log",

cmd/zfs_object_agent/zettaobject/src/heartbeat.rs

@@ -18,6 +18,7 @@ use serde::Serialize;
 use tokio::sync::watch;
 use tokio::sync::watch::Receiver;
 use util::maybe_die_with;
+use util::serde::from_json_slice;
 use util::tunable;
 use uuid::Uuid;
 
@@ -53,7 +54,7 @@ impl HeartbeatPhys {
         let buf = object_access
             .get_object(Self::key(id), ObjectAccessOpType::MetadataGet)
             .await?;
-        let this: Self = serde_json::from_slice(&buf)
+        let this: Self = from_json_slice(&buf)
             .with_context(|| format!("Failed to decode contents of {}", Self::key(id)))?;
         debug!("got {:#?}", this);
         assert_eq!(this.id, id);

cmd/zfs_object_agent/zettaobject/src/object_based_log.rs

@@ -16,6 +16,7 @@ use serde::Deserialize;
 use serde::Serialize;
 use tokio::task::JoinHandle;
 use util::measure;
+use util::serde::from_json_slice;
 use util::tunable;
 use zettacache::base_types::*;
 
@@ -102,7 +103,7 @@ impl<T: ObjectBasedLogEntry> ObjectBasedLogChunk<T> {
             )
             .await?;
         let begin = Instant::now();
-        let this: Self = serde_json::from_slice(&buf).with_context(|| {
+        let this: Self = from_json_slice(&buf).with_context(|| {
             format!(
                 "Failed to decode contents of {}",
                 Self::key(name, generation, chunk)

cmd/zfs_object_agent/zettaobject/src/pool.rs

@@ -46,6 +46,7 @@ use tokio::time::sleep;
 use util::async_cache::GetMethod;
 use util::maybe_die_with;
 use util::measure;
+use util::serde::from_json_slice;
 use util::super_trace;
 use util::tunable;
 use util::tunable::Percent;
@@ -151,7 +152,7 @@ impl PoolOwnerPhys {
         let buf = object_access
             .get_object(Self::key(id), ObjectAccessOpType::MetadataGet)
             .await?;
-        let this: Self = serde_json::from_slice(&buf)
+        let this: Self = from_json_slice(&buf)
             .with_context(|| format!("Failed to decode contents of {}", Self::key(id)))?;
         debug!("got {:#?}", this);
         assert_eq!(this.id, id);
@@ -321,7 +322,7 @@ impl PoolPhys {
         let buf = object_access
             .get_object(Self::key(guid), ObjectAccessOpType::MetadataGet)
             .await?;
-        let this: Self = serde_json::from_slice(&buf)
+        let this: Self = from_json_slice(&buf)
             .with_context(|| format!("Failed to decode contents of {}", Self::key(guid)))?;
         debug!("got {:#?}", this);
         assert_eq!(this.guid, guid);
@@ -382,7 +383,7 @@ impl UberblockPhys {
         let buf = object_access
             .get_object(Self::key(guid, txg), ObjectAccessOpType::MetadataGet)
             .await?;
-        let this: Self = serde_json::from_slice(&buf)
+        let this: Self = from_json_slice(&buf)
             .with_context(|| format!("Failed to decode contents of {}", Self::key(guid, txg)))?;
         debug!("got {:#?}", this);
         assert_eq!(this.guid, guid);