Refactor drop into a single metric, change filters to return a result (…

…#722)

Co-authored-by: Mark Mandel <markmandel@google.com>

docs/src/services/proxy/filters/writing_custom_filters.md

@@ -40,13 +40,13 @@ sent to a downstream client.
 use quilkin::filters::prelude::*;
 
 impl Filter for Greet {
-    fn read(&self, ctx: &mut ReadContext) -> Option<()> {
+    fn read(&self, ctx: &mut ReadContext) -> Result<(), FilterError> {
         ctx.contents.extend(b"Hello");
-        Some(())
+        Ok(())
     }
-    fn write(&self, ctx: &mut WriteContext) -> Option<()> {
+    fn write(&self, ctx: &mut WriteContext) -> Result<(), FilterError> {
         ctx.contents.extend(b"Goodbye");
-        Some(())
+        Ok(())
     }
 }
 ```
@@ -65,7 +65,7 @@ impl StaticFilter for Greet {
     type Configuration = ();
     type BinaryConfiguration = ();
 
-    fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, Error> {
+    fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, CreationError> {
         Ok(Self)
     }
 }

examples/quilkin-filter-example/src/main.rs

@@ -58,15 +58,15 @@ struct Greet {
 }
 
 impl Filter for Greet {
-    fn read(&self, ctx: &mut ReadContext) -> Option<()> {
+    fn read(&self, ctx: &mut ReadContext) -> Result<(), FilterError> {
         ctx.contents
             .splice(0..0, format!("{} ", self.config.greeting).into_bytes());
-        Some(())
+        Ok(())
     }
-    fn write(&self, ctx: &mut WriteContext) -> Option<()> {
+    fn write(&self, ctx: &mut WriteContext) -> Result<(), FilterError> {
         ctx.contents
             .splice(0..0, format!("{} ", self.config.greeting).into_bytes());
-        Some(())
+        Ok(())
     }
 }
 // ANCHOR_END: filter
@@ -79,9 +79,7 @@ impl StaticFilter for Greet {
     type Configuration = Config;
     type BinaryConfiguration = proto::Greet;
 
-    fn try_from_config(
-        config: Option<Self::Configuration>,
-    ) -> Result<Self, quilkin::filters::Error> {
+    fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, CreationError> {
         Ok(Self {
             config: Self::ensure_config_exists(config)?,
         })
@@ -100,6 +98,7 @@ async fn main() -> quilkin::Result<()> {
     config.filters.store(std::sync::Arc::new(
         vec![quilkin::config::Filter {
             name: Greet::NAME.into(),
+            label: None,
             config: None,
         }]
         .try_into()?,

src/cli.rs

@@ -238,6 +238,7 @@ mod tests {
                 vec![
                     Filter {
                         name: Capture::factory().name().into(),
+                        label: None,
                         config: Some(serde_json::json!({
                             "suffix": {
                                 "size": 3,
@@ -247,6 +248,7 @@ mod tests {
                     },
                     Filter {
                         name: TokenRouter::factory().name().into(),
+                        label: None,
                         config: None,
                     },
                 ]

src/cli/proxy.rs

@@ -264,6 +264,7 @@ mod tests {
         config.filters.store(
             crate::filters::FilterChain::try_from(vec![config::Filter {
                 name: "TestFilter".to_string(),
+                label: None,
                 config: None,
             }])
             .map(Arc::new)

src/config.rs

@@ -45,9 +45,6 @@ pub use self::{
 
 base64_serde_type!(pub Base64Standard, base64::STANDARD);
 
-// For some log messages on the hot path (potentially per-packet), we log 1 out
-// of every `LOG_SAMPLING_RATE` occurrences to avoid spamming the logs.
-pub(crate) const LOG_SAMPLING_RATE: u64 = 1000;
 pub(crate) const BACKOFF_INITIAL_DELAY_MILLISECONDS: u64 = 500;
 pub(crate) const BACKOFF_MAX_DELAY_SECONDS: u64 = 30;
 pub(crate) const BACKOFF_MAX_JITTER_MILLISECONDS: u64 = 2000;
@@ -256,11 +253,12 @@ fn default_proxy_id() -> Slot<String> {
 #[serde(deny_unknown_fields)]
 pub struct Filter {
     pub name: String,
+    pub label: Option<String>,
     pub config: Option<serde_json::Value>,
 }
 
 impl TryFrom<crate::xds::config::listener::v3::Filter> for Filter {
-    type Error = Error;
+    type Error = CreationError;
 
     fn try_from(filter: crate::xds::config::listener::v3::Filter) -> Result<Self, Self::Error> {
         use crate::xds::config::listener::v3::filter::ConfigType;
@@ -269,15 +267,15 @@ impl TryFrom<crate::xds::config::listener::v3::Filter> for Filter {
             let config = match config_type {
                 ConfigType::TypedConfig(any) => any,
                 ConfigType::ConfigDiscovery(_) => {
-                    return Err(Error::FieldInvalid {
+                    return Err(CreationError::FieldInvalid {
                         field: "config_type".into(),
                         reason: "ConfigDiscovery is currently unsupported".into(),
                     })
                 }
             };
             Some(
                 crate::filters::FilterRegistry::get_factory(&filter.name)
-                    .ok_or_else(|| Error::NotFound(filter.name.clone()))?
+                    .ok_or_else(|| CreationError::NotFound(filter.name.clone()))?
                     .encode_config_to_json(config)?,
             )
         } else {
@@ -286,21 +284,23 @@ impl TryFrom<crate::xds::config::listener::v3::Filter> for Filter {
 
         Ok(Self {
             name: filter.name,
+            // TODO: keep the label across xDS
+            label: None,
             config,
         })
     }
 }
 
 impl TryFrom<Filter> for crate::xds::config::listener::v3::Filter {
-    type Error = Error;
+    type Error = CreationError;
 
     fn try_from(filter: Filter) -> Result<Self, Self::Error> {
         use crate::xds::config::listener::v3::filter::ConfigType;
 
         let config = if let Some(config) = filter.config {
             Some(
                 crate::filters::FilterRegistry::get_factory(&filter.name)
-                    .ok_or_else(|| Error::NotFound(filter.name.clone()))?
+                    .ok_or_else(|| CreationError::NotFound(filter.name.clone()))?
                     .encode_config_to_protobuf(config)?,
             )
         } else {
@@ -318,7 +318,8 @@ impl From<(String, FilterInstance)> for Filter {
     fn from((name, instance): (String, FilterInstance)) -> Self {
         Self {
             name,
-            config: Some(serde_json::Value::clone(&instance.config)),
+            label: instance.label().map(String::from),
+            config: Some(serde_json::Value::clone(instance.config())),
         }
     }
 }

src/config/config_type.rs

@@ -18,7 +18,7 @@ use std::convert::TryFrom;
 
 use bytes::Bytes;
 
-use crate::filters::Error;
+use crate::filters::CreationError;
 
 /// The configuration of a [`Filter`][crate::filters::Filter] from either a
 /// static or dynamic source.
@@ -51,20 +51,20 @@ impl ConfigType {
     pub fn deserialize<TextConfiguration, BinaryConfiguration>(
         self,
         filter_name: &str,
-    ) -> Result<(serde_json::Value, TextConfiguration), Error>
+    ) -> Result<(serde_json::Value, TextConfiguration), CreationError>
     where
         BinaryConfiguration: prost::Message + Default,
         TextConfiguration:
             serde::Serialize + for<'de> serde::Deserialize<'de> + TryFrom<BinaryConfiguration>,
-        Error: From<<BinaryConfiguration as TryInto<TextConfiguration>>::Error>,
+        CreationError: From<<BinaryConfiguration as TryInto<TextConfiguration>>::Error>,
     {
         match self {
             ConfigType::Static(ref config) => serde_yaml::to_string(config)
                 .and_then(|raw_config| {
                     serde_yaml::from_str::<TextConfiguration>(raw_config.as_str())
                 })
                 .map_err(|err| {
-                    Error::DeserializeFailed(format!(
+                    CreationError::DeserializeFailed(format!(
                         "filter `{filter_name}`: failed to YAML deserialize config: {err}",
                     ))
                 })
@@ -74,7 +74,7 @@ impl ConfigType {
                 }),
             ConfigType::Dynamic(config) => prost::Message::decode(Bytes::from(config.value))
                 .map_err(|err| {
-                    Error::DeserializeFailed(format!(
+                    CreationError::DeserializeFailed(format!(
                         "filter `{filter_name}`: config decode error: {err}",
                     ))
                 })
@@ -87,12 +87,12 @@ impl ConfigType {
     }
 
     // Returns an equivalent json value for the passed in config.
-    fn get_json_config<T>(filter_name: &str, config: &T) -> Result<serde_json::Value, Error>
+    fn get_json_config<T>(filter_name: &str, config: &T) -> Result<serde_json::Value, CreationError>
     where
         T: serde::Serialize + for<'de> serde::Deserialize<'de>,
     {
         serde_json::to_value(config).map_err(|err| {
-            Error::DeserializeFailed(format!(
+            CreationError::DeserializeFailed(format!(
                 "filter `{filter_name}`: failed to serialize config to json: {err}",
             ))
         })

src/config/error.rs

@@ -42,5 +42,5 @@ pub enum ValidationError {
     #[error(transparent)]
     ValueInvalid(#[from] ValueInvalidArgs),
     #[error(transparent)]
-    FilterInvalid(#[from] crate::filters::Error),
+    FilterInvalid(#[from] crate::filters::CreationError),
 }

src/config/slot.rs

@@ -193,11 +193,11 @@ impl<T: JsonSchema + Default> JsonSchema for Slot<T> {
 }
 
 impl<T: crate::filters::Filter + Default> crate::filters::Filter for Slot<T> {
-    fn read(&self, ctx: &mut ReadContext) -> Option<()> {
+    fn read(&self, ctx: &mut ReadContext) -> Result<(), FilterError> {
         self.load().read(ctx)
     }
 
-    fn write(&self, ctx: &mut WriteContext) -> Option<()> {
+    fn write(&self, ctx: &mut WriteContext) -> Result<(), FilterError> {
         self.load().write(ctx)
     }
 }

src/filters.rs

@@ -42,8 +42,8 @@ pub mod token_router;
 /// [`FilterFactory`].
 pub mod prelude {
     pub use super::{
-        ConvertProtoConfigError, CreateFilterArgs, Error, Filter, FilterInstance, ReadContext,
-        StaticFilter, WriteContext,
+        ConvertProtoConfigError, CreateFilterArgs, CreationError, Filter, FilterError,
+        FilterInstance, ReadContext, StaticFilter, WriteContext,
     };
 }
 
@@ -55,7 +55,7 @@ pub use self::{
     concatenate_bytes::ConcatenateBytes,
     debug::Debug,
     drop::Drop,
-    error::{ConvertProtoConfigError, Error},
+    error::{ConvertProtoConfigError, CreationError, FilterError},
     factory::{CreateFilterArgs, DynFilterFactory, FilterFactory, FilterInstance},
     firewall::Firewall,
     load_balancer::LoadBalancer,
@@ -83,13 +83,13 @@ pub use self::chain::FilterChain;
 /// struct Greet;
 ///
 /// impl Filter for Greet {
-///     fn read(&self, ctx: &mut ReadContext) -> Option<()> {
+///     fn read(&self, ctx: &mut ReadContext) -> Result<(), FilterError> {
 ///         ctx.contents.splice(0..0, b"Hello ".into_iter().copied());
-///         Some(())
+///         Ok(())
 ///     }
-///     fn write(&self, ctx: &mut WriteContext) -> Option<()> {
+///     fn write(&self, ctx: &mut WriteContext) -> Result<(), FilterError> {
 ///         ctx.contents.splice(0..0, b"Goodbye ".into_iter().copied());
-///         Some(())
+///         Ok(())
 ///     }
 /// }
 ///
@@ -98,7 +98,7 @@ pub use self::chain::FilterChain;
 ///     type Configuration = ();
 ///     type BinaryConfiguration = ();
 ///
-///     fn try_from_config(_: Option<Self::Configuration>) -> Result<Self, quilkin::filters::Error> {
+///     fn try_from_config(_: Option<Self::Configuration>) -> Result<Self, CreationError> {
 ///         Ok(Self)
 ///     }
 /// }
@@ -107,7 +107,7 @@ pub trait StaticFilter: Filter + Sized
 // This where clause simply states that `Configuration`'s and
 // `BinaryConfiguration`'s `Error` types are compatible with `filters::Error`.
 where
-    Error: From<<Self::Configuration as TryFrom<Self::BinaryConfiguration>>::Error>
+    CreationError: From<<Self::Configuration as TryFrom<Self::BinaryConfiguration>>::Error>
         + From<<Self::BinaryConfiguration as TryFrom<Self::Configuration>>::Error>,
 {
     /// The globally unique name of the filter.
@@ -131,7 +131,7 @@ where
     /// Instantiates a new [`StaticFilter`] from the given configuration, if any.
     /// # Errors
     /// If the provided configuration is invalid.
-    fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, Error>;
+    fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, CreationError>;
 
     /// Instantiates a new [`StaticFilter`] from the given configuration, if any.
     /// # Panics
@@ -152,15 +152,16 @@ where
     /// which require a fully initialized [`Self::Configuration`].
     fn ensure_config_exists(
         config: Option<Self::Configuration>,
-    ) -> Result<Self::Configuration, Error> {
-        config.ok_or(Error::MissingConfig(Self::NAME))
+    ) -> Result<Self::Configuration, CreationError> {
+        config.ok_or(CreationError::MissingConfig(Self::NAME))
     }
 
     fn as_filter_config(
         config: impl Into<Option<Self::Configuration>>,
-    ) -> Result<crate::config::Filter, Error> {
+    ) -> Result<crate::config::Filter, CreationError> {
         Ok(crate::config::Filter {
             name: Self::NAME.into(),
+            label: None,
             config: config
                 .into()
                 .map(|config| serde_json::to_value(&config))
@@ -197,8 +198,8 @@ pub trait Filter: Send + Sync {
     /// This function should return an `Some` if the packet processing should
     /// proceed. If the packet should be rejected, it will return [`None`]
     /// instead. By default, the context passes through unchanged.
-    fn read(&self, _: &mut ReadContext) -> Option<()> {
-        Some(())
+    fn read(&self, _: &mut ReadContext) -> Result<(), FilterError> {
+        Ok(())
     }
 
     /// [`Filter::write`] is invoked when the proxy is about to send data to a
@@ -207,7 +208,7 @@ pub trait Filter: Send + Sync {
     ///
     /// This function should return an `Some` if the packet processing should
     /// proceed. If the packet should be rejected, it will return [`None`]
-    fn write(&self, _: &mut WriteContext) -> Option<()> {
-        Some(())
+    fn write(&self, _: &mut WriteContext) -> Result<(), FilterError> {
+        Ok(())
     }
 }