From 518af95616cb2fbe11a1926438835c511d466893 Mon Sep 17 00:00:00 2001 From: GridexX Date: Fri, 19 Aug 2022 10:06:18 +0200 Subject: [PATCH] feat: implement etcd client Create a new internal_etcd interface witch contains all methods related to the ETCD client. Signed-off-by: GridexX --- Cargo.lock | 137 ++++++++++++++++++++++++++++++--- controller/lib/Cargo.toml | 3 +- controller/lib/src/etcd/mod.rs | 133 ++++++++++++++++++++++++++++++++ controller/lib/src/lib.rs | 1 + 4 files changed, 262 insertions(+), 12 deletions(-) create mode 100644 controller/lib/src/etcd/mod.rs diff --git a/Cargo.lock b/Cargo.lock index ac7396a3..0fa83fa3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -484,11 +484,12 @@ name = "controller_lib" version = "0.1.0" dependencies = [ "actix-web", + "etcd-client", "log", "proto", "serde", "tokio", - "tonic", + "tonic 0.7.2", ] [[package]] @@ -636,6 +637,22 @@ dependencies = [ "termcolor", ] +[[package]] +name = "etcd-client" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ddd9c55213f01e9316a52a2f691f30f2aacd0a2a534f87fb84bdc9d5e507ea4" +dependencies = [ + "http", + "prost 0.11.0", + "tokio", + "tokio-stream", + "tonic 0.8.0", + "tonic-build 0.8.0", + "tower", + "tower-service", +] + [[package]] name = "fastrand" version = "1.8.0" @@ -1372,7 +1389,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "71adf41db68aa0daaefc69bb30bcd68ded9b9abaad5d1fbb6304c4fb390e083e" dependencies = [ "bytes", - "prost-derive", + "prost-derive 0.10.1", +] + +[[package]] +name = "prost" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "399c3c31cdec40583bb68f0b18403400d01ec4289c383aa047560439952c4dd7" +dependencies = [ + "bytes", + "prost-derive 0.11.0", ] [[package]] @@ -1390,8 +1417,28 @@ dependencies = [ "log", "multimap", "petgraph", - "prost", - "prost-types", + "prost 0.10.4", + "prost-types 0.10.1", + "regex", + "tempfile", + "which", +] + +[[package]] +name = "prost-build" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f835c582e6bd972ba8347313300219fed5bfa52caf175298d860b61ff6069bb" +dependencies = [ + "bytes", + "heck", + "itertools", + "lazy_static", + "log", + "multimap", + "petgraph", + "prost 0.11.0", + "prost-types 0.11.1", "regex", "tempfile", "which", @@ -1410,6 +1457,19 @@ dependencies = [ "syn", ] +[[package]] +name = "prost-derive" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7345d5f0e08c0536d7ac7229952590239e77abf0a0100a1b1d890add6ea96364" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "prost-types" version = "0.10.1" @@ -1417,16 +1477,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2d0a014229361011dc8e69c8a1ec6c2e8d0f2af7c91e3ea3f5b2170298461e68" dependencies = [ "bytes", - "prost", + "prost 0.10.4", +] + +[[package]] +name = "prost-types" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4dfaa718ad76a44b3415e6c4d53b17c8f99160dcb3a99b10470fce8ad43f6e3e" +dependencies = [ + "bytes", + "prost 0.11.0", ] [[package]] name = "proto" version = "0.1.0" dependencies = [ - "prost", - "tonic", - "tonic-build", + "prost 0.10.4", + "tonic 0.7.2", + "tonic-build 0.7.2", ] [[package]] @@ -1938,8 +2008,40 @@ dependencies = [ "hyper-timeout", "percent-encoding", "pin-project", - "prost", - "prost-derive", + "prost 0.10.4", + "prost-derive 0.10.1", + "tokio", + "tokio-stream", + "tokio-util", + "tower", + "tower-layer", + "tower-service", + "tracing", + "tracing-futures", +] + +[[package]] +name = "tonic" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "498f271adc46acce75d66f639e4d35b31b2394c295c82496727dafa16d465dd2" +dependencies = [ + "async-stream", + "async-trait", + "axum", + "base64", + "bytes", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "hyper", + "hyper-timeout", + "percent-encoding", + "pin-project", + "prost 0.11.0", + "prost-derive 0.11.0", "tokio", "tokio-stream", "tokio-util", @@ -1958,7 +2060,20 @@ checksum = "d9263bf4c9bfaae7317c1c2faf7f18491d2fe476f70c414b73bf5d445b00ffa1" dependencies = [ "prettyplease", "proc-macro2", - "prost-build", + "prost-build 0.10.4", + "quote", + "syn", +] + +[[package]] +name = "tonic-build" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fbcd2800e34e743b9ae795867d5f77b535d3a3be69fd731e39145719752df8c" +dependencies = [ + "prettyplease", + "proc-macro2", + "prost-build 0.11.1", "quote", "syn", ] diff --git a/controller/lib/Cargo.toml b/controller/lib/Cargo.toml index 95d88a39..fec71d0a 100644 --- a/controller/lib/Cargo.toml +++ b/controller/lib/Cargo.toml @@ -6,9 +6,10 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +etcd-client = "0.10.1" actix-web = "4.1.0" serde = { version = "1.0.139", features = ["derive"] } tonic = "0.7.2" proto = { path = "../../proto" } log = "0.4.0" -tokio = { version = "1.20.0", features = ["rt-multi-thread"] } \ No newline at end of file +tokio = { version = "1.20.0", features = ["rt-multi-thread", "macros"] } diff --git a/controller/lib/src/etcd/mod.rs b/controller/lib/src/etcd/mod.rs new file mode 100644 index 00000000..3b8c894e --- /dev/null +++ b/controller/lib/src/etcd/mod.rs @@ -0,0 +1,133 @@ +use etcd_client::{Client, DeleteResponse, Error, GetOptions, PutResponse}; +use log::{debug, error, info}; + +pub struct EtcdClient { + inner: Client, +} + +impl EtcdClient { + pub async fn new(address: String) -> Self { + info!("Starting ETCD client on {}", address); + EtcdClient { + inner: Client::connect([address], None).await.unwrap(), + } + } + + pub async fn get(&mut self, key: &str) -> Result { + if let Some(kv) = self.inner.get(key, None).await?.kvs().first() { + let res = kv.value_str(); + info!( + "Retrieving value in ETCD : Key \"{}\" is associated with value \"{}\"", + key, + res.as_ref().unwrap() + ); + res.map(|s| s.to_string()) + } else { + Err(Error::from(std::io::Error::new( + std::io::ErrorKind::NotFound, + "Key not found", + ))) + } + } + pub async fn put(&mut self, key: &str, value: &str) -> Result { + info!( + "Inserting value in ETCD : Key \"{}\" associated with value \"{}\"", + key, value + ); + self.inner.put(key, value, None).await + } + pub async fn patch(&mut self, key: &str, value: &str) -> Result { + match self.get(key).await { + Ok(_) => { + info!( + "Updating value in ETCD : Key \"{}\" associated with new value \"{}\"", + key, value + ); + self.inner.put(key, value, None).await + } + Err(err) => Err(err), + } + } + pub async fn delete(&mut self, key: &str) -> Option { + if let Some(_) = self.inner.get(key, None).await.unwrap().kvs().first() { + info!("Deleting value in ETCD : Key \"{}\" ", key); + match self.inner.delete(key, None).await { + Ok(delete) => { + Some(delete) + } + Err(_) => None + } + } else { + None + } + } + + pub async fn get_all(&mut self) -> Result, Error> { + info!("Retrieving all keys in ETCD"); + let resp = self + .inner + .get("", Some(GetOptions::new().with_all_keys())) + .await?; + + let mut values: Vec = vec![]; + for kv in resp.kvs() { + let value = kv.value_str()?; + values.push(value.to_string()) + } + Ok(values) + } +} + +#[cfg(test)] +mod tests { + + use crate::etcd::EtcdClient; + use etcd_client::Error; + + #[tokio::test] + async fn test_value_insertion() -> Result<(), Error> { + let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await; + let _res = etcd_client.put("foo", "bar").await?; + let resp = etcd_client.get("foo").await?; + assert_eq!(resp, "bar"); + Ok(()) + } + + #[tokio::test] + async fn test_value_modification() -> Result<(), Error> { + let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await; + etcd_client.put("foo", "bar").await?; + let _res = etcd_client.patch("foo", "baz").await?; + let resp = etcd_client.get("foo").await?; + assert_eq!(resp, "baz"); + Ok(()) + } + #[tokio::test] + async fn test_value_deletion() -> Result<(), Error> { + let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await; + let _res = etcd_client.put("foo", "bar").await?; + let _res = etcd_client.delete("foo").await?; + let err = etcd_client.get("foo").await; + assert!(err.is_err()); + Ok(()) + } + #[tokio::test] + async fn test_value_deletion_doesnt_exists() -> Result<(), Error> { + let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await; + let _res = etcd_client.put("foo", "bar").await?; + let err = etcd_client.delete("foo2").await; + assert!(err.is_err()); + Ok(()) + } + + #[tokio::test] + async fn test_function_get_all() -> Result<(), Error> { + let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await; + let _res = etcd_client.put("bar", "foo").await; + let _res = etcd_client.put("hello", "world").await; + let values = etcd_client.get_all().await?; + assert_eq!(values[0], "foo"); + assert_eq!(values[1], "world"); + Ok(()) + } +} diff --git a/controller/lib/src/lib.rs b/controller/lib/src/lib.rs index 91866edd..cf6d0ae9 100644 --- a/controller/lib/src/lib.rs +++ b/controller/lib/src/lib.rs @@ -1,2 +1,3 @@ +pub mod etcd; pub mod external_api; pub mod internal_api;