Merge pull request #2130 from influxdata/dom/datatypes-serde

feat: DatabaseRules (de)serialisation support

Cargo.lock

@@ -832,6 +832,7 @@ dependencies = [
  "percent-encoding",
  "regex",
  "serde",
+ "serde_regex",
  "snafu",
  "test_helpers",
 ]
@@ -3751,6 +3752,16 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_regex"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a8136f1a4ea815d7eac4101cfd0b16dc0cb5e1fe1b8609dfd728058656b7badf"
+dependencies = [
+ "regex",
+ "serde",
+]
+
 [[package]]
 name = "serde_urlencoded"
 version = "0.7.0"

data_types/Cargo.toml

@@ -15,6 +15,7 @@ once_cell = { version = "1.4.0", features = ["parking_lot"] }
 percent-encoding = "2.1.0"
 regex = "1.4"
 serde = { version = "1.0", features = ["rc", "derive"] }
+serde_regex = "1.1"
 snafu = "0.6"
 
 [dev-dependencies] # In alphabetical order

data_types/src/consistent_hasher.rs

@@ -1,6 +1,8 @@
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 
+use serde::{Deserialize, Serialize};
+
 /// A ConsistentHasher implements a simple consistent hashing mechanism
 /// that maps a point to the nearest "node" N.
 ///
@@ -11,7 +13,7 @@ use std::hash::{Hash, Hasher};
 ///
 /// e.g. you can use it find the ShardID in vector of ShardIds
 /// that is closest to a given hash value.
-#[derive(Debug, Eq, PartialEq, Clone, Default)]
+#[derive(Debug, Eq, PartialEq, Clone, Default, Serialize, Deserialize)]
 pub struct ConsistentHasher<T>
 where
     T: Copy + Hash,

data_types/src/database_rules.rs

@@ -9,6 +9,7 @@ use std::{
 
 use chrono::{TimeZone, Utc};
 use regex::Regex;
+use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, Snafu};
 
 use influxdb_line_protocol::ParsedLine;
@@ -34,7 +35,7 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
 
 /// DatabaseRules contains the rules for replicating data, sending data to
 /// subscribers, and querying data for a single database.
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct DatabaseRules {
     /// The name of the database
     pub name: DatabaseName<'static>,
@@ -58,13 +59,13 @@ pub struct DatabaseRules {
     pub write_buffer_connection: Option<WriteBufferConnection>,
 }
 
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub enum WriteBufferConnection {
     Writing(String),
     Reading(String),
 }
 
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub enum RoutingRules {
     // A routing config defines the target where all data plane operations for
     // a given database are delegated to.
@@ -120,7 +121,7 @@ pub const DEFAULT_PERSIST_AGE_THRESHOLD_SECONDS: u32 = 30 * 60;
 pub const DEFAULT_LATE_ARRIVE_WINDOW_SECONDS: u32 = 5 * 60;
 
 /// Configures how data automatically flows through the system
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct LifecycleRules {
     /// Once the total amount of buffered data in memory reaches this size start
     /// dropping data from memory
@@ -211,7 +212,7 @@ impl Default for LifecycleRules {
 ///
 /// The key is constructed in order of the template parts; thus ordering changes
 /// what partition key is generated.
-#[derive(Debug, Default, Eq, PartialEq, Clone)]
+#[derive(Debug, Default, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct PartitionTemplate {
     pub parts: Vec<TemplatePart>,
 }
@@ -244,7 +245,7 @@ impl Partitioner for PartitionTemplate {
 
 /// `TemplatePart` specifies what part of a row should be used to compute this
 /// part of a partition key.
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub enum TemplatePart {
     /// The name of a table
     Table,
@@ -264,7 +265,7 @@ pub enum TemplatePart {
 
 /// `RegexCapture` is for pulling parts of a string column into the partition
 /// key.
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct RegexCapture {
     pub column: String,
     pub regex: String,
@@ -279,7 +280,7 @@ pub struct RegexCapture {
 /// For example, a time format of "%Y-%m-%d %H:%M:%S" will produce
 /// partition key parts such as "2021-03-14 12:25:21" and
 /// "2021-04-14 12:24:21"
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct StrftimeColumn {
     pub column: String,
     pub format: String,
@@ -287,7 +288,7 @@ pub struct StrftimeColumn {
 
 /// A routing config defines the destination where to route all data plane operations
 /// for a given database.
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct RoutingConfig {
     pub sink: Sink,
 }
@@ -309,7 +310,7 @@ pub trait Sharder {
 /// based on table name and assign to 1 of 10 shards. Within each
 /// shard you would have partitions, which would likely be based off time.
 /// This makes it possible to horizontally scale out writes.
-#[derive(Debug, Eq, PartialEq, Clone, Default)]
+#[derive(Debug, Eq, PartialEq, Clone, Default, Serialize, Deserialize)]
 pub struct ShardConfig {
     /// Each matcher, if any, is evaluated in order.
     /// If there is a match, the route will be evaluated to
@@ -333,7 +334,7 @@ pub struct ShardConfig {
 }
 
 /// Configuration for a specific IOx sink
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub enum Sink {
     Iox(NodeGroup),
     Kafka(KafkaProducer),
@@ -384,7 +385,7 @@ impl Sharder for ShardConfig {
 
 /// Maps a matcher with specific shard. If the line/row matches
 /// it should be sent to the group.
-#[derive(Debug, Eq, PartialEq, Clone, Default)]
+#[derive(Debug, Eq, PartialEq, Clone, Default, Serialize, Deserialize)]
 pub struct MatcherToShard {
     pub matcher: Matcher,
     pub shard: ShardId,
@@ -393,12 +394,12 @@ pub struct MatcherToShard {
 /// A collection of IOx nodes
 pub type NodeGroup = Vec<ServerId>;
 
-#[derive(Debug, Eq, PartialEq, Clone)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct KafkaProducer {}
 
 /// HashRing is a rule for creating a hash key for a row and mapping that to
 /// an individual node on a ring.
-#[derive(Debug, Eq, PartialEq, Clone, Default)]
+#[derive(Debug, Eq, PartialEq, Clone, Default, Serialize, Deserialize)]
 pub struct HashRing {
     /// If true the table name will be included in the hash key
     pub table_name: bool,
@@ -410,9 +411,10 @@ pub struct HashRing {
 
 /// A matcher is used to match routing rules or subscriptions on a row-by-row
 /// (or line) basis.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct Matcher {
     /// if provided, match if the table name matches against the regex
+    #[serde(with = "serde_regex")]
     pub table_name_regex: Option<Regex>,
     // paul: what should we use for predicate matching here against a single row/line?
     pub predicate: Option<String>,
@@ -455,6 +457,7 @@ pub type PartitionId = String;
 #[cfg(test)]
 mod tests {
     use influxdb_line_protocol::parse_lines;
+    use serde::{Deserialize, Serialize};
 
     use super::*;
 
@@ -724,4 +727,11 @@ mod tests {
     fn parse_line(line: &str) -> ParsedLine<'_> {
         parsed_lines(line).pop().unwrap()
     }
+
+    #[test]
+    fn test_assert_serde() {
+        fn ensure<'de, T: Serialize + Deserialize<'de>>(_t: T) {}
+
+        ensure(DatabaseRules::new(DatabaseName::new("bananas").unwrap()));
+    }
 }