Make quantity field human readable

nativelink-config/BUILD.bazel

@@ -18,6 +18,8 @@ rust_library(
     deps = [
         "@crates//:serde",
         "@crates//:shellexpand",
+        "@crates//:humantime",
+        "@crates//:byte-unit"
     ],
 )
 

nativelink-config/Cargo.toml

@@ -4,5 +4,7 @@ version = "0.3.0"
 edition = "2021"
 
 [dependencies]
+byte-unit = "5.1.4"
+humantime = "2.1.0"
 serde = { version = "1.0.198", features = ["derive"] }
 shellexpand = "3.1.0"

nativelink-config/src/cas_server.rs

@@ -18,6 +18,7 @@ use serde::Deserialize;
 
 use crate::schedulers::SchedulerConfig;
 use crate::serde_utils::{
+    convert_data_size_with_shellexpand, convert_duration_with_shellexpand,
     convert_numeric_with_shellexpand, convert_optional_numeric_with_shellexpand,
     convert_optional_string_with_shellexpand, convert_string_with_shellexpand,
     convert_vec_string_with_shellexpand,
@@ -139,7 +140,7 @@ pub struct ByteStreamConfig {
     /// 16KiB - 64KiB is optimal.
     ///
     /// Defaults: 64KiB
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub max_bytes_per_stream: usize,
 
     /// In the event a client disconnects while uploading a blob, we will hold
@@ -148,7 +149,7 @@ pub struct ByteStreamConfig {
     /// the same blob.
     ///
     /// Defaults: 10 (seconds)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub persist_stream_on_disconnect_timeout: usize,
 }
 
@@ -557,7 +558,7 @@ pub struct LocalWorkerConfig {
     /// longer than this time limit, the task will be rejected. Value in seconds.
     ///
     /// Default: 1200 (seconds / 20 mins)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub max_action_timeout: usize,
 
     /// If timeout is handled in `entrypoint` or another wrapper script.
@@ -667,7 +668,7 @@ pub struct GlobalConfig {
     /// a new file descriptor because the limit has been reached.
     ///
     /// Default: 1000 (1 second)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub idle_file_descriptor_timeout_millis: u64,
 
     /// This flag can be used to prevent metrics from being collected at runtime.
@@ -695,7 +696,7 @@ pub struct GlobalConfig {
     /// digest.
     ///
     /// Default: 1024*1024 (1MiB)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub default_digest_size_health_check: usize,
 }
 

nativelink-config/src/schedulers.rs

@@ -16,7 +16,7 @@ use std::collections::HashMap;
 
 use serde::Deserialize;
 
-use crate::serde_utils::convert_numeric_with_shellexpand;
+use crate::serde_utils::{convert_duration_with_shellexpand, convert_numeric_with_shellexpand};
 use crate::stores::{GrpcEndpoint, Retry, StoreRefName};
 
 #[allow(non_camel_case_types)]
@@ -97,13 +97,13 @@ pub struct SimpleScheduler {
     /// The amount of time to retain completed actions in memory for in case
     /// a WaitExecution is called after the action has completed.
     /// Default: 60 (seconds)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub retain_completed_for_s: u64,
 
     /// Remove workers from pool once the worker has not responded in this
     /// amount of time in seconds.
     /// Default: 5 (seconds)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub worker_timeout_s: u64,
 
     /// If a job returns an internal error or times out this many times when

nativelink-config/src/serde_utils.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use byte_unit::Byte;
+use humantime::parse_duration;
 use std::fmt;
 use std::marker::PhantomData;
 use std::str::FromStr;
@@ -138,3 +140,81 @@ pub fn convert_optional_string_with_shellexpand<'de, D: Deserializer<'de>>(
         Ok(None)
     }
 }
+
+pub fn convert_data_size_with_shellexpand<'de, D, T, E>(deserializer: D) -> Result<T, D::Error>
+where
+    D: Deserializer<'de>,
+    E: fmt::Display,
+    T: TryFrom<i64> + FromStr<Err = E>,
+    <T as TryFrom<i64>>::Error: fmt::Display,
+{
+    // define a visitor that deserializes
+    // `ActualData` encoded as json within a string
+    struct USizeVisitor<T: TryFrom<i64>>(PhantomData<T>);
+
+    impl<'de, T, FromStrErr> de::Visitor<'de> for USizeVisitor<T>
+    where
+        FromStrErr: fmt::Display,
+        T: TryFrom<i64> + FromStr<Err = FromStrErr>,
+        <T as TryFrom<i64>>::Error: fmt::Display,
+    {
+        type Value = T;
+
+        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+            formatter.write_str("a string containing json data")
+        }
+
+        fn visit_i64<E: de::Error>(self, v: i64) -> Result<Self::Value, E> {
+            v.try_into().map_err(de::Error::custom)
+        }
+
+        fn visit_str<E: de::Error>(self, v: &str) -> Result<Self::Value, E> {
+            let byte_size = Byte::parse_str(v, true).map_err(de::Error::custom);
+            let byte_size_str = byte_size?.as_u64().to_string();
+            (*shellexpand::env(&byte_size_str).map_err(de::Error::custom)?)
+                .parse::<T>()
+                .map_err(de::Error::custom)
+        }
+    }
+
+    deserializer.deserialize_any(USizeVisitor::<T>(PhantomData::<T> {}))
+}
+
+pub fn convert_duration_with_shellexpand<'de, D, T, E>(deserializer: D) -> Result<T, D::Error>
+where
+    D: Deserializer<'de>,
+    E: fmt::Display,
+    T: TryFrom<i64> + FromStr<Err = E>,
+    <T as TryFrom<i64>>::Error: fmt::Display,
+{
+    // define a visitor that deserializes
+    // `ActualData` encoded as json within a string
+    struct USizeVisitor<T: TryFrom<i64>>(PhantomData<T>);
+
+    impl<'de, T, FromStrErr> de::Visitor<'de> for USizeVisitor<T>
+    where
+        FromStrErr: fmt::Display,
+        T: TryFrom<i64> + FromStr<Err = FromStrErr>,
+        <T as TryFrom<i64>>::Error: fmt::Display,
+    {
+        type Value = T;
+
+        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+            formatter.write_str("a string containing json data")
+        }
+
+        fn visit_i64<E: de::Error>(self, v: i64) -> Result<Self::Value, E> {
+            v.try_into().map_err(de::Error::custom)
+        }
+
+        fn visit_str<E: de::Error>(self, v: &str) -> Result<Self::Value, E> {
+            let duration = parse_duration(v).map_err(de::Error::custom);
+            let duration_str = duration?.as_secs().to_string();
+            (*shellexpand::env(&duration_str).map_err(de::Error::custom)?)
+                .parse::<T>()
+                .map_err(de::Error::custom)
+        }
+    }
+
+    deserializer.deserialize_any(USizeVisitor::<T>(PhantomData::<T> {}))
+}

nativelink-config/src/stores.rs

@@ -15,6 +15,7 @@
 use serde::{Deserialize, Serialize};
 
 use crate::serde_utils::{
+    convert_data_size_with_shellexpand, convert_duration_with_shellexpand,
     convert_numeric_with_shellexpand, convert_optional_string_with_shellexpand,
     convert_string_with_shellexpand, convert_vec_string_with_shellexpand,
 };
@@ -203,7 +204,7 @@ pub struct ShardStore {
 #[serde(deny_unknown_fields)]
 pub struct SizePartitioningStore {
     /// Size to partition the data on.
-    #[serde(deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(deserialize_with = "convert_data_size_with_shellexpand")]
     pub size: u64,
 
     /// Store to send data when object is < (less than) size.
@@ -243,7 +244,7 @@ pub struct FilesystemStore {
     /// Buffer size to use when reading files. Generally this should be left
     /// to the default value except for testing.
     /// Default: 32k.
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub read_buffer_size: u32,
 
     /// Policy used to evict items out of the store. Failure to set this
@@ -255,7 +256,7 @@ pub struct FilesystemStore {
     /// value is used to determine an entry's actual size on disk consumed
     /// For a 4KB block size filesystem, a 1B file actually consumes 4KB
     /// Default: 4096
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub block_size: u64,
 }
 
@@ -297,7 +298,7 @@ pub struct DedupStore {
     /// deciding where to partition the data.
     ///
     /// Default: 65536 (64k)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub min_size: u32,
 
     /// A best-effort attempt will be made to keep the average size
@@ -311,13 +312,13 @@ pub struct DedupStore {
     /// details.
     ///
     /// Default: 262144 (256k)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub normal_size: u32,
 
     /// Maximum size a chunk is allowed to be.
     ///
     /// Default: 524288 (512k)
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub max_size: u32,
 
     /// Due to implementation detail, we want to prefer to download
@@ -396,7 +397,7 @@ pub struct Lz4Config {
     /// compression ratios.
     ///
     /// Default: 65536 (64k).
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub block_size: u32,
 
     /// Maximum size allowed to attempt to deserialize data into.
@@ -407,7 +408,7 @@ pub struct Lz4Config {
     /// allow you to specify the maximum that we'll attempt deserialize.
     ///
     /// Default: value in `block_size`.
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub max_decode_block_size: u32,
 }
 
@@ -447,19 +448,19 @@ pub struct CompressionStore {
 pub struct EvictionPolicy {
     /// Maximum number of bytes before eviction takes place.
     /// Default: 0. Zero means never evict based on size.
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub max_bytes: usize,
 
     /// When eviction starts based on hitting max_bytes, continue until
     /// max_bytes - evict_bytes is met to create a low watermark.  This stops
     /// operations from thrashing when the store is close to the limit.
     /// Default: 0
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub evict_bytes: usize,
 
     /// Maximum number of seconds for an entry to live before an eviction.
     /// Default: 0. Zero means never evict based on time.
-    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
     pub max_seconds: u32,
 
     /// Maximum size of the store before an eviction takes place.