diff --git a/.env-example b/.env-example index 1f0a3bd..09b30b8 100644 --- a/.env-example +++ b/.env-example @@ -1,6 +1,9 @@ ENVIRONMENT='development' -MODE='dynamic' -REDIS_URL='redis://localhost:6379' -RABBITMQ_URL='amqp://localhost:5672' +MODE='static' +NATS_URL='nats://localhost:4222' +NATS_BUCKET='flow_store' +GRPC_HOST='127.0.0.1' +GRPC_PORT=8081 +WITH_HEALTH_SERVICE=false AQUILA_URL='http://localhost:8080' -PORT=8080 +DEFINITION_PATH='./definitions' \ No newline at end of file diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml deleted file mode 100644 index 81b7fe5..0000000 --- a/.github/workflows/build-and-test.yml +++ /dev/null @@ -1,25 +0,0 @@ -name: Build & Test Draco - -on: - push: - -jobs: - draco: - runs-on: ubuntu-latest - - defaults: - run: - shell: bash - - steps: - - uses: actions/checkout@v5 - - name: Setup rust - run: rustup update --no-self-update stable - - name: Build crate - run: PATH=${{ runner.temp }}/proto/bin:$PATH cargo build - env: - RUST_BACKTRACE: 'full' - - name: Run Tests - run: PATH=${{ runner.temp }}/proto/bin:$PATH cargo test - env: - RUST_BACKTRACE: 'full' diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..b27626c --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,104 @@ +name: Rust CI/CD Pipeline + +on: + push: + pull_request: + types: [opened, reopened] + +env: + RUST_BACKTRACE: 1 + +jobs: + # Job 1: Format Check + format: + name: Format Check + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v5 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt + + - name: Check formatting + run: cargo fmt --all -- --check + + # Job 2: Lint Check + lint: + name: Lint Check + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v5 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: clippy + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-lint-${{ hashFiles('**/Cargo.lock') }} + restore-keys: | + ${{ runner.os }}-cargo- + + - name: Run clippy + run: cargo clippy --all-targets --all-features + + # Job 3: Build + build: + name: Build + needs: [ lint, format ] + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v5 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }} + restore-keys: | + ${{ runner.os }}-cargo- + + - name: Build project + run: cargo build --verbose --all-features + + test: + name: Test + needs: build + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v5 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }} + restore-keys: | + ${{ runner.os }}-cargo- + + - name: Run Tests + run: cargo test --verbose --all-features diff --git a/Cargo.lock b/Cargo.lock index 278120b..94d7bac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -210,6 +210,7 @@ dependencies = [ "async-trait", "code0-flow", "futures-lite", + "log", "prost", "tokio", "tonic", diff --git a/adapter/rest/src/.env b/adapter/rest/src/.env index 9e508bb..9e58e86 100644 --- a/adapter/rest/src/.env +++ b/adapter/rest/src/.env @@ -1,5 +1,11 @@ -HTTP_PORT=8081 -REDIS_URL=redis://localhost:6379 -RABBITMQ_URL=amqp://localhost:5672 -AQUILA_URL=http://localhost:8080 -IS_STATIC=false +HTTP_SERVER_PORT=8080 + +ENVIRONMENT='development' +MODE='static' +NATS_URL='nats://localhost:4222' +NATS_BUCKET='flow_store' +GRPC_HOST='127.0.0.1' +GRPC_PORT=8081 +WITH_HEALTH_SERVICE=false +AQUILA_URL='http://localhost:8080' +DEFINITION_PATH='./definitions' \ No newline at end of file diff --git a/adapter/rest/src/main.rs b/adapter/rest/src/main.rs index 11bb7cc..92bb20a 100644 --- a/adapter/rest/src/main.rs +++ b/adapter/rest/src/main.rs @@ -1,7 +1,7 @@ use base::{ extract_flow_setting_field, runner::{ServerContext, ServerRunner}, - store::FlowIdenfiyResult, + store::FlowIdentifyResult, traits::{IdentifiableFlow, LoadConfig, Server as ServerTrait}, }; use code0_flow::flow_config::env_with_default; @@ -64,7 +64,7 @@ impl ServerTrait for HttpServer { }; match store.get_possible_flow_match(pattern, route).await { - FlowIdenfiyResult::Single(flow) => { + FlowIdentifyResult::Single(flow) => { execute_flow(flow, request, store).await } _ => Some(HttpResponse::internal_server_error( diff --git a/crates/base/Cargo.toml b/crates/base/Cargo.toml index 4cd1517..0ba2d79 100644 --- a/crates/base/Cargo.toml +++ b/crates/base/Cargo.toml @@ -15,3 +15,4 @@ tonic-health = { workspace = true } uuid = { workspace = true } prost = { workspace = true } futures-lite = { workspace = true } +log = { workspace = true } diff --git a/crates/base/src/config.rs b/crates/base/src/config.rs index 8a4da21..27ed1f0 100644 --- a/crates/base/src/config.rs +++ b/crates/base/src/config.rs @@ -32,6 +32,11 @@ pub struct AdapterConfig { /// Port on which the adapter's Health Service server will listen. pub grpc_port: u16, + /// GRPC Host + /// + /// Host on which the adapter's Health Service server will listen. + pub grpc_host: String, + /// Aquila URL /// /// URL of the Aquila server to connect to. @@ -45,7 +50,7 @@ pub struct AdapterConfig { /// Is Monitored /// /// If true the Adapter will expose a grpc health service server. - pub is_monitored: bool, + pub with_health_service: bool, } impl AdapterConfig { @@ -57,6 +62,8 @@ impl AdapterConfig { let nats_bucket = code0_flow::flow_config::env_with_default("NATS_BUCKET", String::from("flow_store")); let grpc_port = code0_flow::flow_config::env_with_default("GRPC_PORT", 50051); + let grpc_host = + code0_flow::flow_config::env_with_default("GRPC_HOST", String::from("localhost")); let aquila_url = code0_flow::flow_config::env_with_default( "AQUILA_URL", String::from("grpc://localhost:50051"), @@ -67,9 +74,10 @@ impl AdapterConfig { let mode = code0_flow::flow_config::env_with_default("MODE", Mode::STATIC); let definition_path = code0_flow::flow_config::env_with_default( "DEFINITION_PATH", - String::from("./definition.yaml"), + String::from("./definition"), ); - let is_monitored = code0_flow::flow_config::env_with_default("IS_MONITORED", false); + let with_health_service = + code0_flow::flow_config::env_with_default("WITH_HEALTH_SERVICE", false); Self { environment, @@ -77,9 +85,10 @@ impl AdapterConfig { mode, nats_url, grpc_port, + grpc_host, aquila_url, definition_path, - is_monitored, + with_health_service, } } diff --git a/crates/base/src/lib.rs b/crates/base/src/lib.rs index 5765fda..b85986a 100644 --- a/crates/base/src/lib.rs +++ b/crates/base/src/lib.rs @@ -17,10 +17,10 @@ pub fn extract_flow_setting_field( let obj = setting.object.as_ref()?; obj.fields.iter().find_map(|(k, v)| { - if k == field_name { - if let Some(Kind::StringValue(s)) = &v.kind { - return Some(s.clone()); - } + if k == field_name + && let Some(Kind::StringValue(s)) = &v.kind + { + return Some(s.clone()); } None }) diff --git a/crates/base/src/runner.rs b/crates/base/src/runner.rs index d449353..38020a6 100644 --- a/crates/base/src/runner.rs +++ b/crates/base/src/runner.rs @@ -61,10 +61,10 @@ impl ServerRunner { definition_service.send().await; } - if config.is_monitored { + if config.with_health_service { let health_service = code0_flow::flow_health::HealthService::new(config.nats_url.clone()); - let address = format!("127.0.0.1:{}", config.grpc_port).parse()?; + let address = format!("{}:{}", config.grpc_host, config.grpc_port).parse()?; tokio::spawn(async move { let _ = Server::builder() @@ -73,7 +73,11 @@ impl ServerRunner { .await; }); - println!("Health server started at 127.0.0.1:{}", config.grpc_port); + log::info!( + "Health server started at {}:{}", + config.grpc_host, + config.grpc_port + ); } self.server.init(&self.context).await?; diff --git a/crates/base/src/store.rs b/crates/base/src/store.rs index d73bd75..8b0e941 100644 --- a/crates/base/src/store.rs +++ b/crates/base/src/store.rs @@ -1,16 +1,16 @@ use crate::traits::IdentifiableFlow; use async_nats::jetstream::kv::Config; +use code0_flow::flow_validator::verify_flow; use futures_lite::StreamExt; use prost::Message; use tucana::shared::{ExecutionFlow, ValidationFlow, Value}; -use code0_flow::flow_validator::verify_flow; pub struct AdapterStore { client: async_nats::Client, kv: async_nats::jetstream::kv::Store, } -pub enum FlowIdenfiyResult { +pub enum FlowIdentifyResult { None, Single(ValidationFlow), Multiple(Vec), @@ -20,19 +20,25 @@ impl AdapterStore { pub async fn from_url(url: String, bucket: String) -> Self { let client = match async_nats::connect(url).await { Ok(client) => client, - Err(err) => panic!("Failed to connect to NATS server: {}", err), + Err(err) => panic!("Failed to connect to NATS server: {:?}", err), }; - let jetstream = async_nats::jetstream::new(client.clone()); + let stream = async_nats::jetstream::new(client.clone()); - let _ = jetstream + match stream .create_key_value(Config { bucket: bucket.clone(), ..Default::default() }) - .await; + .await + { + Ok(_) => { + log::info!("Successfully created NATS bucket"); + } + Err(err) => panic!("Failed to create NATS bucket: {:?}", err), + } - let kv = match jetstream.get_key_value(bucket).await { + let kv = match stream.get_key_value(bucket).await { Ok(kv) => kv, Err(err) => panic!("Failed to get key-value store: {}", err), }; @@ -63,38 +69,35 @@ impl AdapterStore { &self, pattern: String, id: I, - ) -> FlowIdenfiyResult { + ) -> FlowIdentifyResult { let mut collector = Vec::new(); let mut keys = match self.kv.keys().await { Ok(keys) => keys.boxed(), Err(err) => { - eprintln!("Failed to get keys: {}", err); - return FlowIdenfiyResult::None; + log::error!("Failed to get keys: {}", err); + return FlowIdentifyResult::None; } }; while let Ok(Some(key)) = keys.try_next().await { - println!("comparing: key: {} pattern {:?}", key, pattern); - if !Self::is_matching_key(&pattern, &key) { - println!("Key does not match pattern: {}", key); continue; } if let Ok(Some(bytes)) = self.kv.get(key).await { let decoded_flow = ValidationFlow::decode(bytes); - if let Ok(flow) = decoded_flow { - if id.identify(&flow) { - collector.push(flow); - } + if let Ok(flow) = decoded_flow + && id.identify(&flow) + { + collector.push(flow.clone()); }; } } match collector.len() { - 0 => FlowIdenfiyResult::None, - 1 => FlowIdenfiyResult::Single(collector[0].clone()), - _ => FlowIdenfiyResult::Multiple(collector), + 0 => FlowIdentifyResult::None, + 1 => FlowIdentifyResult::Single(collector[0].clone()), + _ => FlowIdentifyResult::Multiple(collector), } } @@ -129,17 +132,14 @@ impl AdapterStore { match result { Ok(message) => match Value::decode(message.payload) { - Ok(value) => { - println!("Response: {:?}", &value); - Some(value) - } + Ok(value) => Some(value), Err(err) => { - eprintln!("Failed to decode response from NATS server: {}", err); - return None; + log::error!("Failed to decode response from NATS server: {:?}", err); + None } }, Err(err) => { - eprintln!("Failed to send request to NATS server: {}", err); + log::error!("Failed to send request to NATS server: {:?}", err); None } } @@ -149,15 +149,14 @@ impl AdapterStore { ExecutionFlow { flow_id: flow.flow_id, starting_node: flow.starting_node, - input_value: input_value, + input_value, } } fn is_matching_key(pattern: &String, key: &String) -> bool { - let splitted_pattern = pattern.split("."); - let splitted_key = key.split(".").collect::>(); - - let zip = splitted_pattern.into_iter().zip(splitted_key); + let split_pattern = pattern.split("."); + let split_key = key.split(".").collect::>(); + let zip = split_pattern.into_iter().zip(split_key); for (pattern_part, key_part) in zip { if pattern_part == "*" { @@ -165,11 +164,9 @@ impl AdapterStore { } if pattern_part != key_part { - println!("matching: pattern: {} key: {}", pattern_part, key_part); return false; } } - println!("pattern was correct"); true } } diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index 591db4d..b770771 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -46,6 +46,12 @@ pub struct HeaderMap { pub fields: HashMap, } +impl Default for HeaderMap { + fn default() -> Self { + Self::new() + } +} + impl HeaderMap { pub fn new() -> Self { HeaderMap { @@ -228,10 +234,10 @@ fn parse_request( }; let mut body = vec![0; size]; - if let Ok(_) = buf_reader.read_exact(&mut body) { - if let Ok(json_value) = serde_json::from_slice::(&body) { - body_values.insert("body".to_string(), from_json_value(json_value)); - } + if buf_reader.read_exact(&mut body).is_ok() + && let Ok(json_value) = serde_json::from_slice::(&body) + { + body_values.insert("body".to_string(), from_json_value(json_value)); } };