DOSE-475 Zpool iostat (openzfs#494)

cmd/zfs_object_agent/client/src/main.rs

@@ -24,8 +24,8 @@ use std::time::{Duration, Instant};
 use tokio::io::AsyncReadExt;
 use zettacache::base_types::*;
 use zettaobject::base_types::*;
-use zettaobject::ObjectAccess;
 use zettaobject::Pool;
+use zettaobject::{ObjectAccess, ObjectAccessStatType};
 mod client;
 
 const ENDPOINT: &str = "https://s3-us-west-2.amazonaws.com";
@@ -409,7 +409,9 @@ async fn do_test_connectivity(object_access: &ObjectAccess) {
     let file = format!("test/test_connectivity_{}", num);
     let content = "test connectivity to S3".as_bytes().to_vec();
 
-    object_access.put_object(file.clone(), content).await;
+    object_access
+        .put_object(file.clone(), content, ObjectAccessStatType::MetadataPut)
+        .await;
     object_access.delete_object(file).await;
 }
 

cmd/zfs_object_agent/object_perf/src/s3perf.rs

@@ -11,7 +11,7 @@ use std::error::Error;
 use std::string::String;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use zettaobject::ObjectAccess;
+use zettaobject::{ObjectAccess, ObjectAccessStatType};
 
 enum WriteTestBounds {
     Time(Duration),
@@ -30,15 +30,20 @@ impl Perf {
     #[measure(Throughput)]
     #[measure(HitCount)]
     async fn put(&self, object_access: &ObjectAccess, key: String, data: Vec<u8>) {
-        object_access.put_object(key, data).await;
+        object_access
+            .put_object(key, data, ObjectAccessStatType::MetadataPut)
+            .await;
     }
 
     #[measure(type = ResponseTime<AtomicHdrHistogram, StdInstantMicros>)]
     #[measure(InFlight)]
     #[measure(Throughput)]
     #[measure(HitCount)]
     async fn get(&self, object_access: &ObjectAccess, key: String) {
-        object_access.get_object(key).await.unwrap();
+        object_access
+            .get_object(key, ObjectAccessStatType::MetadataGet)
+            .await
+            .unwrap();
     }
 
     fn log_metrics(&self, duration: Duration) {

cmd/zfs_object_agent/zettaobject/Cargo.toml

@@ -15,6 +15,7 @@ bytes = "1.0"
 chrono = "0.4"
 conv = "0.3.3"
 cstr-argument = "0.1.1"
+enum-map = "1.1.1"
 fs2 = "0.4.2"
 futures = "0.3.13"
 futures-core = "0.3.13"

cmd/zfs_object_agent/zettaobject/src/data_object.rs

@@ -1,4 +1,5 @@
-use crate::{base_types::*, ObjectAccess};
+use crate::base_types::*;
+use crate::object_access::{ObjectAccess, ObjectAccessStatType};
 use anyhow::{Context, Result};
 use log::*;
 use more_asserts::*;
@@ -100,15 +101,20 @@ impl DataObjectPhys {
         object_access: &ObjectAccess,
         guid: PoolGuid,
         object: ObjectId,
+        stat_type: ObjectAccessStatType,
         bypass_cache: bool,
     ) -> Result<Self> {
         let buf = match bypass_cache {
             true => {
                 object_access
-                    .get_object_uncached(Self::key(guid, object))
+                    .get_object_uncached(Self::key(guid, object), stat_type)
+                    .await?
+            }
+            false => {
+                object_access
+                    .get_object(Self::key(guid, object), stat_type)
                     .await?
             }
-            false => object_access.get_object(Self::key(guid, object)).await?,
         };
         let begin = Instant::now();
         let this: DataObjectPhys = bincode::deserialize(&buf)
@@ -129,11 +135,16 @@ impl DataObjectPhys {
     pub async fn get_from_key(
         object_access: &ObjectAccess,
         key: String,
+        stat_type: ObjectAccessStatType,
         bypass_cache: bool,
     ) -> Result<Self> {
         let buf = match bypass_cache {
-            true => object_access.get_object_uncached(key.clone()).await?,
-            false => object_access.get_object(key.clone()).await?,
+            true => {
+                object_access
+                    .get_object_uncached(key.clone(), stat_type)
+                    .await?
+            }
+            false => object_access.get_object(key.clone(), stat_type).await?,
         };
         let begin = Instant::now();
         let this: DataObjectPhys = bincode::deserialize(&buf)
@@ -149,7 +160,7 @@ impl DataObjectPhys {
         Ok(this)
     }
 
-    pub async fn put(&self, object_access: &ObjectAccess) {
+    pub async fn put(&self, object_access: &ObjectAccess, stat_type: ObjectAccessStatType) {
         let begin = Instant::now();
         let contents = bincode::serialize(self).unwrap();
         trace!(
@@ -161,7 +172,7 @@ impl DataObjectPhys {
         );
         self.verify();
         object_access
-            .put_object(Self::key(self.guid, self.object), contents)
+            .put_object(Self::key(self.guid, self.object), contents, stat_type)
             .await;
     }
 

cmd/zfs_object_agent/zettaobject/src/heartbeat.rs

@@ -1,4 +1,4 @@
-use crate::object_access::{OAError, ObjectAccess};
+use crate::object_access::{OAError, ObjectAccess, ObjectAccessStatType};
 use anyhow::Context;
 use lazy_static::lazy_static;
 use log::{debug, info, trace, warn};
@@ -36,7 +36,9 @@ impl HeartbeatPhys {
     }
 
     pub async fn get(object_access: &ObjectAccess, id: Uuid) -> anyhow::Result<Self> {
-        let buf = object_access.get_object_impl(Self::key(id), None).await?;
+        let buf = object_access
+            .get_object_impl(Self::key(id), ObjectAccessStatType::MetadataGet, None)
+            .await?;
         let this: Self = serde_json::from_slice(&buf)
             .with_context(|| format!("Failed to decode contents of {}", Self::key(id)))?;
         debug!("got {:#?}", this);
@@ -53,7 +55,12 @@ impl HeartbeatPhys {
         trace!("putting {:#?}", self);
         let buf = serde_json::to_vec(&self).unwrap();
         object_access
-            .put_object_timed(Self::key(self.id), buf, timeout)
+            .put_object_timed(
+                Self::key(self.id),
+                buf,
+                ObjectAccessStatType::MetadataPut,
+                timeout,
+            )
             .await
     }
 

cmd/zfs_object_agent/zettaobject/src/lib.rs

@@ -8,6 +8,7 @@ mod data_object;
 mod features;
 mod heartbeat;
 pub mod init;
+pub use object_access::{OAError, ObjectAccessStatType};
 mod object_access;
 mod object_based_log;
 mod object_block_map;