shotover config: Fix ignored fields (#1291)

custom-transforms-example/src/redis_get_rewrite.rs

@@ -6,6 +6,7 @@ use shotover::message::Messages;
 use shotover::transforms::{Transform, TransformBuilder, TransformConfig, Transforms, Wrapper};
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct RedisGetRewriteConfig {
  pub result: String,
 }

shotover-proxy/tests/cassandra_int_tests/protect.rs

@@ -5,6 +5,7 @@ use test_helpers::connection::cassandra::{
 };
 
 #[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct Protected {
  _cipher: Vec<u8>,
  _nonce: Nonce,

shotover/src/config/chain.rs

@@ -7,6 +7,7 @@ use std::fmt::{self, Debug};
 use std::iter;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct TransformChainConfig(
  #[serde(rename = "TransformChain", deserialize_with = "vec_transform_config")]
  pub Vec<Box<dyn TransformConfig>>,

shotover/src/config/mod.rs

@@ -5,6 +5,7 @@ pub mod chain;
 pub mod topology;
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct Config {
  pub main_log_level: String,
  pub observability_interface: String,

shotover/src/config/topology.rs

@@ -9,6 +9,7 @@ use tokio::sync::watch;
 use tracing::info;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct Topology {
  pub sources: HashMap<String, SourceConfig>,
  pub chain_config: HashMap<String, TransformChainConfig>,

shotover/src/message/mod.rs

@@ -426,6 +426,7 @@ pub enum Encodable {
 }
 
 #[derive(PartialEq, Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub enum QueryType {
  Read,
  Write,

shotover/src/sources/cassandra.rs

@@ -12,6 +12,7 @@ use tokio::task::JoinHandle;
 use tracing::{error, info};
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct CassandraConfig {
  pub listen_addr: String,
  pub connection_limit: Option<usize>,

shotover/src/sources/kafka.rs

@@ -11,6 +11,7 @@ use tokio::task::JoinHandle;
 use tracing::{error, info};
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct KafkaConfig {
  pub listen_addr: String,
  pub connection_limit: Option<usize>,

shotover/src/sources/mod.rs

@@ -12,6 +12,7 @@ pub mod kafka;
 pub mod redis;
 
 #[derive(Deserialize, Debug, Clone, Copy)]
+#[serde(deny_unknown_fields)]
 pub enum Transport {
  Tcp,
  WebSocket,
@@ -35,6 +36,7 @@ impl Source {
 }
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub enum SourceConfig {
  Cassandra(CassandraConfig),
  Redis(RedisConfig),

shotover/src/sources/redis.rs

@@ -11,6 +11,7 @@ use tokio::task::JoinHandle;
 use tracing::{error, info};
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct RedisConfig {
  pub listen_addr: String,
  pub connection_limit: Option<usize>,

shotover/src/tls.rs

@@ -19,6 +19,7 @@ use tokio_rustls::server::TlsStream as TlsStreamServer;
 use tokio_rustls::{TlsAcceptor as RustlsAcceptor, TlsConnector as RustlsConnector};
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct TlsAcceptorConfig {
  /// Path to the certificate authority in PEM format
  pub certificate_authority_path: Option<String>,
@@ -120,6 +121,7 @@ impl TlsAcceptor {
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct TlsConnectorConfig {
  /// Path to the certificate authority in PEM format
  pub certificate_authority_path: String,

shotover/src/transforms/cassandra/peers_rewrite.rs

@@ -14,6 +14,7 @@ use cql3_parser::select::SelectElement;
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct CassandraPeersRewriteConfig {
  pub port: u16,
 }

shotover/src/transforms/cassandra/sink_cluster/mod.rs

@@ -48,6 +48,7 @@ const SYSTEM_KEYSPACES: [IdentifierRef<'static>; 3] = [
 ];
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct CassandraSinkClusterConfig {
  /// contact points must be within the specified data_center and rack.
  /// If this is not followed, shotover's invariants will still be upheld but shotover will communicate with a
@@ -181,6 +182,7 @@ impl TransformBuilder for CassandraSinkClusterBuilder {
 }
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct ShotoverNode {
  pub address: SocketAddr,
  pub data_center: String,

shotover/src/transforms/cassandra/sink_single.rs

@@ -16,6 +16,7 @@ use tokio::sync::{mpsc, oneshot};
 use tracing::trace;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct CassandraSinkSingleConfig {
  #[serde(rename = "remote_address")]
  pub address: String,

shotover/src/transforms/coalesce.rs

@@ -14,6 +14,7 @@ pub struct Coalesce {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct CoalesceConfig {
  pub flush_when_buffered_message_count: Option<usize>,
  pub flush_when_millis_since_last_flush: Option<u128>,

shotover/src/transforms/debug/force_parse.rs

@@ -15,6 +15,7 @@ use serde::Deserialize;
 /// Messages that pass through this transform will be parsed.
 /// Must be individually enabled at the request or response level.
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct DebugForceParseConfig {
  pub parse_requests: bool,
  pub parse_responses: bool,
@@ -37,6 +38,7 @@ impl TransformConfig for DebugForceParseConfig {
 /// Messages that pass through this transform will be parsed and then reencoded.
 /// Must be individually enabled at the request or response level.
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct DebugForceEncodeConfig {
  pub encode_requests: bool,
  pub encode_responses: bool,

shotover/src/transforms/debug/log_to_file.rs

@@ -9,6 +9,7 @@ use std::sync::Arc;
 use tracing::{error, info};
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct DebugLogToFileConfig;
 
 #[cfg(feature = "alpha-transforms")]

shotover/src/transforms/debug/printer.rs

@@ -6,6 +6,7 @@ use serde::Deserialize;
 use tracing::info;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct DebugPrinterConfig;
 
 #[typetag::deserialize(name = "DebugPrinter")]

shotover/src/transforms/debug/returner.rs

@@ -6,6 +6,7 @@ use async_trait::async_trait;
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct DebugReturnerConfig {
  #[serde(flatten)]
  response: Response,
@@ -20,6 +21,7 @@ impl TransformConfig for DebugReturnerConfig {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub enum Response {
  #[serde(skip)]
  Message(Messages),

shotover/src/transforms/distributed/tuneable_consistency_scatter.rs

@@ -12,6 +12,7 @@ use std::collections::HashMap;
 use tracing::{error, warn};
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct TuneableConsistencyScatterConfig {
  pub route_map: HashMap<String, TransformChainConfig>,
  pub write_consistency: i32,

shotover/src/transforms/filter.rs

@@ -15,6 +15,7 @@ pub struct QueryTypeFilter {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct QueryTypeFilterConfig {
  pub filter: QueryType,
 }

shotover/src/transforms/kafka/sink_cluster.rs

@@ -29,6 +29,7 @@ use tokio::time::timeout;
 use uuid::Uuid;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct KafkaSinkClusterConfig {
  pub first_contact_points: Vec<String>,
  pub shotover_nodes: Vec<String>,

shotover/src/transforms/kafka/sink_single.rs

@@ -15,6 +15,7 @@ use tokio::sync::{mpsc, oneshot};
 use tokio::time::timeout;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct KafkaSinkSingleConfig {
  #[serde(rename = "remote_address")]
  pub address: String,

shotover/src/transforms/load_balance.rs

@@ -10,6 +10,7 @@ use std::sync::Arc;
 use tokio::sync::Mutex;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct ConnectionBalanceAndPoolConfig {
  pub name: String,
  pub max_connections: usize,

shotover/src/transforms/null.rs

@@ -5,6 +5,7 @@ use async_trait::async_trait;
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct NullSinkConfig;
 
 #[typetag::deserialize(name = "NullSink")]

shotover/src/transforms/parallel_map.rs

@@ -64,6 +64,7 @@ where
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct ParallelMapConfig {
  pub parallelism: u32,
  pub chain: TransformChainConfig,

shotover/src/transforms/protect/key_management.rs

@@ -18,6 +18,7 @@ pub enum KeyManager {
 }
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub enum KeyManagerConfig {
  AWSKms {
  region: String,

shotover/src/transforms/protect/local_kek.rs

@@ -13,6 +13,7 @@ pub struct LocalKeyManagement {
 }
 
 #[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct DEKStructure {
  pub nonce: Nonce,
  pub key: Vec<u8>,

shotover/src/transforms/protect/mod.rs

@@ -21,6 +21,7 @@ mod local_kek;
 mod pkcs_11;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct ProtectConfig {
  pub keyspace_table_columns: HashMap<String, HashMap<String, Vec<String>>>,
  pub key_manager: KeyManagerConfig,

shotover/src/transforms/query_counter.rs

@@ -14,6 +14,7 @@ pub struct QueryCounter {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct QueryCounterConfig {
  pub name: String,
 }

shotover/src/transforms/redis/cache.rs

@@ -49,12 +49,14 @@ enum CacheableState {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct TableCacheSchemaConfig {
  partition_key: Vec<String>,
  range_key: Vec<String>,
 }
 
 #[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
 pub struct TableCacheSchema {
  partition_key: Vec<Identifier>,
  range_key: Vec<Identifier>,
@@ -74,6 +76,7 @@ impl From<&TableCacheSchemaConfig> for TableCacheSchema {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RedisConfig {
  pub caching_schema: HashMap<String, TableCacheSchemaConfig>,
  pub chain: TransformChainConfig,

shotover/src/transforms/redis/cluster_ports_rewrite.rs

@@ -8,6 +8,7 @@ use bytes::{BufMut, Bytes, BytesMut};
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RedisClusterPortsRewriteConfig {
  pub new_port: u16,
 }

shotover/src/transforms/redis/sink_cluster.rs

@@ -37,6 +37,7 @@ const SLOT_SIZE: usize = 16384;
 type ChannelMap = HashMap<String, Vec<UnboundedSender<Request>>>;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RedisSinkClusterConfig {
  pub first_contact_points: Vec<String>,
  pub direct_destination: Option<String>,

shotover/src/transforms/redis/sink_single.rs

@@ -21,6 +21,7 @@ use tokio_util::codec::{FramedRead, FramedWrite};
 use tracing::Instrument;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RedisSinkSingleConfig {
  #[serde(rename = "remote_address")]
  pub address: String,

shotover/src/transforms/redis/timestamp_tagging.rs

@@ -12,6 +12,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use tracing::{debug, trace};
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RedisTimestampTaggerConfig;
 
 #[typetag::deserialize(name = "RedisTimestampTagger")]

shotover/src/transforms/tee.rs

@@ -96,6 +96,7 @@ pub enum ConsistencyBehavior {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct TeeConfig {
  pub behavior: Option<ConsistencyBehaviorConfig>,
  pub timeout_micros: Option<u64>,
@@ -104,6 +105,7 @@ pub struct TeeConfig {
 }
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub enum ConsistencyBehaviorConfig {
  Ignore,
  FailOnMismatch,

shotover/src/transforms/throttling.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use super::Transforms;
 
 #[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
 pub struct RequestThrottlingConfig {
  pub max_requests_per_second: NonZeroU32,
 }