Feature/config via env variables (#43)

* Add example service back to cargo.toml

* Add env variables to overwrite config and remove Result from config

* update readme and changelog

* Add test

* Fix DLQ

* cargo fmt

Author: toelo3

CHANGELOG.md

@@ -14,10 +14,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Add Default implementation for Dsh::Properties
   - Points to localhost:9092 for kafka, localhost:8081 for schemastore
-  - local_datastreams.json is now optional
+  - local_datastreams.json is now optional as it falssback to default values
+- Overwrite Kafka config via environment variables for producer and consumer
 - Add extra check in github actions to check for compile issues for all features independently
 
 ### Changed
+- **Breaking change:** consumer_rdkafka_config and producer_rdkafka_config returns `ClientConfig` instead of `Result<ClientConfig>`
+- **Breaking change:** certificates and keys are now returned as `T` instead of `Result<T>` 
 - **Breaking change:** Private key is based on ECDSA instead of RSA 
 - **Breaking change:** Error enum is now non_exhaustive
 

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "dsh_sdk",
+    "example_dsh_service",
 ]
 readme = "README.md"
 resolver = "2"

dsh_sdk/Cargo.toml

@@ -45,4 +45,5 @@ rdkafka-ssl = ["rdkafka", "rdkafka/ssl"]
 mockito = "1.1.1"
 openssl = "0.10"
 tokio = "^1.35"
-hyper = { version = "1.2.0", features = ["full"]}
+hyper = { version = "1.2.0", features = ["full"]}
+serial_test = "3.1.0"

dsh_sdk/README.md

@@ -42,7 +42,7 @@ use dsh_sdk::rdkafka::consumer::{Consumer, StreamConsumer};
 fn main() -> Result<(), Box<dyn std::error::Error>>{
     let dsh_properties = Properties::get();
     // get a rdkafka consumer config for example
-    let consumer: StreamConsumer = dsh_properties.consumer_rdkafka_config()?.create()?;
+    let consumer: StreamConsumer = dsh_properties.consumer_rdkafka_config().create()?;
 }
 ```
 
@@ -53,7 +53,6 @@ The following features are available in this library and can be enabled/disabled
 | **feature** | **default** | **Description** |
 |---|---|---|
 | `bootstrap` | &check; | Generate signed certificate and fetch datastreams info <br> Also makes certificates available, to be used as lowlevel API |
-| ~~`local`~~ | &check; | Will be deprecated in v0.4.0 |
 | `metrics` | &check; | Enable (custom) metrics for your service |
 | `graceful_shutdown` | &check; | Create a signal handler for implementing a graceful shutdown |
 | `dlq` | &cross; | Dead Letter Queue implementation (experimental) |
@@ -62,11 +61,43 @@ The following features are available in this library and can be enabled/disabled
 
 See api documentation for more information on how to use these features including.
 
+## Environment variables
+The default RDKafka config can be overwritten by setting the following environment variables:
+
+### `KAFKA_BOOTSTRAP_SERVERS`
+- Usage: Overwrite hostnames of brokers (useful for local testing)
+- Default: Brokers based on datastreams
+- Required: `false`
+
+### `KAFKA_CONSUMER_GROUP_TYPE`
+- Usage: Picks group_id based on type from datastreams
+- Default: Shared
+- Options: private, shared
+- Required: `false`
+
+### `KAFKA_GROUP_ID`
+- Usage: Custom group id
+- Default: NA
+- Required: `false`
+- Remark: Overrules `KAFKA_CONSUMER_GROUP_TYPE`. Mandatory to start with tenant name. (will prefix tenant name automatically if not set)
+
+### `KAFKA_ENABLE_AUTO_COMMIT`
+- Usage: Enable/Disable auto commit
+- Default: `false`
+- Required: `false`
+- Options: `true`, `false`
+
+### `KAFKA_AUTO_OFFSET_RESET`
+- Usage: Set the offset reset settings to start consuming from set option.
+- Default: earliest
+- Required: `false`
+- Options: smallest, earliest, beginning, largest, latest, end
+
 ## Api doc
 See the [api documentation](https://docs.rs/dsh_sdk/latest/dsh_sdk/) for more information on how to use this library.
 
 ### Local development
-You can start the [docker-compose](../docker/docker-compose.yml) file in the root of this project to start a local Kafka broker and Schema Registry.
+You can start the [docker-compose](../docker/docker-compose.yml) file to start a local Kafka broker and Schema Registry.
 
 When running the SDK on your local machine, it will automatically try to connect to the local Kafka broker and Schema Registry
 
@@ -75,7 +106,7 @@ When running the SDK on your local machine, it will automatically try to connect
 | Kafka | `localhost:9092` |
 | Schema Registry | `localhost:8081/apis/ccompat/v7` |
 
-If you want manipulate these endpoints, or want to use specific datastream info, you can add a [local_datastreams.json](local_datastreams.json) to your project root to overwrite the default values.
+If you want manipulate these endpoints, or want to use specific datastream info, you can add a [local_datastreams.json](local_datastreams.json) to your project root to overwrite the default values or set the environment variables accordingly.
 
 ### Note
 Rdkafka and thereby this library is dependent on CMAKE. Make sure it is installed in your environment and/or Dockerfile where you are compiling.

dsh_sdk/examples/produce_consume.rs

@@ -46,13 +46,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let topic = "scratch.local.local-tenant";
 
     // Create a new producer based on the properties default config
-    let mut producer: FutureProducer = dsh_properties.producer_rdkafka_config()?.create()?;
+    let mut producer: FutureProducer = dsh_properties.producer_rdkafka_config().create()?;
 
     // Produce messages towards topic
     produce(&mut producer, topic).await;
 
     // Create a new consumer based on the properties default config
-    let mut consumer: StreamConsumer = dsh_properties.consumer_rdkafka_config()?.create()?;
+    let mut consumer: StreamConsumer = dsh_properties.consumer_rdkafka_config().create()?;
 
     consume(&mut consumer, topic).await;
     Ok(())

dsh_sdk/src/dlq.rs

@@ -230,17 +230,7 @@ impl Dlq {
     }
 
     fn build_producer(dsh_prop: &Properties) -> Result<FutureProducer, rdkafka::error::KafkaError> {
-        let producer_config = match dsh_prop.producer_rdkafka_config() {
-            Ok(config) => config,
-            Err(e) => {
-                error!("Error creating producer config");
-                return Err(rdkafka::error::KafkaError::ClientCreation(format!(
-                    "Error creating producer config: {}",
-                    e
-                )));
-            }
-        };
-        producer_config.create()
+        dsh_prop.producer_rdkafka_config().create()
     }
 }
 

dsh_sdk/src/dsh/certificates.rs

@@ -76,7 +76,7 @@ impl Cert {
     pub fn reqwest_client_config(&self) -> Result<reqwest::ClientBuilder, DshError> {
         let pem_identity = Cert::create_identity(
             self.dsh_kafka_certificate_pem().as_bytes(),
-            self.private_key_pem()?.as_bytes(),
+            self.private_key_pem().as_bytes(),
         )?;
         let reqwest_cert =
             reqwest::tls::Certificate::from_pem(self.dsh_ca_certificate_pem().as_bytes())?;
@@ -91,7 +91,7 @@ impl Cert {
     pub fn reqwest_blocking_client_config(&self) -> Result<ClientBuilder, DshError> {
         let pem_identity = Cert::create_identity(
             self.dsh_kafka_certificate_pem().as_bytes(),
-            self.private_key_pem()?.as_bytes(),
+            self.private_key_pem().as_bytes(),
         )?;
         let reqwest_cert =
             reqwest::tls::Certificate::from_pem(self.dsh_ca_certificate_pem().as_bytes())?;
@@ -112,23 +112,23 @@ impl Cert {
     }
 
     /// Get the private key as PKCS8 and return bytes based on asn1 DER format.
-    pub fn private_key_pkcs8(&self) -> Result<Vec<u8>, DshError> {
-        Ok(self.key_pair.serialize_der())
+    pub fn private_key_pkcs8(&self) -> Vec<u8> {
+        self.key_pair.serialize_der()
     }
 
     /// Get the private key as PEM string. Equivalent to client.key.
-    pub fn private_key_pem(&self) -> Result<String, DshError> {
-        Ok(self.key_pair.serialize_pem())
+    pub fn private_key_pem(&self) -> String {
+        self.key_pair.serialize_pem()
     }
 
     /// Get the public key as PEM string.
-    pub fn public_key_pem(&self) -> Result<String, DshError> {
-        Ok(self.key_pair.public_key_pem())
+    pub fn public_key_pem(&self) -> String {
+        self.key_pair.public_key_pem()
     }
 
     /// Get the public key as DER bytes.
-    pub fn public_key_der(&self) -> Result<Vec<u8>, DshError> {
-        Ok(self.key_pair.public_key_der())
+    pub fn public_key_der(&self) -> Vec<u8> {
+        self.key_pair.public_key_der()
     }
 
     /// Create the ca.crt, client.pem, and client.key files in a desired directory.
@@ -152,7 +152,7 @@ impl Cert {
         std::fs::create_dir_all(dir)?;
         Self::create_file(dir.join("ca.crt"), self.dsh_ca_certificate_pem())?;
         Self::create_file(dir.join("client.pem"), self.dsh_kafka_certificate_pem())?;
-        Self::create_file(dir.join("client.key"), self.private_key_pem()?)?;
+        Self::create_file(dir.join("client.key"), self.private_key_pem())?;
         Ok(())
     }
 
@@ -212,7 +212,7 @@ mod tests {
         let pkey = PKey::private_key_from_der(der.as_slice()).unwrap();
         let pkey_pem_bytes = pkey.private_key_to_pem_pkcs8().unwrap();
 
-        let key_pem = cert.private_key_pem().unwrap();
+        let key_pem = cert.private_key_pem();
         let pkey_pem = String::from_utf8_lossy(pkey_pem_bytes.as_slice());
         assert_eq!(key_pem, pkey_pem);
     }
@@ -224,7 +224,7 @@ mod tests {
         let pkey = PKey::private_key_from_der(der.as_slice()).unwrap();
         let pkey_pub_pem_bytes = pkey.public_key_to_pem().unwrap();
 
-        let pub_pem = cert.public_key_pem().unwrap();
+        let pub_pem = cert.public_key_pem();
         let pkey_pub_pem = String::from_utf8_lossy(pkey_pub_pem_bytes.as_slice());
         assert_eq!(pub_pem, pkey_pub_pem);
     }
@@ -236,7 +236,7 @@ mod tests {
         let pkey = PKey::private_key_from_der(der.as_slice()).unwrap();
         let pkey_pub_der = pkey.public_key_to_der().unwrap();
 
-        let pub_der = cert.public_key_der().unwrap();
+        let pub_der = cert.public_key_der();
         assert_eq!(pub_der, pkey_pub_der);
     }
 
@@ -247,7 +247,7 @@ mod tests {
         let pkey = PKey::private_key_from_der(der.as_slice()).unwrap();
         let pkey = pkey.private_key_to_pkcs8().unwrap();
 
-        let key = cert.private_key_pkcs8().unwrap();
+        let key = cert.private_key_pkcs8();
         assert_eq!(key, pkey);
     }
 
@@ -292,7 +292,7 @@ mod tests {
         let dn = Dn::parse_string("CN=Test CN,OU=Test OU,O=Test Org").unwrap();
         let csr = Cert::generate_csr(&cert.key_pair, dn).unwrap();
         let csr_pem = csr.pem().unwrap();
-        let key = cert.private_key_pkcs8().unwrap();
+        let key = cert.private_key_pkcs8();
         let pkey = PKey::private_key_from_der(&key).unwrap();
 
         let req = X509Req::from_pem(csr_pem.as_bytes()).unwrap();

dsh_sdk/src/dsh/datastream.rs

@@ -134,7 +134,7 @@ impl Datastream {
         let mut file = match file_result {
             Ok(file) => file,
             Err(e) => {
-                error!(
+                warn!(
                     "Error opening local_datastreams.json ({}): {}",
                     path_buf.display(),
                     e