diff --git a/.env-example b/.env-example index f4e39c6..ffa195b 100644 --- a/.env-example +++ b/.env-example @@ -1,19 +1,6 @@ -ENVIRONMENT='production' -MODE='dynamic' +# Optional configuration file locations +# AQUILA_CONFIG_PATH=./aquila.yml +# AQUILA_SERVICE_CONFIG_PATH=./service.configuration.json -NATS_URL='nats://localhost:4222' -NATS_BUCKET='flow_store' - -RUNTIME_TOKEN="" -FLOW_FALLBACK_PATH='./path/to/flow.json' - -GRPC_HOST='127.0.0.1' -GRPC_PORT=50051 - -WITH_HEALTH_SERVICE=false - -SAGITTARIUS_URL='http://localhost:50051' - -RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS=90 -RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS=180 -RUNTIME_STATUS_MONITOR_INTERVAL_SECS=30 +# Optional secret override for dynamic_config.backend_token in aquila.yml +AQUILA_BACKEND_TOKEN= diff --git a/Cargo.lock b/Cargo.lock index f70c938..696bfca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,6 +73,7 @@ version = "0.1.0" dependencies = [ "async-nats", "code0-flow", + "config", "env_logger", "futures", "futures-core", @@ -88,6 +89,12 @@ dependencies = [ "uuid", ] +[[package]] +name = "arraydeque" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" + [[package]] name = "async-nats" version = "0.49.1" @@ -207,6 +214,9 @@ name = "bitflags" version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" +dependencies = [ + "serde_core", +] [[package]] name = "block-buffer" @@ -295,12 +305,61 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "config" +version = "0.15.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b85f248a4de22d204ceabc6299d89d2c70fbd7f09fea53c06c852369652d8139" +dependencies = [ + "async-trait", + "convert_case", + "json5", + "pathdiff", + "ron", + "rust-ini", + "serde-untagged", + "serde_core", + "serde_json", + "toml", + "winnow", + "yaml-rust2", +] + [[package]] name = "const-oid" version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "tiny-keccak", +] + +[[package]] +name = "convert_case" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec182b0ca2f35d8fc196cf3404988fd8b8c739a4d270ff118a398feb0cbec1ca" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -335,6 +394,12 @@ dependencies = [ "libc", ] +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-common" version = "0.1.7" @@ -419,6 +484,15 @@ dependencies = [ "syn", ] +[[package]] +name = "dlv-list" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "442039f5147480ba31067cb00ada1adae6892028e40e45fc5de7b7df6dcc1b5f" +dependencies = [ + "const-random", +] + [[package]] name = "dotenv" version = "0.15.0" @@ -453,6 +527,15 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "env_filter" version = "2.0.0" @@ -482,6 +565,17 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "erased-serde" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2add8a07dd6a8d93ff627029c51de145e12686fbc36ecb298ac22e74cf02dec" +dependencies = [ + "serde", + "serde_core", + "typeid", +] + [[package]] name = "errno" version = "0.3.14" @@ -489,7 +583,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -528,6 +622,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -679,13 +779,19 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + [[package]] name = "hashbrown" version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "foldhash", + "foldhash 0.1.5", ] [[package]] @@ -693,6 +799,18 @@ name = "hashbrown" version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "foldhash 0.2.0", +] + +[[package]] +name = "hashlink" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824e001ac4f3012dd16a264bec811403a67ca9deb6c102fc5049b32c4574b35f" +dependencies = [ + "hashbrown 0.16.1", +] [[package]] name = "heck" @@ -976,6 +1094,17 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "json5" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96b0db21af676c1ce64250b5f40f3ce2cf27e4e47cb91ed91eb6fe9350b430c1" +dependencies = [ + "pest", + "pest_derive", + "serde", +] + [[package]] name = "leb128fmt" version = "0.1.0" @@ -1098,6 +1227,22 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "ordered-multimap" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49203cdcae0030493bad186b28da2fa25645fa276a51b6fec8010d281e02ef79" +dependencies = [ + "dlv-list", + "hashbrown 0.14.5", +] + +[[package]] +name = "pathdiff" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" + [[package]] name = "pbjson" version = "0.9.0" @@ -1150,6 +1295,48 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pest" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47627dd7305c6a2d6c8c6bcd24c5a4c17dbbf425f4f9c5313e724b38fc9782e9" +dependencies = [ + "memchr", + "ucd-trie", +] + +[[package]] +name = "pest_derive" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b4254325ecad416ab689e27ba51da03ba01a9632bc6e108f5fe7c3c4ad29d58" +dependencies = [ + "pest", + "pest_generator", +] + +[[package]] +name = "pest_generator" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c4c0e91ead7a8f7acecbca6f003fc2e8282b1dbe2dd9c9d2f16aba42995e0a7" +dependencies = [ + "pest", + "pest_meta", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "pest_meta" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9744bc48116fee06334924bb5f2bad41eed5e89bd26e29b0b799f9a3f82c210" +dependencies = [ + "pest", +] + [[package]] name = "petgraph" version = "0.8.3" @@ -1433,6 +1620,30 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "ron" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81116b9531d61eabc41aeb228e4b6b2435bcca3233b98cf3b3077d4e6e9debb3" +dependencies = [ + "bitflags", + "once_cell", + "serde", + "serde_derive", + "typeid", + "unicode-ident", +] + +[[package]] +name = "rust-ini" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "796e8d2b6696392a43bea58116b667fb4c29727dc5abd27d6acf338bb4f688c7" +dependencies = [ + "cfg-if", + "ordered-multimap", +] + [[package]] name = "rustc_version" version = "0.4.1" @@ -1452,7 +1663,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1564,6 +1775,18 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-untagged" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9faf48a4a2d2693be24c6289dbe26552776eb7737074e6722891fadbe6c5058" +dependencies = [ + "erased-serde", + "serde", + "serde_core", + "typeid", +] + [[package]] name = "serde_core" version = "1.0.228" @@ -1617,6 +1840,15 @@ dependencies = [ "syn", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "sha2" version = "0.10.9" @@ -1748,7 +1980,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1802,6 +2034,15 @@ dependencies = [ "time-core", ] +[[package]] +name = "tiny-keccak" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" +dependencies = [ + "crunchy", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -1895,6 +2136,37 @@ dependencies = [ "webpki-roots 0.26.11", ] +[[package]] +name = "toml" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81f3d15e84cbcd896376e6730314d59fb5a87f31e4b038454184435cd57defee" +dependencies = [ + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "winnow", +] + +[[package]] +name = "toml_datetime" +version = "1.1.1+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" +dependencies = [ + "serde_core", +] + +[[package]] +name = "toml_parser" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +dependencies = [ + "winnow", +] + [[package]] name = "tonic" version = "0.14.6" @@ -2074,12 +2346,24 @@ dependencies = [ "tonic-prost-build", ] +[[package]] +name = "typeid" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" + [[package]] name = "typenum" version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" +[[package]] +name = "ucd-trie" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" + [[package]] name = "unicase" version = "2.9.0" @@ -2092,6 +2376,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -2291,7 +2581,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2382,6 +2672,15 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.51.0" @@ -2476,6 +2775,17 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "yaml-rust2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "631a50d867fafb7093e709d75aaee9e0e0d5deb934021fcea25ac2fe09edc51e" +dependencies = [ + "arraydeque", + "encoding_rs", + "hashlink", +] + [[package]] name = "yoke" version = "0.8.2" diff --git a/Cargo.toml b/Cargo.toml index c363d1b..0a54e7c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,3 +19,4 @@ tokio-stream = "0.1.17" uuid = { version = "1.18.0", features = ["v4"] } serde = "1.0.228" futures-core = "0.3.32" +config = "0.15.25" diff --git a/aquila.yml b/aquila.yml new file mode 100644 index 0000000..0e05d3d --- /dev/null +++ b/aquila.yml @@ -0,0 +1,51 @@ +# Deployment environment sent to connected services. +# Valid values: development, staging, production. +environment: development + +# Startup mode. Static loads flows from a local file; dynamic synchronizes with Sagittarius. +# Valid values: static, dynamic. +mode: static + +# Default env_logger filter. RUST_LOG takes precedence when it is set. +log_level: debug + +# NATS server and JetStream key-value bucket used for flow storage. +nats: + url: nats://localhost:4222 + bucket: flow_store + +# Settings used only in static mode. +static_config: + # JSON flow export loaded during startup. + flow_path: ./flowExport.json + +# Settings used only in dynamic mode. +dynamic_config: + # Sagittarius gRPC endpoint. + backend_url: http://localhost:50051 + + # Authentication token for Sagittarius. + # AQUILA_BACKEND_TOKEN overrides this value and should be used for deployed secrets. + backend_token: default_session_token + + # Timeout for unary Sagittarius RPCs, in seconds. + backend_unary_timeout_secs: 5 + +# Aquila gRPC server bind settings. +grpc: + host: 127.0.0.1 + port: 8081 + + # Expose the standard gRPC health service. + health_service: false + +# Runtime heartbeat state-transition timing, in seconds. +runtime_status: + # Time without a heartbeat before a runtime becomes NOT_RESPONDING. + not_responding_after_secs: 90 + + # Additional time in NOT_RESPONDING before a runtime becomes STOPPED. + stopped_after_not_responding_secs: 180 + + # Frequency of runtime heartbeat checks. + monitor_interval_secs: 30 diff --git a/docs/dev.md b/docs/dev.md index daeec71..d03b83f 100644 --- a/docs/dev.md +++ b/docs/dev.md @@ -124,4 +124,4 @@ Aquila --> NATS ``` > In static mode, flow data and service authorization are loaded from local -files (for example `FLOW_FALLBACK_PATH` and `SERVICE_CONFIG_PATH`) before runtime execution starts. +files (for example `static_config.flow_path` and `AQUILA_SERVICE_CONFIG_PATH`) before runtime execution starts. diff --git a/docs/installation.mdx b/docs/installation.mdx index 159b431..5ab807d 100644 --- a/docs/installation.mdx +++ b/docs/installation.mdx @@ -35,10 +35,10 @@ Clone this repository to your local machine. -#### **Set up environment variables** +#### **Configure Aquila** -Configure `.env` in the project root with the required settings. -You can use `.env-example` as a starting point. +Configure `aquila.yml` in the project root. To keep the Sagittarius token out of the file, set +`AQUILA_BACKEND_TOKEN` in the environment or a local `.env` file. @@ -54,7 +54,7 @@ You can use `.env-example` as a starting point. **Sagittarius**: -- Required for dynamic mode (`MODE=dynamic`) +- Required for dynamic mode (`mode: dynamic`) - Ensure the configured endpoint is reachable @@ -74,57 +74,76 @@ cargo run --- -## Environment Variables +## Configuration -Aquila configuration is split into shared variables and mode-specific variables. +Aquila reads `aquila.yml` from the working directory. Built-in defaults are used when the file is +absent. Configuration values are read from this file, except that `AQUILA_BACKEND_TOKEN` can +override `dynamic_config.backend_token` so the token can be injected as a secret. + +Select a different configuration file with `AQUILA_CONFIG_PATH`: + +```bash +AQUILA_CONFIG_PATH=/path/to/aquila.yml cargo run +``` + +The `log_level` setting controls the default log level. The standard `RUST_LOG` variable can still +provide a more specific runtime filter, for example `RUST_LOG=aquila::server=trace`. ### Common (Static + Dynamic) -| Name | Description | Default | -|-----------------------|--------------------------------------------------------------------------------------------------------------|---------------------------------| -| `MODE` | Runtime mode. `static` loads local flows. `dynamic` enables dynamic synchronization with Sagittarius. | `static` | -| `ENVIRONMENT` | Runtime environment (`development`, `staging`, `production`). | `development` | -| `NATS_URL` | URL of the NATS instance Aquila connects to. | `nats://localhost:4222` | -| `NATS_BUCKET` | Name of the NATS KV bucket used to store flows. | `flow_store` | -| `GRPC_HOST` | Hostname for the Aquila gRPC server. | `127.0.0.1` | -| `GRPC_PORT` | Port for the Aquila gRPC server. | `8081` | -| `WITH_HEALTH_SERVICE` | Enables the gRPC health service when set to `true`. | `false` | -| `SERVICE_CONFIG_PATH` | Path to the JSON service-configuration file used for runtime/action authorization and default action configs. | `./service.configuration.json` | -| `RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS` | Seconds before a runtime service is marked as `not_responding` when no heartbeat was received. | `90s` | -| `RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS` | Additional seconds before a non-responding runtime service is marked as `stopped`. | `180s` | -| `RUNTIME_STATUS_MONITOR_INTERVAL_SECS` | Interval in which runtime service heartbeats are checked. | `30s` | -| `SAGITTARIUS_UNARY_RPC_TIMEOUT_SECS` | Timeout in seconds for unary RPC calls from Aquila to Sagittarius. | `5s` | +| YAML key | Description | +|----------|-------------| +| `mode` | `static` loads local flows; `dynamic` synchronizes with Sagittarius. | +| `environment` | `development`, `staging`, or `production`. | +| `log_level` | Default application log filter. | +| `nats.url` | NATS server URL. | +| `nats.bucket` | NATS KV bucket used to store flows. | +| `grpc.host` | Aquila gRPC bind host. | +| `grpc.port` | Aquila gRPC bind port. | +| `grpc.health_service` | Enables the gRPC health service. | +| `runtime_status.not_responding_after_secs` | Heartbeat timeout before `not_responding`. | +| `runtime_status.stopped_after_not_responding_secs` | Additional timeout before `stopped`. | +| `runtime_status.monitor_interval_secs` | Heartbeat monitor interval. | ### Static Mode -Set `MODE=static` to load flows from a local JSON file and insert them into the NATS KV store on startup. +Set `mode: static` to load flows from a local JSON file and insert them into the NATS KV store on startup. -| Name | Description | Default | -|----------------------|-----------------------------------------------|---------------------| -| `FLOW_FALLBACK_PATH` | Path to the flow JSON file loaded at startup. | `./flowExport.json` | +| YAML key | Description | Default | +|----------|-------------|---------| +| `static_config.flow_path` | Path to the flow JSON file loaded at startup. | `./flowExport.json` | ### Dynamic Mode -Set `MODE=dynamic` to keep flows synchronized from Sagittarius. +Set `mode: dynamic` to keep flows synchronized from Sagittarius. -| Name | Description | Default | -|-------------------|-----------------------------------------------------------------------------------|--------------------------| -| `SAGITTARIUS_URL` | URL of the Sagittarius instance Aquila connects to for flow and action updates. | `http://localhost:50051` | -| `RUNTIME_TOKEN` | Token used by Aquila to authenticate with Sagittarius. | `default_session_token` | +| YAML key | Environment override | Description | Default | +|----------|----------------------|-------------|---------| +| `dynamic_config.backend_url` | — | URL of the Sagittarius instance Aquila connects to for flow and action updates. | `http://localhost:50051` | +| `dynamic_config.backend_token` | `AQUILA_BACKEND_TOKEN` | Token used by Aquila to authenticate with Sagittarius. | `default_session_token` | +| `dynamic_config.backend_unary_timeout_secs` | — | Timeout for unary Sagittarius RPCs. | `5` | --- ## Service Configuration File -To authorize services like `Taurus`, `Draco`, or an `Action`, they must be declared in Aquila's service configuration file. +To authorize services like `Taurus`, `Draco`, or an `Action`, they must be declared in Aquila's +separate service configuration file. + +The service configuration is optional. Aquila starts with an empty service configuration when +`AQUILA_SERVICE_CONFIG_PATH` is unset. Load one independently with: + +```bash +AQUILA_SERVICE_CONFIG_PATH=/path/to/service.configuration.json cargo run +``` -`SERVICE_CONFIG_PATH` points to a JSON file that defines: +The file defines: - Allowed runtime tokens - Allowed action tokens - Optional default action configurations -This file is loaded on startup. If the file is missing or invalid, Aquila starts with an empty service configuration. +If the selected file is missing or invalid, Aquila starts with an empty service configuration. Default: diff --git a/src/configuration/config.rs b/src/configuration/config.rs index 2039fa0..ccfa4fe 100644 --- a/src/configuration/config.rs +++ b/src/configuration/config.rs @@ -1,113 +1,278 @@ -use code0_flow::flow_config::{env_with_default, environment::Environment, mode::Mode}; +use std::{fmt, path::Path}; -/// Struct for all relevant `Aquila` startup configurations -pub struct Config { - /// Aquila environment - /// - /// Options: - /// `development` (default) - /// `staging` - /// `production` - pub environment: Environment, +use config::{Config as ConfigLoader, ConfigError, File}; +use serde::{Deserialize, Serialize}; - /// Aquila mode - /// - /// Options: - /// `static` (default) - /// `dynamic` - pub mode: Mode, +use super::{env::Environment, mode::Mode}; - /// URL to the NATS Server. - pub nats_url: String, +const CONFIG_FILE: &str = "aquila"; +const BACKEND_TOKEN_ENV: &str = "AQUILA_BACKEND_TOKEN"; - /// Name of the NATS Bucket. - pub nats_bucket: String, +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct Config { + pub environment: Environment, + pub mode: Mode, + pub log_level: String, + pub nats: Nats, + pub static_config: StaticConfig, + pub dynamic_config: DynamicConfig, + pub grpc: Grpc, + pub runtime_status: RuntimeStatus, +} - /// Fallback file to load flows if gRPC & scheduling is disabled. - pub flow_fallback_path: String, +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct Nats { + pub url: String, + pub bucket: String, +} - /// Verification Token required for internal communication - pub runtime_token: String, +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct StaticConfig { + pub flow_path: String, +} - /// URL to the `Sagittarius` Server. +#[derive(Clone, Deserialize, Serialize)] +#[serde(default)] +pub struct DynamicConfig { pub backend_url: String, + pub backend_token: String, + pub backend_unary_timeout_secs: u64, +} - // Port of the `Aquila` Server - pub grpc_port: u16, +impl std::fmt::Debug for DynamicConfig { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("DynamicConfig") + .field("backend_url", &self.backend_url) + .field("backend_token", &"[FILTERED]") + .field( + "backend_unary_timeout_secs", + &self.backend_unary_timeout_secs, + ) + .finish() + } +} - // Host of the `Aquila` Server - pub grpc_host: String, +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct Grpc { + pub host: String, + pub port: u16, + pub health_service: bool, +} - pub with_health_service: bool, +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct RuntimeStatus { + pub not_responding_after_secs: u64, + pub stopped_after_not_responding_secs: u64, + pub monitor_interval_secs: u64, +} - pub service_config_path: String, +impl Default for Config { + fn default() -> Self { + Self { + environment: Environment::Development, + mode: Mode::Static, + log_level: "debug".into(), + nats: Nats::default(), + static_config: StaticConfig::default(), + dynamic_config: DynamicConfig::default(), + grpc: Grpc::default(), + runtime_status: RuntimeStatus::default(), + } + } +} - /// Runtime heartbeat timeout in seconds before a service is marked as NOT_RESPONDING. - pub runtime_status_not_responding_after_secs: u64, +impl Default for Nats { + fn default() -> Self { + Self { + url: "nats://localhost:4222".into(), + bucket: "flow_store".into(), + } + } +} - /// Additional timeout in seconds after NOT_RESPONDING before a service is marked as STOPPED. - pub runtime_status_stopped_after_not_responding_secs: u64, +impl Default for StaticConfig { + fn default() -> Self { + Self { + flow_path: "./flowExport.json".into(), + } + } +} - /// Interval in seconds for the runtime status timeout monitor loop. - pub runtime_status_monitor_interval_secs: u64, +impl Default for DynamicConfig { + fn default() -> Self { + Self { + backend_url: "http://localhost:50051".into(), + backend_token: "default_session_token".into(), + backend_unary_timeout_secs: 5, + } + } +} - /// Timeout in seconds for unary RPC calls from Aquila to Sagittarius. - pub sagittarius_unary_rpc_timeout_secs: u64, +impl Default for Grpc { + fn default() -> Self { + Self { + host: "127.0.0.1".into(), + port: 8081, + health_service: false, + } + } } -/// Implementation for all relevant `Aquila` startup configurations -/// -/// Behavior: -/// Searches for the env. file at root level. Filename: `.env` -impl Default for Config { +impl Default for RuntimeStatus { fn default() -> Self { - Self::new() + Self { + not_responding_after_secs: 90, + stopped_after_not_responding_secs: 180, + monitor_interval_secs: 30, + } } } impl Config { pub fn new() -> Self { - Config { - environment: env_with_default("ENVIRONMENT", Environment::Development), - mode: env_with_default("MODE", Mode::STATIC), - nats_url: env_with_default("NATS_URL", String::from("nats://localhost:4222")), - nats_bucket: env_with_default("NATS_BUCKET", String::from("flow_store")), - flow_fallback_path: env_with_default( - "FLOW_FALLBACK_PATH", - String::from("./flowExport.json"), - ), - grpc_port: env_with_default("GRPC_PORT", 8081), - grpc_host: env_with_default("GRPC_HOST", String::from("127.0.0.1")), - with_health_service: env_with_default("WITH_HEALTH_SERVICE", false), - runtime_token: env_with_default("RUNTIME_TOKEN", String::from("default_session_token")), - backend_url: env_with_default( - "SAGITTARIUS_URL", - String::from("http://localhost:50051"), - ), - service_config_path: env_with_default( - "SERVICE_CONFIG_PATH", - String::from("./service.configuration.json"), - ), - runtime_status_not_responding_after_secs: env_with_default( - "RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS", - 90_u64, - ), - runtime_status_stopped_after_not_responding_secs: env_with_default( - "RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS", - 180_u64, - ), - runtime_status_monitor_interval_secs: env_with_default( - "RUNTIME_STATUS_MONITOR_INTERVAL_SECS", - 30_u64, - ), - sagittarius_unary_rpc_timeout_secs: env_with_default( - "SAGITTARIUS_UNARY_RPC_TIMEOUT_SECS", - 5_u64, - ), + Self::try_new() + .unwrap_or_else(|error| panic!("failed to load Aquila configuration: {error}")) + } + + pub fn try_new() -> Result { + Self::try_from_optional_path(None) + } + + pub fn try_from_path(path: impl AsRef) -> Result { + Self::try_from_optional_path(Some(path.as_ref())) + } + + fn try_from_optional_path(path: Option<&Path>) -> Result { + let mut builder = + ConfigLoader::builder().add_source(ConfigLoader::try_from(&Self::default())?); + + builder = match path { + Some(path) => builder.add_source(File::from(path).required(true)), + None => builder.add_source(File::with_name(CONFIG_FILE).required(false)), + }; + + if let Ok(token) = std::env::var(BACKEND_TOKEN_ENV) { + builder = builder.set_override("dynamic_config.backend_token", token)?; } + + builder.build()?.try_deserialize() } pub fn is_static(&self) -> bool { - self.mode == Mode::STATIC + self.mode == Mode::Static + } +} + +impl fmt::Display for Config { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + writeln!(formatter, "Aquila configuration")?; + writeln!(formatter, " Environment: {}", self.environment)?; + writeln!(formatter, " Mode: {}", self.mode)?; + writeln!(formatter, " Log level: {}", self.log_level)?; + writeln!(formatter, " NATS")?; + writeln!(formatter, " URL: {}", self.nats.url)?; + writeln!(formatter, " Bucket: {}", self.nats.bucket)?; + writeln!(formatter, " gRPC")?; + writeln!( + formatter, + " Address: {}:{}", + self.grpc.host, self.grpc.port + )?; + writeln!( + formatter, + " Health service: {}", + self.grpc.health_service + )?; + writeln!(formatter, " Static mode")?; + writeln!(formatter, " Flow path: {}", self.static_config.flow_path)?; + writeln!(formatter, " Dynamic mode")?; + writeln!( + formatter, + " Backend URL: {}", + self.dynamic_config.backend_url + )?; + writeln!(formatter, " Backend token: [FILTERED]")?; + writeln!( + formatter, + " Request timeout: {}s", + self.dynamic_config.backend_unary_timeout_secs + )?; + writeln!(formatter, " Runtime status")?; + writeln!( + formatter, + " Not responding after: {}s", + self.runtime_status.not_responding_after_secs + )?; + writeln!( + formatter, + " Stopped after: {}s", + self.runtime_status.stopped_after_not_responding_secs + )?; + write!( + formatter, + " Monitor interval: {}s", + self.runtime_status.monitor_interval_secs + ) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::Config; + + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn environment_overrides_backend_token() { + let _guard = ENV_LOCK.lock().expect("environment test lock poisoned"); + + // SAFETY: access to these process-wide variables is serialized for this test. + unsafe { + std::env::set_var("AQUILA_BACKEND_TOKEN", "environment-token"); + } + + let config = Config::try_new().expect("configuration should load"); + + // SAFETY: access to these process-wide variables is serialized for this test. + unsafe { + std::env::remove_var("AQUILA_BACKEND_TOKEN"); + } + + assert_eq!(config.dynamic_config.backend_token, "environment-token"); + } + + #[test] + fn debug_output_filters_backend_token() { + let mut config = Config::default(); + config.dynamic_config.backend_token = "super-secret".into(); + + let output = format!("{config:#?}"); + + assert!(output.contains("[FILTERED]")); + assert!(!output.contains("super-secret")); + } + + #[test] + fn display_output_is_readable_and_filters_backend_token() { + let mut config = Config::default(); + config.dynamic_config.backend_token = "super-secret".into(); + + let output = config.to_string(); + + assert!(output.starts_with("Aquila configuration\n")); + assert!(output.contains(" Environment: development")); + assert!(output.contains(" Address: 127.0.0.1:8081")); + assert!(output.contains(" Request timeout: 5s")); + assert!(output.contains(" Backend token: [FILTERED]")); + assert!(!output.contains("super-secret")); + assert!(!output.contains("Config {")); } } diff --git a/src/configuration/env.rs b/src/configuration/env.rs new file mode 100644 index 0000000..a3f6aa9 --- /dev/null +++ b/src/configuration/env.rs @@ -0,0 +1,23 @@ +use std::fmt; + +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum Environment { + #[default] + Development, + Staging, + Production, +} + +impl fmt::Display for Environment { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let value = match self { + Self::Development => "development", + Self::Staging => "staging", + Self::Production => "production", + }; + formatter.write_str(value) + } +} diff --git a/src/configuration/mod.rs b/src/configuration/mod.rs index d05ec52..ffed714 100644 --- a/src/configuration/mod.rs +++ b/src/configuration/mod.rs @@ -1,3 +1,5 @@ pub mod config; +pub mod env; +pub mod mode; pub mod service; pub mod state; diff --git a/src/configuration/mode.rs b/src/configuration/mode.rs new file mode 100644 index 0000000..bab984e --- /dev/null +++ b/src/configuration/mode.rs @@ -0,0 +1,22 @@ +use std::fmt; + +use serde::{Deserialize, Serialize}; + +/// Controls whether flows are loaded locally or synchronized with Sagittarius. +#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum Mode { + #[default] + Static, + Dynamic, +} + +impl fmt::Display for Mode { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let value = match self { + Self::Static => "static", + Self::Dynamic => "dynamic", + }; + formatter.write_str(value) + } +} diff --git a/src/configuration/service.rs b/src/configuration/service.rs index 68a68f2..4e31ab3 100644 --- a/src/configuration/service.rs +++ b/src/configuration/service.rs @@ -1,6 +1,6 @@ use serde::{Deserialize, Serialize}; use serde_json::from_str; -use std::{fs::File, io::Read}; +use std::{fs::File, io::Read, path::Path}; use tucana::shared::{ModuleConfigurations, helper::value::from_json_value}; #[derive(Serialize, Deserialize, Clone)] @@ -151,7 +151,7 @@ impl ServiceConfiguration { } } - pub fn from_path(path: &String) -> Self { + pub fn from_path(path: impl AsRef) -> Self { let mut data = String::new(); let mut file = match File::open(path) { @@ -231,10 +231,7 @@ mod tests { &String::from("taurus-token"), &String::from("taurus-runtime-01") )); - assert!(config.has_runtime( - &String::from("taurus-token"), - &String::from("taurus") - )); + assert!(config.has_runtime(&String::from("taurus-token"), &String::from("taurus"))); assert!(config.has_runtime( &String::from("draco-rest-token"), &String::from("draco-rest") @@ -243,14 +240,8 @@ mod tests { &String::from("draco-cron-token"), &String::from("draco-cron") )); - assert!(!config.has_runtime( - &String::from("taurus-token"), - &String::from("draco-rest") - )); - assert!(!config.has_runtime( - &String::from("draco-rest-token"), - &String::from("taurus-x") - )); + assert!(!config.has_runtime(&String::from("taurus-token"), &String::from("draco-rest"))); + assert!(!config.has_runtime(&String::from("draco-rest-token"), &String::from("taurus-x"))); assert!(!config.has_runtime( &String::from("taurus-token"), &String::from("unknown-runtime") @@ -269,20 +260,14 @@ mod tests { &String::from("taurus-token"), &String::from("action-identifier") )); - assert!(!config.has_action( - &String::from("action-token"), - &String::from("action-other") - )); + assert!(!config.has_action(&String::from("action-token"), &String::from("action-other"))); } #[test] fn has_service_returns_true_for_valid_runtime_or_action_pairings() { let config = fixture(); - assert!(config.has_service( - &String::from("taurus-token"), - &String::from("taurus-x") - )); + assert!(config.has_service(&String::from("taurus-token"), &String::from("taurus-x"))); assert!(config.has_service( &String::from("draco-rest-token"), &String::from("draco-rest") @@ -295,9 +280,6 @@ mod tests { &String::from("draco-rest-token"), &String::from("action-identifier") )); - assert!(!config.has_service( - &String::from("action-token"), - &String::from("taurus-x") - )); + assert!(!config.has_service(&String::from("action-token"), &String::from("taurus-x"))); } } diff --git a/src/main.rs b/src/main.rs index 0202a5e..d6356d9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,26 +10,57 @@ pub mod sagittarius; pub mod server; pub mod startup; +const CONFIG_PATH_ENV: &str = "AQUILA_CONFIG_PATH"; +const SERVICE_CONFIG_PATH_ENV: &str = "AQUILA_SERVICE_CONFIG_PATH"; + #[tokio::main] async fn main() { - env_logger::Builder::from_default_env() - .filter_level(log::LevelFilter::Debug) - .init(); + // Load .env before config-rs applies environment overrides. + load_env_file(); + let config_result = match std::env::var(CONFIG_PATH_ENV) { + Ok(path) => AquilaConfig::try_from_path(path), + Err(_) => AquilaConfig::try_new(), + }; + + let log_level = config_result + .as_ref() + .map(|config| config.log_level.as_str()) + .unwrap_or("debug"); + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(log_level)).init(); + install_panic_logging(); + + let config = config_result + .unwrap_or_else(|error| panic!("failed to load Aquila configuration: {error}")); log::info!("Starting Aquila"); - // Load environment variables from .env file - load_env_file(); - let config = AquilaConfig::new(); let app_readiness = AppReadiness::new(); - let service_config = ServiceConfiguration::from_path(&config.service_config_path); - log::debug!( - "Configuration loaded mode={:?} environment={:?} grpc={}:{} health_service={}", - config.mode, - config.environment, - config.grpc_host, - config.grpc_port, - config.with_health_service - ); + let service_config = std::env::var_os(SERVICE_CONFIG_PATH_ENV) + .map(ServiceConfiguration::from_path) + .unwrap_or_default(); + log::debug!("{config}"); startup::run(config, app_readiness, service_config).await; } + +fn install_panic_logging() { + std::panic::set_hook(Box::new(move |panic_info| { + let message = if let Some(message) = panic_info.payload().downcast_ref::<&str>() { + *message + } else if let Some(message) = panic_info.payload().downcast_ref::() { + message.as_str() + } else { + "" + }; + + match panic_info.location() { + Some(location) => log::error!( + "Process panic message={} file={} line={} column={}", + message, + location.file(), + location.line(), + location.column() + ), + None => log::error!("Process panic message={} location=unknown", message), + } + })); +} diff --git a/src/sagittarius/flow_service_client_impl.rs b/src/sagittarius/flow_service_client_impl.rs index 31b9ced..2ad64f7 100644 --- a/src/sagittarius/flow_service_client_impl.rs +++ b/src/sagittarius/flow_service_client_impl.rs @@ -71,12 +71,14 @@ impl SagittariusFlowClient { return; } - log::info!("Will export flows into file because env is set to `DEVELOPMENT`"); - let json = match serde_json::to_vec_pretty(&flows) { Ok(b) => b, Err(e) => { - log::error!("Failed to serialize flows to JSON: {:?}", e); + log::error!( + "Failed to serialize development flow export flow_count={} error={:?}", + flows.flows.len(), + e + ); return; } }; @@ -85,16 +87,48 @@ impl SagittariusFlowClient { let tmp_path = Path::new("flowExport.json.tmp"); if let Err(e) = fs::write(tmp_path, &json).await { - log::error!("Failed to write {}: {}", tmp_path.display(), e); + log::error!( + "Failed to write development flow export path={} error={}", + tmp_path.display(), + e + ); return; } if let Err(e) = fs::rename(tmp_path, final_path).await { - log::warn!("rename failed (will try remove+rename): {}", e); - let _ = fs::remove_file(final_path).await; + log::warn!( + "Could not atomically replace development flow export path={} error={}; retrying after removing destination", + final_path.display(), + e + ); + match fs::remove_file(final_path).await { + Ok(()) => log::debug!( + "Removed previous development flow export path={}", + final_path.display() + ), + Err(remove_error) if remove_error.kind() == std::io::ErrorKind::NotFound => {} + Err(remove_error) => log::warn!( + "Failed to remove previous development flow export path={} error={}", + final_path.display(), + remove_error + ), + } if let Err(e2) = fs::rename(tmp_path, final_path).await { - log::error!("Failed to move export into place: {}", e2); - let _ = fs::remove_file(tmp_path).await; + log::error!( + "Failed to replace development flow export source_path={} destination_path={} initial_rename_error={} retry_error={}", + tmp_path.display(), + final_path.display(), + e, + e2 + ); + if let Err(cleanup_error) = fs::remove_file(tmp_path).await { + log::warn!( + "Failed to clean up temporary development flow export path={} error={}", + tmp_path.display(), + cleanup_error + ); + } + return; } } @@ -107,24 +141,21 @@ impl SagittariusFlowClient { async fn handle_response(&mut self, response: FlowResponse) { let data = match response.data { - Some(data) => { - log::info!("Received a FlowResponse"); - data - } + Some(data) => data, None => { - log::error!("Received a empty FlowResponse"); + log::warn!("Received empty Sagittarius flow response"); return; } }; match data { Data::DeletedFlowId(id) => { - log::info!("Deleting the Flow with the id: {}", id); + log::debug!("Applying flow deletion flow_id={}", id); let mut keys = match self.store.keys().await { Ok(keys) => keys.boxed(), Err(err) => { log::error!( - "Failed to list flows while deleting flow {}. Reason: {:?}", + "Failed to list stored flows for deletion flow_id={} error={:?}", id, err ); @@ -141,7 +172,7 @@ impl SagittariusFlowClient { match self.store.delete(&key).await { Ok(_) => deleted_count += 1, Err(err) => log::error!( - "Failed to delete flow {} with key {}. Reason: {:?}", + "Failed to delete stored flow flow_id={} key={} error={:?}", id, key, err @@ -150,7 +181,7 @@ impl SagittariusFlowClient { } if deleted_count == 0 { - log::warn!("No stored flow found with the id: {}", id); + log::warn!("Flow deletion matched no stored keys flow_id={}", id); } else { log::info!( "Flow deleted successfully id={} deleted_keys={}", @@ -160,23 +191,35 @@ impl SagittariusFlowClient { } } Data::UpdatedFlow(flow) => { - log::info!("Updating the Flow with the id: {}", &flow.flow_id); let key = get_flow_identifier(&flow); + let flow_id = flow.flow_id.clone(); let bytes = flow.encode_to_vec(); - match self.store.put(key, bytes.into()).await { - Ok(_) => log::info!("Flow updated successfully"), - Err(err) => log::error!("Failed to update flow. Reason: {:?}", err), + match self.store.put(key.clone(), bytes.into()).await { + Ok(_) => log::info!("Stored flow update flow_id={} key={}", flow_id, key), + Err(err) => log::error!( + "Failed to store flow update flow_id={} key={} error={:?}", + flow_id, + key, + err + ), }; } Data::Flows(flows) => { - log::info!("Dropping all Flows & inserting the new ones!"); + let received_count = flows.flows.len(); + log::info!( + "Replacing stored flows from Sagittarius received_count={}", + received_count + ); self.export_flows_json_overwrite(flows.clone()).await; let mut keys = match self.store.keys().await { Ok(keys) => keys.boxed(), Err(err) => { - log::error!("Service wasn't able to get keys. Reason: {:?}", err); + log::error!( + "Failed to list stored flows before replacement error={:?}", + err + ); return; } }; @@ -185,21 +228,34 @@ impl SagittariusFlowClient { while let Ok(Some(key)) = keys.try_next().await { match self.store.purge(&key).await { Ok(_) => purged_count += 1, - Err(e) => log::error!("Failed to purge key {}: {}", key, e), + Err(e) => { + log::error!("Failed to purge stored flow key={} error={}", key, e) + } } } - log::info!("Purged {} existing keys", purged_count); - + let mut stored_count = 0; for flow in flows.flows { let key = get_flow_identifier(&flow); - log::debug!("trying to insert: {}", key); let bytes = flow.encode_to_vec(); - match self.store.put(key, bytes.into()).await { - Ok(_) => log::info!("Flow updated successfully"), - Err(err) => log::error!("Failed to update flow. Reason: {:?}", err), + match self.store.put(key.clone(), bytes.into()).await { + Ok(_) => { + stored_count += 1; + log::debug!("Stored replacement flow key={}", key); + } + Err(err) => log::error!( + "Failed to store replacement flow key={} error={:?}", + key, + err + ), }; } + log::info!( + "Finished replacing stored flows received_count={} purged_count={} stored_count={}", + received_count, + purged_count, + stored_count + ); } Data::ModuleConfigurations(action_configurations) => { let (project_count, config_count) = module_config_stats(&action_configurations); diff --git a/src/sagittarius/module_service_client_impl.rs b/src/sagittarius/module_service_client_impl.rs index ace4fbd..4419ad4 100644 --- a/src/sagittarius/module_service_client_impl.rs +++ b/src/sagittarius/module_service_client_impl.rs @@ -60,8 +60,11 @@ impl SagittariusModuleServiceClient { } Err(err) => { log::error!( - "Failed to update Modules via Sagittarius RPC transport: {:?}", - err + "Sagittarius module update RPC failed module_count={} code={} message={} timeout_ms={}", + module_count, + err.code(), + err.message(), + self.unary_rpc_timeout.as_millis() ); tucana::aquila::ModuleUpdateResponse { success: false } } diff --git a/src/sagittarius/retry.rs b/src/sagittarius/retry.rs index 87d161f..bc3eb93 100644 --- a/src/sagittarius/retry.rs +++ b/src/sagittarius/retry.rs @@ -30,7 +30,11 @@ pub async fn create_channel_with_retry( let channel = match Endpoint::from_shared(url.clone()) { Ok(c) => { - log::debug!("Creating a new endpoint for the: {} Service", channel_name); + log::debug!( + "Created Sagittarius endpoint channel={} url={}", + channel_name, + url + ); c.connect_timeout(Duration::from_secs(2)) } Err(err) => { @@ -67,13 +71,10 @@ pub async fn create_channel_with_retry( retries += 1; if retries >= MAX_RETRIES { - log::error!( - "Reached max retries channel={} url={} max_retries={}", - channel_name, - url, - MAX_RETRIES + panic!( + "Sagittarius connection retries exhausted channel={} url={} attempts={} last_error={:?}", + channel_name, url, MAX_RETRIES, err ); - panic!("Reached max retries to url {}", url) } } } diff --git a/src/sagittarius/runtime_status_service_client_impl.rs b/src/sagittarius/runtime_status_service_client_impl.rs index 0914fba..829d02a 100644 --- a/src/sagittarius/runtime_status_service_client_impl.rs +++ b/src/sagittarius/runtime_status_service_client_impl.rs @@ -39,7 +39,12 @@ impl SagittariusRuntimeStatusServiceClient { response.into_inner() } Err(err) => { - log::error!("Failed to update RuntimeStatus: {:?}", err); + log::error!( + "Sagittarius runtime status update RPC failed code={} message={} timeout_ms={}", + err.code(), + err.message(), + self.unary_rpc_timeout.as_millis() + ); return tucana::aquila::RuntimeStatusUpdateResponse { success: false }; } }; diff --git a/src/sagittarius/test_execution_client_impl.rs b/src/sagittarius/test_execution_client_impl.rs index 92f343e..ff3af22 100644 --- a/src/sagittarius/test_execution_client_impl.rs +++ b/src/sagittarius/test_execution_client_impl.rs @@ -289,16 +289,24 @@ impl SagittariusTestExecutionServiceClient { Some(flow) } Err(err) => { - log::error!("Cannot decode ValidationFlow for {}: {:?}", flow_id, err); + log::error!( + "Failed to decode validation flow flow_id={} error={:?}", + flow_id, + err + ); None } }, Ok(None) => { - log::error!("No flow found with id: {}", flow_id); + log::error!("Validation flow was not found flow_id={}", flow_id); None } Err(err) => { - log::error!("Error fetching flow {}: {:?}", flow_id, err); + log::error!( + "Failed to fetch validation flow flow_id={} error={:?}", + flow_id, + err + ); None } } @@ -314,7 +322,10 @@ impl SagittariusTestExecutionServiceClient { log::debug!("Queueing Sagittarius execution stream logon before opening stream"); if let Err(err) = tx.send(logon).await { - log::error!("Failed to queue test execution logon: {:?}", err); + log::error!( + "Failed to queue Sagittarius execution stream logon reason=channel_closed error={:?}", + err + ); self.response_sender.clear().await; return; } @@ -335,8 +346,9 @@ impl SagittariusTestExecutionServiceClient { } Err(error) => { log::error!( - "Failed to establish Sagittarius execution stream: {:?}", - error + "Failed to establish Sagittarius execution stream code={} message={}", + error.code(), + error.message() ); self.response_sender.clear().await; return; diff --git a/src/server/action_transfer_service_server_impl.rs b/src/server/action_transfer_service_server_impl.rs index ccb99f7..6bac2fd 100644 --- a/src/server/action_transfer_service_server_impl.rs +++ b/src/server/action_transfer_service_server_impl.rs @@ -56,6 +56,7 @@ impl AquilaActionTransferServiceServer { } } +#[derive(Debug)] enum FlowIdentificationError { KVError, } @@ -258,15 +259,15 @@ async fn handle_logon( pending_replies: PendingReplies, cfg_forwarder_started: &mut bool, ) -> Result { - log::info!("Action logon attempt payload={:?}", action_logon); - let module = match action_logon.module.as_mut() { Some(m) => m, None => { + log::warn!("Rejected action logon reason=missing_module"); return Err(Status::aborted("Please provide a module configuration.")); } }; let identifier = module.identifier.clone(); + log::info!("Action logon attempt identifier={}", identifier); { let lock = actions.lock().await; @@ -308,7 +309,8 @@ async fn handle_logon( Ok(s) => s, Err(err) => { log::error!( - "Could not subscribe to action: {}. Reason: {:?}", + "Failed to subscribe to action execution subject identifier={} subject=action.{}.* error={:?}", + identifier, identifier, err ); @@ -320,7 +322,7 @@ async fn handle_logon( if let Err(err) = client.flush().await { log::error!( - "Could not flush action subscription: {}. Reason: {:?}", + "Failed to flush action execution subscription identifier={} error={:?}", identifier, err ); @@ -388,19 +390,32 @@ async fn handle_event( ) { let pattern = format!("{}.*.{}.*", event.event_type, event.project_id); log::debug!( - "Handling event type {} for project {}", + "Handling action event event_type={} project_id={}", event.event_type, event.project_id ); - let flows = match get_flows(pattern, kv).await { + let flows = match get_flows(pattern.clone(), kv).await { Ok(f) => f, - Err(_) => { - log::error!("Cound not find any flows"); + Err(err) => { + log::error!( + "Failed to find flows for action event event_type={} project_id={} pattern={} error={:?}", + event.event_type, + event.project_id, + pattern, + err + ); return; } }; + let matched_count = flows.flows.len(); + log::info!( + "Matched flows for action event event_type={} project_id={} flow_count={}", + event.event_type, + event.project_id, + matched_count + ); for flow in flows.flows { let uuid = uuid::Uuid::new_v4().to_string(); let flow_id = flow.flow_id; @@ -408,18 +423,20 @@ async fn handle_event( let bytes = execution_flow.encode_to_vec(); let topic = format!("execution.{}", uuid); - log::info!("{:#?}", execution_flow); - log::info!( - "Requesting execution flow_id={} execution_id={}", + "Requesting execution flow_id={} execution_id={} event_type={} project_id={}", flow_id, - uuid + uuid, + event.event_type, + event.project_id ); - if let Err(err) = client.request(topic, bytes.into()).await { + if let Err(err) = client.request(topic.clone(), bytes.into()).await { log::error!( - "Failed to request execution for flow {}: {:?}", + "Failed to request execution flow_id={} execution_id={} topic={} error={:?}", flow_id, + uuid, + topic, err ); } @@ -448,19 +465,20 @@ async fn handle_result( }; log::debug!( - "Publishing execution result for {} to reply subject {}", + "Publishing execution result execution_id={} reply_subject={}", execution_id, pending_reply.reply_subject ); let payload = execution_result.encode_to_vec(); if let Err(err) = client - .publish(pending_reply.reply_subject, payload.into()) + .publish(pending_reply.reply_subject.clone(), payload.into()) .await { log::error!( - "Failed to publish action result for execution {}: {:?}", + "Failed to publish action result execution_id={} reply_subject={} error={:?}", execution_id, + pending_reply.reply_subject, err ); return; @@ -468,7 +486,7 @@ async fn handle_result( if let Err(err) = client.flush().await { log::error!( - "Failed to flush action result for execution {}: {:?}", + "Failed to flush action result execution_id={} error={:?}", execution_id, err ); @@ -510,7 +528,7 @@ impl ActionTransferService for AquilaActionTransferServiceServer { let transfer_request = match next { Ok(tr) => tr, Err(status) => { - log::error!("Action transfer stream closed status={:?}", status); + log::warn!("Action transfer input stream failed status={:?}", status); break; } }; @@ -531,7 +549,7 @@ impl ActionTransferService for AquilaActionTransferServiceServer { let identifier = match action_logon.module { Some(ref m) => m.identifier.clone(), None => { - log::error!("Logon failed (no module present)"); + log::warn!("Rejected action logon reason=missing_module"); break; } }; @@ -553,7 +571,12 @@ impl ActionTransferService for AquilaActionTransferServiceServer { { Ok(v) => v, Err(status) => { - log::error!("Action logon failed status={:?}", status); + log::warn!( + "Action logon failed identifier={} code={:?} message={}", + identifier, + status.code(), + status.message() + ); break; } }; @@ -597,7 +620,10 @@ impl ActionTransferService for AquilaActionTransferServiceServer { match data { tucana::aquila::action_transfer_request::Data::Logon(_) => { - log::error!("Received duplicate logon"); + log::warn!( + "Action stream protocol violation identifier={} reason=duplicate_logon", + identifier + ); break; } tucana::aquila::action_transfer_request::Data::Event(event) => { @@ -633,12 +659,15 @@ async fn forward_nats_to_action( log::debug!("Waiting for incoming action execution request"); while let Some(msg) = sub.next().await { - log::debug!("Received RemoteRuntime execution request"); - let mut execution = match ActionExecutionRequest::decode(msg.payload.as_ref()) { Ok(req) => req, Err(err) => { - log::error!("Invalid execution request payload: {:?}", err); + log::error!( + "Invalid action execution request payload subject={} payload_bytes={} error={:?}", + msg.subject, + msg.payload.len(), + err + ); continue; } }; @@ -688,9 +717,9 @@ async fn forward_nats_to_action( ); log::debug!( - "Forwarding execution request to action execution_id={} request={:#?}", + "Forwarding execution request to action execution_id={} subject={}", execution_id, - execution + msg.subject ); let resp = ActionTransferResponse { diff --git a/src/server/dynamic_server.rs b/src/server/dynamic_server.rs index 8eeaf55..f424ad4 100644 --- a/src/server/dynamic_server.rs +++ b/src/server/dynamic_server.rs @@ -54,7 +54,7 @@ impl AquilaDynamicServer { action_config_tx: tokio::sync::broadcast::Sender, execution_response_sender: SagittariusExecutionResponseSender, ) -> Self { - let address = match format!("{}:{}", config.grpc_host, config.grpc_port).parse() { + let address = match format!("{}:{}", config.grpc.host, config.grpc.port).parse() { Ok(addr) => { info!("Listening on {:?}", &addr); addr @@ -63,9 +63,9 @@ impl AquilaDynamicServer { }; AquilaDynamicServer { - token: config.runtime_token.clone(), - nats_url: config.nats_url.clone(), - with_health_service: config.with_health_service, + token: config.dynamic_config.backend_token.clone(), + nats_url: config.nats.url.clone(), + with_health_service: config.grpc.health_service, address, app_readiness, channel, @@ -75,17 +75,15 @@ impl AquilaDynamicServer { action_config_tx, execution_response_sender, runtime_status_not_responding_after_secs: config - .runtime_status_not_responding_after_secs - .clone(), + .runtime_status + .not_responding_after_secs, runtime_status_stopped_after_not_responding_secs: config - .runtime_status_stopped_after_not_responding_secs - .clone(), - runtime_status_monitor_interval_secs: config - .runtime_status_monitor_interval_secs - .clone(), + .runtime_status + .stopped_after_not_responding_secs, + runtime_status_monitor_interval_secs: config.runtime_status.monitor_interval_secs, sagittarius_unary_rpc_timeout: Duration::from_secs( - config.sagittarius_unary_rpc_timeout_secs, + config.dynamic_config.backend_unary_timeout_secs, ), } } diff --git a/src/server/static_server.rs b/src/server/static_server.rs index 55dd6d4..7b12ad5 100644 --- a/src/server/static_server.rs +++ b/src/server/static_server.rs @@ -31,7 +31,7 @@ impl AquilaStaticServer { kv_store: Arc, action_config_tx: tokio::sync::broadcast::Sender, ) -> Self { - let address = match format!("{}:{}", config.grpc_host, config.grpc_port).parse() { + let address = match format!("{}:{}", config.grpc.host, config.grpc.port).parse() { Ok(addr) => { info!("Listening on {:?}", &addr); addr @@ -40,8 +40,8 @@ impl AquilaStaticServer { }; AquilaStaticServer { - nats_url: config.nats_url.clone(), - with_health_service: config.with_health_service, + nats_url: config.nats.url.clone(), + with_health_service: config.grpc.health_service, address, app_readiness, service_configuration, diff --git a/src/startup/dynamic_mode.rs b/src/startup/dynamic_mode.rs index aafe881..2ce1601 100644 --- a/src/startup/dynamic_mode.rs +++ b/src/startup/dynamic_mode.rs @@ -24,12 +24,12 @@ pub async fn run( ) { log::info!( "Dynamic mode starting grpc={}:{} backend_url={}", - config.grpc_host, - config.grpc_port, - config.backend_url + config.grpc.host, + config.grpc.port, + config.dynamic_config.backend_url ); - let backend_url_flow = config.backend_url.clone(); + let backend_url_flow = config.dynamic_config.backend_url.clone(); let sagittarius_channel = create_channel_with_retry( "Sagittarius Endpoint", backend_url_flow, @@ -62,22 +62,20 @@ pub async fn run( let kv_for_test_execution = kv_store.clone(); let kv_for_flow = kv_store.clone(); - let backend_url_for_test_execution = config.backend_url.clone(); - let runtime_token_for_test_execution = config.runtime_token.clone(); + let backend_url_for_test_execution = config.dynamic_config.backend_url.clone(); + let runtime_token_for_test_execution = config.dynamic_config.backend_token.clone(); let sagittarius_ready_for_test_execution = app_readiness.sagittarius_ready.clone(); let nats_client_for_test_execution = client.clone(); let execution_response_sender_for_test_execution = execution_response_sender.clone(); - let backend_url_for_flow = config.backend_url.clone(); - let runtime_token_for_flow = config.runtime_token.clone(); + let backend_url_for_flow = config.dynamic_config.backend_url.clone(); + let runtime_token_for_flow = config.dynamic_config.backend_token.clone(); let sagittarius_ready_for_flow = app_readiness.sagittarius_ready.clone(); let env = match config.environment { - code0_flow::flow_config::environment::Environment::Development => { - String::from("DEVELOPMENT") - } - code0_flow::flow_config::environment::Environment::Staging => String::from("STAGING"), - code0_flow::flow_config::environment::Environment::Production => String::from("PRODUCTION"), + crate::configuration::env::Environment::Development => String::from("DEVELOPMENT"), + crate::configuration::env::Environment::Staging => String::from("STAGING"), + crate::configuration::env::Environment::Production => String::from("PRODUCTION"), }; let mut test_execution_task = tokio::spawn(async move { @@ -169,18 +167,30 @@ pub async fn run( let sigterm = std::future::pending::<()>(); tokio::select! { - _ = &mut server_task => { - log::warn!("gRPC server task finished, shutting down"); + result = &mut server_task => { + match result { + Ok(()) => log::warn!("gRPC server task exited unexpectedly; shutting down"), + Err(err) if err.is_panic() => {} + Err(err) => log::error!("gRPC server task failed; shutting down error={:?}", err), + } flow_task.abort(); test_execution_task.abort(); } - _ = &mut test_execution_task => { - log::warn!("Test execution stream task finished, shutting down"); + result = &mut test_execution_task => { + match result { + Ok(()) => log::warn!("Test execution stream task exited unexpectedly; shutting down"), + Err(err) if err.is_panic() => {} + Err(err) => log::error!("Test execution stream task failed; shutting down error={:?}", err), + } server_task.abort(); flow_task.abort(); } - _ = &mut flow_task => { - log::warn!("Flow stream task finished, shutting down"); + result = &mut flow_task => { + match result { + Ok(()) => log::warn!("Flow stream task exited unexpectedly; shutting down"), + Err(err) if err.is_panic() => {} + Err(err) => log::error!("Flow stream task failed; shutting down error={:?}", err), + } server_task.abort(); test_execution_task.abort(); } diff --git a/src/startup/mod.rs b/src/startup/mod.rs index f9fb682..c07f7b9 100644 --- a/src/startup/mod.rs +++ b/src/startup/mod.rs @@ -19,40 +19,51 @@ pub async fn run( } else { "dynamic" }, - config.nats_url, - config.nats_bucket + config.nats.url, + config.nats.bucket ); // Create connection to JetStream - let client = match async_nats::connect(config.nats_url.clone()).await { + let client = match async_nats::connect(config.nats.url.clone()).await { Ok(client) => { log::info!("Connected to NATS"); client } Err(err) => { - log::error!("Failed to connect to NATS: {:?}", err); panic!("Failed to connect to NATS server: {:?}", err) } }; let jet_stream = async_nats::jetstream::new(client.clone()); - let _ = jet_stream + match jet_stream .create_key_value(Config { - bucket: config.nats_bucket.clone(), + bucket: config.nats.bucket.clone(), ..Default::default() }) - .await; - log::debug!("Ensured NATS key-value bucket exists"); + .await + { + Ok(_) => log::debug!( + "NATS key-value bucket is available bucket={}", + config.nats.bucket + ), + Err(err) => log::debug!( + "NATS key-value bucket creation skipped or failed; attempting to open existing bucket bucket={} error={:?}", + config.nats.bucket, + err + ), + } - let kv_store = match jet_stream.get_key_value(config.nats_bucket.clone()).await { + let kv_store = match jet_stream.get_key_value(config.nats.bucket.clone()).await { Ok(kv) => { - log::info!("Connected to JetStream"); + log::info!("Opened NATS key-value store bucket={}", config.nats.bucket); Arc::new(kv) } Err(err) => { - log::error!("Failed to get key-value store: {:?}", err); - panic!("Failed to get key-value store: {:?}", err) + panic!( + "Failed to open NATS key-value store bucket={} error={:?}", + config.nats.bucket, err + ) } }; diff --git a/src/startup/static_mode.rs b/src/startup/static_mode.rs index be1a7b6..a40aae0 100644 --- a/src/startup/static_mode.rs +++ b/src/startup/static_mode.rs @@ -19,15 +19,19 @@ pub async fn run( ) { log::info!( "Static mode starting grpc={}:{} fallback_path={}", - config.grpc_host, - config.grpc_port, - config.flow_fallback_path + config.grpc.host, + config.grpc.port, + config.static_config.flow_path ); app_readiness .sagittarius_ready .store(true, Ordering::SeqCst); - init_flows_from_json(config.flow_fallback_path.clone(), flow_store_client.clone()).await; + init_flows_from_json( + config.static_config.flow_path.clone(), + flow_store_client.clone(), + ) + .await; let (action_config_tx, _) = tokio::sync::broadcast::channel::(64); @@ -61,8 +65,12 @@ pub async fn run( let sigterm = std::future::pending::<()>(); tokio::select! { - _ = &mut server_task => { - log::warn!("gRPC server task finished, shutting down"); + result = &mut server_task => { + match result { + Ok(()) => log::warn!("gRPC server task exited unexpectedly; shutting down"), + Err(err) if err.is_panic() => {} + Err(err) => log::error!("gRPC server task failed; shutting down error={:?}", err), + } } _ = tokio::signal::ctrl_c() => { log::info!("Ctrl+C/Exit signal received, shutting down"); @@ -84,48 +92,57 @@ async fn init_flows_from_json( let mut data = String::new(); log::info!("Loading fallback flows from {}", path); - let mut file = match File::open(path) { + let mut file = match File::open(&path) { Ok(file) => file, Err(error) => { - panic!("There was a problem opening the file: {:?}", error); + panic!("Failed to open fallback flow file path={path}: {error:?}"); } }; match file.read_to_string(&mut data) { - Ok(_) => { - log::info!("Successfully read data from file"); + Ok(byte_count) => { + log::debug!("Read fallback flow file path={} bytes={}", path, byte_count); } Err(error) => { - panic!("There was a problem reading the file: {:?}", error); + panic!("Failed to read fallback flow file path={path}: {error:?}"); } } let flows: Flows = match from_str(&data) { Ok(flows) => flows, Err(error) => { - panic!( - "There was a problem deserializing the json file: {:?}", - error - ); + panic!("Failed to deserialize fallback flow file path={path}: {error:?}"); } }; let flow_count = flows.flows.len(); if flow_count == 0 { - log::warn!("Fallback flow file contains zero flows"); + log::warn!("Fallback flow file contains zero flows path={}", path); } else { - log::info!("Loaded {} fallback flows", flow_count); + log::info!( + "Parsed fallback flow file path={} flow_count={}", + path, + flow_count + ); } + let mut stored_count = 0; for flow in flows.flows { let key = get_flow_identifier(&flow); let bytes = flow.encode_to_vec(); - log::info!("Inserting flow with key {}", &key); - match flow_store_client.put(key, bytes.into()).await { - Ok(_) => log::info!("Flow updated successfully"), - Err(err) => log::error!("Failed to update flow. Reason: {:?}", err), + match flow_store_client.put(key.clone(), bytes.into()).await { + Ok(_) => { + stored_count += 1; + log::debug!("Stored fallback flow key={}", key); + } + Err(err) => log::error!("Failed to store fallback flow key={} error={:?}", key, err), }; } - log::info!("Successfully inserted all flows from the JSON file"); + log::info!( + "Finished loading fallback flows path={} parsed_count={} stored_count={}", + path, + flow_count, + stored_count + ); }