Add matches filter (#447)

Cargo.toml

@@ -43,7 +43,7 @@ quilkin-macros = { version = "0.3.0-dev", path = "./macros" }
 # Crates.io
 base64 = "0.13.0"
 base64-serde = "0.6.1"
-bytes = "1.1.0"
+bytes = { version = "1.1.0", features = ["serde"] }
 clap = { version = "3", features = ["cargo"] }
 dashmap = "4.0.2"
 either = "1.6.1"

build.rs

@@ -28,14 +28,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         "proto/data-plane-api/envoy/type/metadata/v3/metadata.proto",
         "proto/data-plane-api/envoy/type/tracing/v3/custom_tag.proto",
         "proto/udpa/xds/core/v3/resource_name.proto",
-        "proto/quilkin/extensions/filters/debug/v1alpha1/debug.proto",
         "proto/quilkin/extensions/filters/capture_bytes/v1alpha1/capture_bytes.proto",
         "proto/quilkin/extensions/filters/compress/v1alpha1/compress.proto",
         "proto/quilkin/extensions/filters/concatenate_bytes/v1alpha1/concatenate_bytes.proto",
+        "proto/quilkin/extensions/filters/debug/v1alpha1/debug.proto",
+        "proto/quilkin/extensions/filters/firewall/v1alpha1/firewall.proto",
         "proto/quilkin/extensions/filters/load_balancer/v1alpha1/load_balancer.proto",
         "proto/quilkin/extensions/filters/local_rate_limit/v1alpha1/local_rate_limit.proto",
+        "proto/quilkin/extensions/filters/matches/v1alpha1/matches.proto",
         "proto/quilkin/extensions/filters/token_router/v1alpha1/token_router.proto",
-        "proto/quilkin/extensions/filters/firewall/v1alpha1/firewall.proto",
     ]
     .iter()
     .map(|name| std::env::current_dir().unwrap().join(name))

docs/src/filters/matches.md

@@ -0,0 +1,43 @@
+# Matches
+
+The `Matches` filter's job is to provide a mechanism to change behaviour based
+on dynamic metadata. This filter behaves similarly to the `match` expression
+in Rust or `switch` statements in other languages.
+
+#### Filter name
+```text
+quilkin.extensions.filters.matches.v1alpha1.Matches
+```
+
+### Configuration Examples
+```rust
+# let yaml = "
+version: v1alpha1
+static:
+  endpoints:
+    - address: 127.0.0.1:26000
+    - address: 127.0.0.1:26001
+  filters:
+    - name: quilkin.extensions.filters.capture_bytes.v1alpha1.CaptureBytes
+      config:
+          strategy: PREFIX
+          metadataKey: myapp.com/token
+          size: 3
+          remove: false
+    - name: quilkin.extensions.filters.matches.v1alpha1.Matches
+      config:
+          on_read:
+            metadataKey: myapp.com/token
+            branches:
+                - value: abc
+                  filter: quilkin.extensions.filters.concatenate_bytes.v1alpha1.ConcatenateBytes
+                  config:
+                    on_read: APPEND
+                    bytes: eHl6 # "xyz"
+# ";
+# let config = quilkin::config::Config::from_reader(yaml.as_bytes()).unwrap();
+# assert_eq!(config.source.get_static_filters().unwrap().len(), 1);
+# quilkin::Builder::from(std::sync::Arc::new(config)).validate().unwrap();
+```
+
+View the [Matches](../../api/quilkin/filters/matches/struct.Config.html) filter documentation for more details.

docs/src/filters/writing_custom_filters.md

@@ -238,7 +238,7 @@ impl FilterFactory for GreetFilterFactory {
     fn create_filter(&self, args: CreateFilterArgs) -> Result<FilterInstance, Error> {
         let config = match args.config.unwrap() {
           ConfigType::Static(config) => {
-              serde_yaml::from_str::<Config>(serde_yaml::to_string(config).unwrap().as_str())
+              serde_yaml::from_str::<Config>(serde_yaml::to_string(&config).unwrap().as_str())
                 .unwrap()
           }
           ConfigType::Dynamic(_) => unimplemented!("dynamic config is not yet supported for this filter"),
@@ -263,7 +263,7 @@ has a [Dynamic][ConfigType::dynamic] variant.
 ```rust,ignore
 let config = match args.config.unwrap() {
     ConfigType::Static(config) => {
-        serde_yaml::from_str::<Config>(serde_yaml::to_string(config).unwrap().as_str())
+        serde_yaml::from_str::<Config>(serde_yaml::to_string(&config).unwrap().as_str())
          .unwrap()
     }
     ConfigType::Dynamic(_) => unimplemented!("dynamic config is not yet supported for this filter"),

proto/quilkin/extensions/filters/matches/v1alpha1/matches.proto

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto3";
+
+package quilkin.extensions.filters.matches.v1alpha1;
+
+import "google/protobuf/wrappers.proto";
+import "google/protobuf/struct.proto";
+import "google/protobuf/any.proto";
+
+message Matches {
+    message Branch {
+        google.protobuf.Value value = 1;
+        google.protobuf.StringValue filter = 2;
+        optional google.protobuf.Any config = 3;
+    }
+
+    message FallthroughFilter {
+        google.protobuf.StringValue filter = 2;
+        optional google.protobuf.Any config = 3;
+    }
+
+    message DirectionalConfig {
+        google.protobuf.StringValue metadata_key = 1;
+        repeated Branch branches = 2;
+        oneof fallthrough {
+            google.protobuf.NullValue pass = 3;
+            google.protobuf.NullValue drop = 4;
+            FallthroughFilter filter = 5;
+        }
+    }
+
+    optional DirectionalConfig on_read = 1;
+    optional DirectionalConfig on_write = 2;
+}

src/config/config_type.rs

@@ -22,15 +22,15 @@ use crate::filters::{ConvertProtoConfigError, Error};
 
 /// The configuration of a [`Filter`][crate::filters::Filter] from either a
 /// static or dynamic source.
-#[derive(Debug)]
-pub enum ConfigType<'a> {
+#[derive(Clone, Debug, PartialEq)]
+pub enum ConfigType {
     /// Static configuration from YAML.
-    Static(&'a serde_yaml::Value),
+    Static(serde_yaml::Value),
     /// Dynamic configuration from Protobuf.
     Dynamic(prost_types::Any),
 }
 
-impl ConfigType<'_> {
+impl ConfigType {
     /// Deserializes takes two type arguments `Static` and `Dynamic` representing
     /// the types of a static and dynamic configuration respectively.
     ///
@@ -57,7 +57,7 @@ impl ConfigType<'_> {
             + TryFrom<Dynamic, Error = ConvertProtoConfigError>,
     {
         match self {
-            ConfigType::Static(config) => serde_yaml::to_string(config)
+            ConfigType::Static(ref config) => serde_yaml::to_string(config)
                 .and_then(|raw_config| serde_yaml::from_str::<Static>(raw_config.as_str()))
                 .map_err(|err| {
                     Error::DeserializeFailed(format!(
@@ -95,6 +95,29 @@ impl ConfigType<'_> {
     }
 }
 
+impl<'de> serde::Deserialize<'de> for ConfigType {
+    fn deserialize<D>(de: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        serde_yaml::Value::deserialize(de).map(ConfigType::Static)
+    }
+}
+
+impl<'de> serde::Serialize for ConfigType {
+    fn serialize<S>(&self, ser: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            Self::Static(value) => value.serialize(ser),
+            Self::Dynamic(_) => Err(serde::ser::Error::custom(
+                "Protobuf configs can't be serialized.",
+            )),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::ConfigType;

src/config/error.rs

@@ -24,7 +24,7 @@ pub struct ValueInvalidArgs {
 }
 
 /// Validation failure for a Config
-#[derive(Debug, PartialEq)]
+#[derive(Debug)]
 pub enum ValidationError {
     NotUnique(String),
     EmptyList(String),

src/filters.rs

@@ -33,6 +33,7 @@ pub mod debug;
 pub mod firewall;
 pub mod load_balancer;
 pub mod local_rate_limit;
+pub mod matches;
 pub mod metadata;
 pub mod token_router;
 

src/filters/capture_bytes.rs

@@ -121,7 +121,8 @@ mod tests {
     use super::capture::{Capture, Prefix, Suffix};
 
     use crate::filters::{
-        metadata::CAPTURED_BYTES, CreateFilterArgs, Filter, FilterFactory, ReadContext,
+        metadata::CAPTURED_BYTES, CreateFilterArgs, Filter, FilterFactory, FilterRegistry,
+        ReadContext,
     };
 
     const TOKEN_KEY: &str = "TOKEN";
@@ -147,8 +148,9 @@ mod tests {
 
         let filter = factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map)),
+                Some(Value::Mapping(map)),
             ))
             .unwrap()
             .filter;
@@ -162,8 +164,9 @@ mod tests {
         map.insert(Value::String("size".into()), Value::Number(3.into()));
         let filter = factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map)),
+                Some(Value::Mapping(map)),
             ))
             .unwrap()
             .filter;
@@ -177,8 +180,9 @@ mod tests {
         map.insert(Value::String("size".into()), Value::String("WRONG".into()));
 
         let result = factory.create_filter(CreateFilterArgs::fixed(
+            FilterRegistry::default(),
             Registry::default(),
-            Some(&Value::Mapping(map)),
+            Some(Value::Mapping(map)),
         ));
         assert!(result.is_err(), "Should be an error");
     }

src/filters/chain.rs

@@ -119,8 +119,12 @@ impl FilterChain {
         for filter_config in filter_configs {
             match filter_registry.get(
                 &filter_config.name,
-                CreateFilterArgs::fixed(metrics_registry.clone(), filter_config.config.as_ref())
-                    .with_metrics_registry(metrics_registry.clone()),
+                CreateFilterArgs::fixed(
+                    filter_registry.clone(),
+                    metrics_registry.clone(),
+                    filter_config.config,
+                )
+                .with_metrics_registry(metrics_registry.clone()),
             ) {
                 Ok(filter) => filters.push((filter_config.name, filter)),
                 Err(err) => {

src/filters/compress.rs

@@ -181,7 +181,7 @@ mod tests {
     use crate::endpoint::{Endpoint, Endpoints, UpstreamEndpoints};
     use crate::filters::{
         compress::{compressor::Snappy, Compressor},
-        CreateFilterArgs, Filter, FilterFactory, ReadContext, WriteContext,
+        CreateFilterArgs, Filter, FilterFactory, FilterRegistry, ReadContext, WriteContext,
     };
 
     use super::quilkin::extensions::filters::compress::v1alpha1::{
@@ -293,8 +293,9 @@ mod tests {
         );
         let filter = factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map)),
+                Some(Value::Mapping(map)),
             ))
             .expect("should create a filter")
             .filter;
@@ -315,7 +316,8 @@ mod tests {
             Value::String("COMPRESS".into()),
         );
         let config = Value::Mapping(map);
-        let args = CreateFilterArgs::fixed(Registry::default(), Some(&config));
+        let args =
+            CreateFilterArgs::fixed(FilterRegistry::default(), Registry::default(), Some(config));
 
         let filter = factory
             .create_filter(args)

src/filters/debug.rs

@@ -115,6 +115,7 @@ impl TryFrom<ProtoDebug> for Config {
 
 #[cfg(test)]
 mod tests {
+    use crate::filters::FilterRegistry;
     use crate::test_utils::{assert_filter_read_no_change, assert_write_no_change};
     use serde_yaml::Mapping;
     use serde_yaml::Value;
@@ -148,8 +149,9 @@ mod tests {
         map.insert(Value::from("id"), Value::from("name"));
         assert!(factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map)),
+                Some(Value::Mapping(map)),
             ))
             .is_ok());
     }
@@ -162,8 +164,9 @@ mod tests {
         map.insert(Value::from("id"), Value::from("name"));
         assert!(factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map)),
+                Some(Value::Mapping(map)),
             ))
             .is_ok());
     }
@@ -176,8 +179,9 @@ mod tests {
         map.insert(Value::from("id"), Value::Sequence(vec![]));
         assert!(factory
             .create_filter(CreateFilterArgs::fixed(
+                FilterRegistry::default(),
                 Registry::default(),
-                Some(&Value::Mapping(map))
+                Some(Value::Mapping(map))
             ))
             .is_err());
     }

src/filters/error.rs

@@ -66,9 +66,9 @@ pub struct ConvertProtoConfigError {
 }
 
 impl ConvertProtoConfigError {
-    pub fn new(reason: impl Into<String>, field: Option<String>) -> Self {
+    pub fn new(reason: impl std::fmt::Display, field: Option<String>) -> Self {
         Self {
-            reason: reason.into(),
+            reason: reason.to_string(),
             field,
         }
     }