From 94d6f962500059fa0bb461f32e57733d7d4defbc Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 20:09:50 +0800 Subject: [PATCH 1/8] feat: add remote-manage --- Cargo.lock | 171 +++++++++---- crates/arkflow-core/Cargo.toml | 1 + crates/arkflow-core/src/cli/mod.rs | 90 +++++-- crates/arkflow-core/src/lib.rs | 1 + crates/arkflow-core/src/remote_config.rs | 307 +++++++++++++++++++++++ docs/remote-config.md | 239 ++++++++++++++++++ 6 files changed, 745 insertions(+), 64 deletions(-) create mode 100644 crates/arkflow-core/src/remote_config.rs create mode 100644 docs/remote-config.md diff --git a/Cargo.lock b/Cargo.lock index 0d069d96..9f6a649d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -253,6 +253,7 @@ dependencies = [ "futures", "lazy_static", "num_cpus", + "reqwest 0.11.27", "serde", "serde_json", "serde_yaml", @@ -298,7 +299,7 @@ dependencies = [ "rdkafka", "rdkafka-sys", "redis", - "reqwest", + "reqwest 0.12.22", "rumqttc", "sasl2-sys", "serde", @@ -552,7 +553,7 @@ version = "55.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af7686986a3bf2254c9fb130c623cdcb2f8e1f15763e7c71c310f0834da3d292" dependencies = [ - "bitflags", + "bitflags 2.9.1", "serde", ] @@ -1114,7 +1115,7 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sync_wrapper", + "sync_wrapper 1.0.2", "tokio", "tower 0.5.2", "tower-layer", @@ -1137,7 +1138,7 @@ dependencies = [ "mime", "pin-project-lite", "rustversion", - "sync_wrapper", + "sync_wrapper 1.0.2", "tower-layer", "tower-service", "tracing", @@ -1368,7 +1369,7 @@ version = "0.69.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271383c67ccabffb7381723dea0672a673f292304fcb45c01cc648c7a8d58088" dependencies = [ - "bitflags", + "bitflags 2.9.1", "cexpr", "clang-sys", "itertools 0.12.1", @@ -1391,10 +1392,10 @@ version = "0.71.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f58bf3d7db68cfbac37cfc485a8d711e87e064c3d0fe0435b92f7a407f9d6b3" dependencies = [ - "bitflags", + "bitflags 2.9.1", "cexpr", "clang-sys", - "itertools 0.12.1", + "itertools 0.13.0", "proc-macro2", "quote", "regex", @@ -1418,6 +1419,12 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.9.1" @@ -3212,7 +3219,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "778e2ac28f6c47af28e4907f13ffd1e1ddbd400980a9abd7c8df189bf578a5ad" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.60.2", ] [[package]] @@ -3312,7 +3319,7 @@ version = "25.2.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1045398c1bfd89168b5fd3f1fc11f6e70b34f6f66300c87d44d3de849463abf1" dependencies = [ - "bitflags", + "bitflags 2.9.1", "rustc_version", ] @@ -3733,7 +3740,7 @@ checksum = "2b4cabd42d46aac490d01b80596dfd930a8316b4ee3e8b0c94a7cba888791c0f" dependencies = [ "aes", "base64 0.22.1", - "bitflags", + "bitflags 2.9.1", "bytes", "cbc", "chrono", @@ -4009,6 +4016,19 @@ dependencies = [ "tower-service", ] +[[package]] +name = "hyper-tls" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" +dependencies = [ + "bytes", + "hyper 0.14.32", + "native-tls", + "tokio", + "tokio-native-tls", +] + [[package]] name = "hyper-tls" version = "0.6.0" @@ -4044,7 +4064,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2", - "system-configuration", + "system-configuration 0.6.1", "tokio", "tower-service", "tracing", @@ -4270,7 +4290,7 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b86e202f00093dcba4275d4636b93ef9dd75d025ae560d2521b45ea28ab49013" dependencies = [ - "bitflags", + "bitflags 2.9.1", "cfg-if", "libc", ] @@ -4558,7 +4578,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07033963ba89ebaf1584d767badaa2e8fcec21aedea6b8c0346d487d49c28667" dependencies = [ "cfg-if", - "windows-targets 0.48.5", + "windows-targets 0.53.2", ] [[package]] @@ -4583,7 +4603,7 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1580801010e535496706ba011c15f8532df6b42297d2e471fec38ceadd8c0638" dependencies = [ - "bitflags", + "bitflags 2.9.1", "libc", "redox_syscall", ] @@ -4917,7 +4937,7 @@ checksum = "6e0ec195e788c95f36b7cf88127d538465fc2f7773e6e47af01834738eab0aee" dependencies = [ "base64 0.22.1", "bigdecimal", - "bitflags", + "bitflags 2.9.1", "btoi", "byteorder", "bytes", @@ -5203,7 +5223,7 @@ dependencies = [ "percent-encoding", "quick-xml", "rand 0.9.1", - "reqwest", + "reqwest 0.12.22", "ring", "rustls-pemfile 2.2.0", "serde", @@ -5256,7 +5276,7 @@ version = "6.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "336b9c63443aceef14bea841b899035ae3abe89b7c486aaf4c5bd8aafedac3f0" dependencies = [ - "bitflags", + "bitflags 2.9.1", "libc", "once_cell", "onig_sys", @@ -5284,7 +5304,7 @@ version = "0.10.73" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8505734d46c8ab1e19a1dce3aef597ad87dcb4c37e7188231769bd6bd51cebf8" dependencies = [ - "bitflags", + "bitflags 2.9.1", "cfg-if", "foreign-types", "libc", @@ -5850,8 +5870,8 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ - "heck 0.4.1", - "itertools 0.12.1", + "heck 0.5.0", + "itertools 0.14.0", "log", "multimap", "once_cell", @@ -5871,7 +5891,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.12.1", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.101", @@ -6120,7 +6140,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6312,7 +6332,7 @@ version = "0.5.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d04b7d0ee6b4a0207a0a7adb104d23ecb0b47d6beae7152d0fa34b692b29fd6" dependencies = [ - "bitflags", + "bitflags 2.9.1", ] [[package]] @@ -6388,6 +6408,46 @@ dependencies = [ "bytecheck", ] +[[package]] +name = "reqwest" +version = "0.11.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd67538700a17451e7cba03ac727fb961abb7607553461627b97de0b89cf4a62" +dependencies = [ + "base64 0.21.7", + "bytes", + "encoding_rs", + "futures-core", + "futures-util", + "h2 0.3.26", + "http 0.2.12", + "http-body 0.4.6", + "hyper 0.14.32", + "hyper-tls 0.5.0", + "ipnet", + "js-sys", + "log", + "mime", + "native-tls", + "once_cell", + "percent-encoding", + "pin-project-lite", + "rustls-pemfile 1.0.4", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper 0.1.2", + "system-configuration 0.5.1", + "tokio", + "tokio-native-tls", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "winreg", +] + [[package]] name = "reqwest" version = "0.12.22" @@ -6405,7 +6465,7 @@ dependencies = [ "http-body-util", "hyper 1.6.0", "hyper-rustls 0.27.5", - "hyper-tls", + "hyper-tls 0.6.0", "hyper-util", "js-sys", "log", @@ -6420,7 +6480,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", - "sync_wrapper", + "sync_wrapper 1.0.2", "tokio", "tokio-native-tls", "tokio-rustls 0.26.2", @@ -6546,7 +6606,7 @@ version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" dependencies = [ - "bitflags", + "bitflags 2.9.1", "fallible-iterator 0.3.0", "fallible-streaming-iterator", "hashlink 0.9.1", @@ -6603,11 +6663,11 @@ version = "0.38.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" dependencies = [ - "bitflags", + "bitflags 2.9.1", "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6616,11 +6676,11 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c71e83d6afe7ff64890ec6b71d6a69bb8a610ab78ce364b3352876bb4c801266" dependencies = [ - "bitflags", + "bitflags 2.9.1", "errno", "libc", "linux-raw-sys 0.9.4", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6899,7 +6959,7 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" dependencies = [ - "bitflags", + "bitflags 2.9.1", "core-foundation 0.9.4", "core-foundation-sys", "libc", @@ -6912,7 +6972,7 @@ version = "3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271720403f46ca04f7ba6f55d438f8bd878d6b8ca0a1046e8228c4145bcbb316" dependencies = [ - "bitflags", + "bitflags 2.9.1", "core-foundation 0.10.0", "core-foundation-sys", "libc", @@ -7237,7 +7297,7 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03c3c6b7927ffe7ecaa769ee0e3994da3b8cafc8f444578982c83ecb161af917" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.101", @@ -7415,7 +7475,7 @@ checksum = "0afdd3aa7a629683c2d750c2df343025545087081ab5942593a5288855b1b7a7" dependencies = [ "atoi", "base64 0.22.1", - "bitflags", + "bitflags 2.9.1", "byteorder", "bytes", "crc", @@ -7457,7 +7517,7 @@ checksum = "a0bedbe1bbb5e2615ef347a5e9d8cd7680fb63e77d9dafc0f29be15e53f1ebe6" dependencies = [ "atoi", "base64 0.22.1", - "bitflags", + "bitflags 2.9.1", "byteorder", "crc", "dotenvy", @@ -7526,7 +7586,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -7642,6 +7702,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160" + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -7672,15 +7738,36 @@ dependencies = [ "nom", ] +[[package]] +name = "system-configuration" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7" +dependencies = [ + "bitflags 1.3.2", + "core-foundation 0.9.4", + "system-configuration-sys 0.5.0", +] + [[package]] name = "system-configuration" version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" dependencies = [ - "bitflags", + "bitflags 2.9.1", "core-foundation 0.9.4", - "system-configuration-sys", + "system-configuration-sys 0.6.0", +] + +[[package]] +name = "system-configuration-sys" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9" +dependencies = [ + "core-foundation-sys", + "libc", ] [[package]] @@ -7732,7 +7819,7 @@ dependencies = [ "getrandom 0.3.3", "once_cell", "rustix 1.0.7", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -8206,7 +8293,7 @@ dependencies = [ "futures-core", "futures-util", "pin-project-lite", - "sync_wrapper", + "sync_wrapper 1.0.2", "tokio", "tower-layer", "tower-service", @@ -8219,7 +8306,7 @@ version = "0.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2" dependencies = [ - "bitflags", + "bitflags 2.9.1", "bytes", "futures-util", "http 1.3.1", @@ -9318,7 +9405,7 @@ version = "0.39.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6f42320e61fe2cfd34354ecb597f86f413484a798ba44a8ca1165c58d42da6c1" dependencies = [ - "bitflags", + "bitflags 2.9.1", ] [[package]] diff --git a/crates/arkflow-core/Cargo.toml b/crates/arkflow-core/Cargo.toml index d1986276..58395534 100644 --- a/crates/arkflow-core/Cargo.toml +++ b/crates/arkflow-core/Cargo.toml @@ -27,4 +27,5 @@ clap = { workspace = true } colored = { workspace = true } flume = { workspace = true } axum = { workspace = true } +reqwest = { version = "0.11", features = ["json"] } num_cpus = "1.17.0" \ No newline at end of file diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index f2d1686c..57267195 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -14,6 +14,7 @@ use crate::config::{EngineConfig, LogFormat}; use crate::engine::Engine; +use crate::remote_config::RemoteConfigManager; use clap::{Arg, Command}; use std::process; use tracing::{info, Level}; @@ -21,10 +22,14 @@ use tracing_subscriber::fmt; pub struct Cli { pub config: Option, + pub remote_config_manager: Option, } impl Default for Cli { fn default() -> Self { - Self { config: None } + Self { + config: None, + remote_config_manager: None, + } } } @@ -40,7 +45,7 @@ impl Cli { .long("config") .value_name("FILE") .help("Specify the profile path.") - .required(true), + .required_unless_present("remote-config-url"), ) .arg( Arg::new("validate") @@ -49,38 +54,79 @@ impl Cli { .help("Only the profile is verified, not the engine is started.") .action(clap::ArgAction::SetTrue), ) + .arg( + Arg::new("remote-config-url") + .long("remote-config-url") + .value_name("URL") + .help("Remote configuration API endpoint URL for automatic pipeline management.") + .required_unless_present("config"), + ) + .arg( + Arg::new("remote-config-interval") + .long("remote-config-interval") + .value_name("SECONDS") + .help("Interval in seconds for polling remote configuration (default: 30).") + .default_value("30"), + ) + .arg( + Arg::new("remote-config-token") + .long("remote-config-token") + .value_name("TOKEN") + .help("Authentication token for remote configuration API."), + ) .get_matches(); - // Get the profile path - let config_path = matches.get_one::("config").unwrap(); + // Check if using remote configuration + if let Some(remote_url) = matches.get_one::("remote-config-url") { + // Initialize remote configuration manager + let interval = matches + .get_one::("remote-config-interval") + .unwrap() + .parse::() + .unwrap_or(30); + let token = matches.get_one::("remote-config-token").cloned(); - // Get the profile path - let config = match EngineConfig::from_file(config_path) { - Ok(config) => config, - Err(e) => { - println!("Failed to load configuration file: {}", e); - process::exit(1); + let remote_manager = RemoteConfigManager::new(remote_url.clone(), interval, token); + + self.remote_config_manager = Some(remote_manager); + info!("Using remote configuration from: {}", remote_url); + } else { + // Use local configuration file + let config_path = matches.get_one::("config").unwrap(); + + let config = match EngineConfig::from_file(config_path) { + Ok(config) => config, + Err(e) => { + println!("Failed to load configuration file: {}", e); + process::exit(1); + } + }; + + // If you just verify the configuration, exit it + if matches.get_flag("validate") { + info!("The config is validated."); + return Ok(()); } - }; - // If you just verify the configuration, exit it - if matches.get_flag("validate") { - info!("The config is validated."); - return Ok(()); + self.config = Some(config); } - self.config = Some(config); Ok(()) } pub async fn run(&self) -> Result<(), Box> { - // Initialize the logging system - let config = self.config.clone().unwrap(); - init_logging(&config); - let engine = Engine::new(config); - engine.run().await?; + if let Some(remote_manager) = &self.remote_config_manager { + // Run with remote configuration management + remote_manager.run().await?; + } else { + // Run with local configuration + let config = self.config.clone().unwrap(); + init_logging(&config); + let engine = Engine::new(config); + engine.run().await?; + } Ok(()) } } -fn init_logging(config: &EngineConfig) -> () { +pub fn init_logging(config: &EngineConfig) -> () { let log_level = match config.logging.level.as_str() { "trace" => Level::TRACE, "debug" => Level::DEBUG, diff --git a/crates/arkflow-core/src/lib.rs b/crates/arkflow-core/src/lib.rs index c7f73d0a..570b2485 100644 --- a/crates/arkflow-core/src/lib.rs +++ b/crates/arkflow-core/src/lib.rs @@ -35,6 +35,7 @@ pub mod input; pub mod output; pub mod pipeline; pub mod processor; +pub mod remote_config; pub mod stream; pub mod temporary; diff --git a/crates/arkflow-core/src/remote_config.rs b/crates/arkflow-core/src/remote_config.rs new file mode 100644 index 00000000..49b39ce7 --- /dev/null +++ b/crates/arkflow-core/src/remote_config.rs @@ -0,0 +1,307 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! Remote Configuration Management Module +//! +//! This module provides functionality to automatically fetch configuration from remote APIs +//! and manage stream processing pipelines dynamically. + +use crate::config::{EngineConfig, LoggingConfig}; +use crate::stream::StreamConfig; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::RwLock; +use tokio::time::interval; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +/// Remote configuration response structure +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RemoteConfigResponse { + /// Configuration version for change detection + pub version: String, + /// List of pipeline configurations + pub streams: Vec, +} + +/// Stream information from remote API +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StreamInfo { + /// Unique pipeline identifier + pub id: String, + /// Pipeline name + pub name: String, + /// Pipeline status (active, inactive, deleted) + pub status: StreamStatus, + /// Stream configuration + pub config: StreamConfig, + /// Configuration version + pub version: String, +} + +/// Steam status enumeration +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "lowercase")] +pub enum StreamStatus { + Active, + Inactive, + Deleted, +} + +/// Stream runtime information +#[derive(Debug)] +struct StreamRuntime { + /// Pipeline information + info: StreamInfo, + /// Cancellation token for stopping the pipeline + cancellation_token: CancellationToken, + /// Task handle + handle: Option>, +} + +/// Remote configuration manager +pub struct RemoteConfigManager { + /// Remote API endpoint URL + api_url: String, + /// Polling interval in seconds + poll_interval: u64, + /// Authentication token + auth_token: Option, + /// HTTP client + client: reqwest::Client, + /// Currently running pipelines + streams: Arc>>, + /// Last known configuration version + last_version: Arc>>, +} + +impl RemoteConfigManager { + /// Create a new remote configuration manager + pub fn new(api_url: String, poll_interval: u64, auth_token: Option) -> Self { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .build() + .expect("Failed to create HTTP client"); + + Self { + api_url, + poll_interval, + auth_token, + client, + streams: Arc::new(RwLock::new(HashMap::new())), + last_version: Arc::new(RwLock::new(None)), + } + } + + /// Start the remote configuration management loop + pub async fn run(&self) -> Result<(), Box> { + info!("Starting remote configuration manager"); + info!("Polling interval: {} seconds", self.poll_interval); + info!("API endpoint: {}", self.api_url); + + // Initialize default logging + self.init_default_logging(); + + let mut interval_timer = interval(Duration::from_secs(self.poll_interval)); + + loop { + if let Err(e) = self.fetch_and_update_config().await { + error!("Failed to fetch remote configuration: {}", e); + } + interval_timer.tick().await; + } + } + + /// Initialize default logging configuration + fn init_default_logging(&self) { + let default_config = EngineConfig { + streams: vec![], + logging: LoggingConfig::default(), + health_check: crate::config::HealthCheckConfig::default(), + }; + crate::cli::init_logging(&default_config); + } + + /// Fetch configuration from remote API and update pipelines + async fn fetch_and_update_config(&self) -> Result<(), Box> { + let config = self.fetch_remote_config().await?; + + // Check if configuration has changed + let last_version = self.last_version.read().await; + if let Some(ref last_ver) = *last_version { + if last_ver == &config.version { + // No changes, skip update + return Ok(()); + } + } + drop(last_version); + + info!( + "Configuration changed, updating pipelines (version: {})", + config.version + ); + + // Update pipelines + self.update_streams(config.streams).await?; + + // Update version + let mut last_version = self.last_version.write().await; + *last_version = Some(config.version); + + Ok(()) + } + + /// Fetch configuration from remote API + async fn fetch_remote_config( + &self, + ) -> Result> { + let mut request = self.client.get(&self.api_url); + + // Add authentication header if token is provided + if let Some(ref token) = self.auth_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + + let response = request.send().await?; + + if !response.status().is_success() { + return Err(format!("HTTP error: {}", response.status()).into()); + } + + let config: RemoteConfigResponse = response.json().await?; + Ok(config) + } + + /// Update pipelines based on remote configuration + async fn update_streams( + &self, + new_streams: Vec, + ) -> Result<(), Box> { + let mut streams = self.streams.write().await; + let mut new_streams_ids = std::collections::HashSet::new(); + + // Process new/updated pipelines + for stream_info in new_streams { + new_streams_ids.insert(stream_info.id.clone()); + + match stream_info.status { + StreamStatus::Active => { + if let Some(existing) = streams.get(&stream_info.id) { + // Check if pipeline needs to be restarted + if existing.info.version != stream_info.version { + info!( + "Restarting pipeline '{}' (version: {} -> {})", + stream_info.name, existing.info.version, stream_info.version + ); + + // Stop existing stream + existing.cancellation_token.cancel(); + if let Some(handle) = &existing.handle { + let _ = handle.abort(); + } + + // Start new stream + self.start_stream(&mut streams, stream_info).await?; + } + } else { + // Start new stream + info!("Starting new stream '{}'", stream_info.name); + self.start_stream(&mut streams, stream_info).await?; + } + } + StreamStatus::Inactive => { + if let Some(existing) = streams.get(&stream_info.id) { + info!("Stopping stream '{}'", stream_info.name); + existing.cancellation_token.cancel(); + if let Some(handle) = &existing.handle { + let _ = handle.abort(); + } + streams.remove(&stream_info.id); + } + } + StreamStatus::Deleted => { + if let Some(existing) = streams.get(&stream_info.id) { + info!("Deleting stream '{}'", stream_info.name); + existing.cancellation_token.cancel(); + if let Some(handle) = &existing.handle { + let _ = handle.abort(); + } + streams.remove(&stream_info.id); + } + } + } + } + + // Remove stream that are no longer in the configuration + let current_ids: Vec = streams.keys().cloned().collect(); + for id in current_ids { + if !new_streams_ids.contains(&id) { + if let Some(existing) = streams.get(&id) { + warn!( + "Removing stream '{}' (no longer in remote config)", + existing.info.name + ); + existing.cancellation_token.cancel(); + if let Some(handle) = &existing.handle { + let _ = handle.abort(); + } + } + streams.remove(&id); + } + } + + Ok(()) + } + + /// Start a new stream + async fn start_stream( + &self, + streams: &mut HashMap, + stream_info: StreamInfo, + ) -> Result<(), Box> { + let cancellation_token = CancellationToken::new(); + let token_clone = cancellation_token.clone(); + let config_clone = stream_info.config.clone(); + let stream_name = stream_info.name.clone(); + let stream_id = stream_info.id.clone(); + + // Build and start the stream + let handle = tokio::spawn(async move { + match config_clone.build() { + Ok(mut stream) => { + info!("Stream '{}' started successfully", stream_name); + if let Err(e) = stream.run(token_clone).await { + error!("Stream '{}' error: {}", stream_name, e); + } else { + info!("Stream '{}' completed", stream_name); + } + } + Err(e) => { + error!("Failed to build stream '{}': {}", stream_name, e); + } + } + }); + + let runtime = StreamRuntime { + info: stream_info, + cancellation_token, + handle: Some(handle), + }; + + streams.insert(stream_id, runtime); + Ok(()) + } +} diff --git a/docs/remote-config.md b/docs/remote-config.md new file mode 100644 index 00000000..870e6d22 --- /dev/null +++ b/docs/remote-config.md @@ -0,0 +1,239 @@ +# 远端配置管理功能 + +## 概述 + +远端配置管理功能允许ArkFlow流处理引擎自动从远端API接口拉取配置信息，并动态地创建、修改、删除和启停流处理管道。这个功能特别适用于需要集中管理多个流处理实例的场景。 + +## 功能特性 + +- **自动配置拉取**: 定期从远端API获取最新的管道配置 +- **动态管道管理**: 支持运行时创建、更新、停止和删除管道 +- **版本控制**: 基于配置版本进行变更检测，避免不必要的重启 +- **认证支持**: 支持Bearer Token认证 +- **健康检查**: 保持原有的健康检查功能 +- **错误处理**: 优雅处理网络错误和配置错误 + +## 使用方法 + +### 命令行参数 + +```bash +# 使用远端配置启动 +arkflow --remote-config-url "http://api.example.com/pipelines" \ + --remote-config-interval 30 \ + --remote-config-token "your-auth-token" + +# 参数说明: +# --remote-config-url: 远端配置API的URL地址 +# --remote-config-interval: 轮询间隔（秒），默认30秒 +# --remote-config-token: 认证令牌（可选） +``` + +### 远端API接口规范 + +#### 请求格式 + +``` +GET /pipelines +Authorization: Bearer # 如果提供了token +Content-Type: application/json +``` + +#### 响应格式 + +```json +{ + "version": "v1.0.0", + "pipelines": [ + { + "id": "pipeline-001", + "name": "数据处理管道", + "status": "active", + "version": "1.0.0", + "config": { + "input": { + "type": "generate", + "config": { + "interval": "1s", + "count": 100, + "mapping": "root.message = 'Hello World'" + } + }, + "pipeline": { + "thread_num": 1, + "processors": [ + { + "type": "json", + "config": { + "operator": "select", + "mapping": "root.processed_at = now()" + } + } + ] + }, + "output": { + "type": "stdout", + "config": {} + } + } + } + ] +} +``` + +#### 字段说明 + +- `version`: 配置版本号，用于变更检测 +- `pipelines`: 管道配置数组 + - `id`: 管道唯一标识符 + - `name`: 管道名称 + - `status`: 管道状态 + - `active`: 激活状态，管道将运行 + - `inactive`: 非激活状态，管道将停止 + - `deleted`: 删除状态，管道将被移除 + - `version`: 管道配置版本 + - `config`: 标准的ArkFlow流配置 + +## 管道生命周期管理 + +### 创建管道 + +当远端API返回新的管道配置时，引擎会自动创建并启动该管道。 + +### 更新管道 + +当管道的`version`字段发生变化时，引擎会： +1. 停止当前运行的管道 +2. 使用新配置重新创建管道 +3. 启动新管道 + +### 停止管道 + +当管道状态变为`inactive`时，引擎会停止该管道但保留其配置。 + +### 删除管道 + +当管道状态变为`deleted`或从配置中移除时，引擎会停止并删除该管道。 + +## 配置示例 + +### 本地配置文件 + +当使用远端配置时，本地配置文件主要用于设置日志和健康检查： + +```yaml +logging: + level: info + format: plain + +health_check: + enabled: true + address: "0.0.0.0:8080" + +# 远端配置模式下，streams数组可以为空 +streams: [] +``` + +### 远端API实现示例 + +```python +# Flask示例 +from flask import Flask, jsonify, request + +app = Flask(__name__) + +@app.route('/pipelines') +def get_pipelines(): + # 验证认证令牌 + auth_header = request.headers.get('Authorization') + if auth_header: + token = auth_header.replace('Bearer ', '') + # 验证token逻辑 + + # 返回管道配置 + return jsonify({ + "version": "v1.0.0", + "pipelines": [ + # 管道配置... + ] + }) + +if __name__ == '__main__': + app.run(host='0.0.0.0', port=8000) +``` + +## 监控和日志 + +引擎会记录以下关键事件： + +- 配置拉取成功/失败 +- 管道创建、更新、停止、删除 +- 网络错误和重试 +- 配置解析错误 + +示例日志输出： + +``` +2024-01-15T10:30:00Z INFO Starting remote configuration manager +2024-01-15T10:30:00Z INFO Polling interval: 30 seconds +2024-01-15T10:30:00Z INFO API endpoint: http://api.example.com/pipelines +2024-01-15T10:30:30Z INFO Configuration changed, updating pipelines (version: v1.0.1) +2024-01-15T10:30:30Z INFO Starting new pipeline 'Data Processing Pipeline' +2024-01-15T10:30:30Z INFO Pipeline 'Data Processing Pipeline' started successfully +``` + +## 错误处理 + +### 网络错误 + +当无法连接到远端API时，引擎会： +- 记录错误日志 +- 继续运行现有管道 +- 在下一个轮询周期重试 + +### 配置错误 + +当远端配置格式错误时，引擎会： +- 记录详细错误信息 +- 保持现有管道运行 +- 跳过本次更新 + +### 管道启动失败 + +当管道配置无效导致启动失败时，引擎会： +- 记录错误详情 +- 继续处理其他管道 +- 不影响已运行的管道 + +## 最佳实践 + +1. **版本管理**: 确保每次配置变更都更新版本号 +2. **渐进式部署**: 先在测试环境验证配置，再推送到生产环境 +3. **监控告警**: 监控API可用性和管道健康状态 +4. **备份策略**: 保留配置历史版本以便回滚 +5. **认证安全**: 使用强认证令牌并定期轮换 + +## 故障排除 + +### 常见问题 + +1. **管道无法启动** + - 检查配置格式是否正确 + - 验证输入/输出源的连接性 + - 查看详细错误日志 + +2. **配置未更新** + - 确认API返回了新的版本号 + - 检查网络连接 + - 验证认证令牌 + +3. **性能问题** + - 调整轮询间隔 + - 优化管道配置 + - 监控资源使用情况 + +### 调试技巧 + +- 设置日志级别为`debug`获取详细信息 +- 使用健康检查端点监控状态 +- 检查管道指标和性能数据 \ No newline at end of file From d57ec080cc636c7cd5e3ea5086d65680c5ed40e4 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 21:24:45 +0800 Subject: [PATCH 2/8] x --- Cargo.lock | 148 +++++------------------ crates/arkflow-core/Cargo.toml | 2 +- crates/arkflow-core/src/cli/mod.rs | 34 +++++- crates/arkflow-core/src/engine/mod.rs | 22 +--- crates/arkflow-core/src/remote_config.rs | 16 ++- 5 files changed, 73 insertions(+), 149 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9f6a649d..da284f1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -253,7 +253,7 @@ dependencies = [ "futures", "lazy_static", "num_cpus", - "reqwest 0.11.27", + "reqwest", "serde", "serde_json", "serde_yaml", @@ -299,7 +299,7 @@ dependencies = [ "rdkafka", "rdkafka-sys", "redis", - "reqwest 0.12.22", + "reqwest", "rumqttc", "sasl2-sys", "serde", @@ -553,7 +553,7 @@ version = "55.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af7686986a3bf2254c9fb130c623cdcb2f8e1f15763e7c71c310f0834da3d292" dependencies = [ - "bitflags 2.9.1", + "bitflags", "serde", ] @@ -1115,7 +1115,7 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sync_wrapper 1.0.2", + "sync_wrapper", "tokio", "tower 0.5.2", "tower-layer", @@ -1138,7 +1138,7 @@ dependencies = [ "mime", "pin-project-lite", "rustversion", - "sync_wrapper 1.0.2", + "sync_wrapper", "tower-layer", "tower-service", "tracing", @@ -1369,7 +1369,7 @@ version = "0.69.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271383c67ccabffb7381723dea0672a673f292304fcb45c01cc648c7a8d58088" dependencies = [ - "bitflags 2.9.1", + "bitflags", "cexpr", "clang-sys", "itertools 0.12.1", @@ -1392,7 +1392,7 @@ version = "0.71.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f58bf3d7db68cfbac37cfc485a8d711e87e064c3d0fe0435b92f7a407f9d6b3" dependencies = [ - "bitflags 2.9.1", + "bitflags", "cexpr", "clang-sys", "itertools 0.13.0", @@ -1419,12 +1419,6 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" -[[package]] -name = "bitflags" -version = "1.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" - [[package]] name = "bitflags" version = "2.9.1" @@ -3319,7 +3313,7 @@ version = "25.2.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1045398c1bfd89168b5fd3f1fc11f6e70b34f6f66300c87d44d3de849463abf1" dependencies = [ - "bitflags 2.9.1", + "bitflags", "rustc_version", ] @@ -3740,7 +3734,7 @@ checksum = "2b4cabd42d46aac490d01b80596dfd930a8316b4ee3e8b0c94a7cba888791c0f" dependencies = [ "aes", "base64 0.22.1", - "bitflags 2.9.1", + "bitflags", "bytes", "cbc", "chrono", @@ -4016,19 +4010,6 @@ dependencies = [ "tower-service", ] -[[package]] -name = "hyper-tls" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" -dependencies = [ - "bytes", - "hyper 0.14.32", - "native-tls", - "tokio", - "tokio-native-tls", -] - [[package]] name = "hyper-tls" version = "0.6.0" @@ -4064,7 +4045,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2", - "system-configuration 0.6.1", + "system-configuration", "tokio", "tower-service", "tracing", @@ -4290,7 +4271,7 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b86e202f00093dcba4275d4636b93ef9dd75d025ae560d2521b45ea28ab49013" dependencies = [ - "bitflags 2.9.1", + "bitflags", "cfg-if", "libc", ] @@ -4603,7 +4584,7 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1580801010e535496706ba011c15f8532df6b42297d2e471fec38ceadd8c0638" dependencies = [ - "bitflags 2.9.1", + "bitflags", "libc", "redox_syscall", ] @@ -4937,7 +4918,7 @@ checksum = "6e0ec195e788c95f36b7cf88127d538465fc2f7773e6e47af01834738eab0aee" dependencies = [ "base64 0.22.1", "bigdecimal", - "bitflags 2.9.1", + "bitflags", "btoi", "byteorder", "bytes", @@ -5223,7 +5204,7 @@ dependencies = [ "percent-encoding", "quick-xml", "rand 0.9.1", - "reqwest 0.12.22", + "reqwest", "ring", "rustls-pemfile 2.2.0", "serde", @@ -5276,7 +5257,7 @@ version = "6.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "336b9c63443aceef14bea841b899035ae3abe89b7c486aaf4c5bd8aafedac3f0" dependencies = [ - "bitflags 2.9.1", + "bitflags", "libc", "once_cell", "onig_sys", @@ -5304,7 +5285,7 @@ version = "0.10.73" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8505734d46c8ab1e19a1dce3aef597ad87dcb4c37e7188231769bd6bd51cebf8" dependencies = [ - "bitflags 2.9.1", + "bitflags", "cfg-if", "foreign-types", "libc", @@ -6332,7 +6313,7 @@ version = "0.5.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d04b7d0ee6b4a0207a0a7adb104d23ecb0b47d6beae7152d0fa34b692b29fd6" dependencies = [ - "bitflags 2.9.1", + "bitflags", ] [[package]] @@ -6408,46 +6389,6 @@ dependencies = [ "bytecheck", ] -[[package]] -name = "reqwest" -version = "0.11.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd67538700a17451e7cba03ac727fb961abb7607553461627b97de0b89cf4a62" -dependencies = [ - "base64 0.21.7", - "bytes", - "encoding_rs", - "futures-core", - "futures-util", - "h2 0.3.26", - "http 0.2.12", - "http-body 0.4.6", - "hyper 0.14.32", - "hyper-tls 0.5.0", - "ipnet", - "js-sys", - "log", - "mime", - "native-tls", - "once_cell", - "percent-encoding", - "pin-project-lite", - "rustls-pemfile 1.0.4", - "serde", - "serde_json", - "serde_urlencoded", - "sync_wrapper 0.1.2", - "system-configuration 0.5.1", - "tokio", - "tokio-native-tls", - "tower-service", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", - "winreg", -] - [[package]] name = "reqwest" version = "0.12.22" @@ -6465,7 +6406,7 @@ dependencies = [ "http-body-util", "hyper 1.6.0", "hyper-rustls 0.27.5", - "hyper-tls 0.6.0", + "hyper-tls", "hyper-util", "js-sys", "log", @@ -6480,7 +6421,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", - "sync_wrapper 1.0.2", + "sync_wrapper", "tokio", "tokio-native-tls", "tokio-rustls 0.26.2", @@ -6606,7 +6547,7 @@ version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" dependencies = [ - "bitflags 2.9.1", + "bitflags", "fallible-iterator 0.3.0", "fallible-streaming-iterator", "hashlink 0.9.1", @@ -6663,7 +6604,7 @@ version = "0.38.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" dependencies = [ - "bitflags 2.9.1", + "bitflags", "errno", "libc", "linux-raw-sys 0.4.15", @@ -6676,7 +6617,7 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c71e83d6afe7ff64890ec6b71d6a69bb8a610ab78ce364b3352876bb4c801266" dependencies = [ - "bitflags 2.9.1", + "bitflags", "errno", "libc", "linux-raw-sys 0.9.4", @@ -6959,7 +6900,7 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" dependencies = [ - "bitflags 2.9.1", + "bitflags", "core-foundation 0.9.4", "core-foundation-sys", "libc", @@ -6972,7 +6913,7 @@ version = "3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271720403f46ca04f7ba6f55d438f8bd878d6b8ca0a1046e8228c4145bcbb316" dependencies = [ - "bitflags 2.9.1", + "bitflags", "core-foundation 0.10.0", "core-foundation-sys", "libc", @@ -7475,7 +7416,7 @@ checksum = "0afdd3aa7a629683c2d750c2df343025545087081ab5942593a5288855b1b7a7" dependencies = [ "atoi", "base64 0.22.1", - "bitflags 2.9.1", + "bitflags", "byteorder", "bytes", "crc", @@ -7517,7 +7458,7 @@ checksum = "a0bedbe1bbb5e2615ef347a5e9d8cd7680fb63e77d9dafc0f29be15e53f1ebe6" dependencies = [ "atoi", "base64 0.22.1", - "bitflags 2.9.1", + "bitflags", "byteorder", "crc", "dotenvy", @@ -7702,12 +7643,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "sync_wrapper" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160" - [[package]] name = "sync_wrapper" version = "1.0.2" @@ -7738,36 +7673,15 @@ dependencies = [ "nom", ] -[[package]] -name = "system-configuration" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7" -dependencies = [ - "bitflags 1.3.2", - "core-foundation 0.9.4", - "system-configuration-sys 0.5.0", -] - [[package]] name = "system-configuration" version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" dependencies = [ - "bitflags 2.9.1", + "bitflags", "core-foundation 0.9.4", - "system-configuration-sys 0.6.0", -] - -[[package]] -name = "system-configuration-sys" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9" -dependencies = [ - "core-foundation-sys", - "libc", + "system-configuration-sys", ] [[package]] @@ -8293,7 +8207,7 @@ dependencies = [ "futures-core", "futures-util", "pin-project-lite", - "sync_wrapper 1.0.2", + "sync_wrapper", "tokio", "tower-layer", "tower-service", @@ -8306,7 +8220,7 @@ version = "0.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2" dependencies = [ - "bitflags 2.9.1", + "bitflags", "bytes", "futures-util", "http 1.3.1", @@ -9405,7 +9319,7 @@ version = "0.39.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6f42320e61fe2cfd34354ecb597f86f413484a798ba44a8ca1165c58d42da6c1" dependencies = [ - "bitflags 2.9.1", + "bitflags", ] [[package]] diff --git a/crates/arkflow-core/Cargo.toml b/crates/arkflow-core/Cargo.toml index 58395534..a32df716 100644 --- a/crates/arkflow-core/Cargo.toml +++ b/crates/arkflow-core/Cargo.toml @@ -27,5 +27,5 @@ clap = { workspace = true } colored = { workspace = true } flume = { workspace = true } axum = { workspace = true } -reqwest = { version = "0.11", features = ["json"] } +reqwest = { workspace = true, features = ["json"] } num_cpus = "1.17.0" \ No newline at end of file diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index 57267195..d0932171 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -17,6 +17,8 @@ use crate::engine::Engine; use crate::remote_config::RemoteConfigManager; use clap::{Arg, Command}; use std::process; +use tokio::signal::unix::{signal, SignalKind}; +use tokio_util::sync::CancellationToken; use tracing::{info, Level}; use tracing_subscriber::fmt; @@ -81,8 +83,7 @@ impl Cli { // Initialize remote configuration manager let interval = matches .get_one::("remote-config-interval") - .unwrap() - .parse::() + .and_then(|s| s.parse::().ok()) .unwrap_or(30); let token = matches.get_one::("remote-config-token").cloned(); @@ -92,7 +93,9 @@ impl Cli { info!("Using remote configuration from: {}", remote_url); } else { // Use local configuration file - let config_path = matches.get_one::("config").unwrap(); + let config_path = matches + .get_one::("config") + .ok_or("Configuration not found")?; let config = match EngineConfig::from_file(config_path) { Ok(config) => config, @@ -113,20 +116,39 @@ impl Cli { Ok(()) } pub async fn run(&self) -> Result<(), Box> { + let token = CancellationToken::new(); + if let Some(remote_manager) = &self.remote_config_manager { // Run with remote configuration management - remote_manager.run().await?; + remote_manager.run(token.clone()).await?; } else { // Run with local configuration let config = self.config.clone().unwrap(); init_logging(&config); let engine = Engine::new(config); - engine.run().await?; + engine.run(token.clone()).await?; } + + // Set up signal handlers + let mut sigint = signal(SignalKind::interrupt()).expect("Failed to set signal handler"); + let mut sigterm = signal(SignalKind::terminate()).expect("Failed to set signal handler"); + + tokio::spawn(async move { + tokio::select! { + _ = sigint.recv() => { + info!("Received SIGINT, exiting..."); + + }, + _ = sigterm.recv() => { + info!("Received SIGTERM, exiting..."); + } + } + token.cancel(); + }); Ok(()) } } -pub fn init_logging(config: &EngineConfig) -> () { +pub(crate) fn init_logging(config: &EngineConfig) -> () { let log_level = match config.logging.level.as_str() { "trace" => Level::TRACE, "debug" => Level::DEBUG, diff --git a/crates/arkflow-core/src/engine/mod.rs b/crates/arkflow-core/src/engine/mod.rs index e5f8d3a0..ef5ab9d2 100644 --- a/crates/arkflow-core/src/engine/mod.rs +++ b/crates/arkflow-core/src/engine/mod.rs @@ -16,7 +16,6 @@ use crate::config::EngineConfig; use std::process; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use tokio::signal::unix::{signal, SignalKind}; use tokio_util::sync::CancellationToken; use tracing::{error, info}; @@ -217,9 +216,7 @@ impl Engine { /// 5. Waits for all streams to complete /// /// Returns an error if any part of the initialization or execution fails - pub async fn run(&self) -> Result<(), Box> { - let token = CancellationToken::new(); - + pub async fn run(&self, token: CancellationToken) -> Result<(), Box> { // Start the health check server self.start_health_check_server(token.clone()).await?; @@ -243,23 +240,6 @@ impl Engine { // Set the readiness status self.health_state.is_ready.store(true, Ordering::SeqCst); - // Set up signal handlers - let mut sigint = signal(SignalKind::interrupt()).expect("Failed to set signal handler"); - let mut sigterm = signal(SignalKind::terminate()).expect("Failed to set signal handler"); - let token_clone = token.clone(); - tokio::spawn(async move { - tokio::select! { - _ = sigint.recv() => { - info!("Received SIGINT, exiting..."); - - }, - _ = sigterm.recv() => { - info!("Received SIGTERM, exiting..."); - } - } - - token_clone.cancel(); - }); for (i, mut stream) in streams.into_iter().enumerate() { info!("Starting flow #{}", i + 1); diff --git a/crates/arkflow-core/src/remote_config.rs b/crates/arkflow-core/src/remote_config.rs index 49b39ce7..fe44a2da 100644 --- a/crates/arkflow-core/src/remote_config.rs +++ b/crates/arkflow-core/src/remote_config.rs @@ -107,7 +107,7 @@ impl RemoteConfigManager { } /// Start the remote configuration management loop - pub async fn run(&self) -> Result<(), Box> { + pub async fn run(&self, token: CancellationToken) -> Result<(), Box> { info!("Starting remote configuration manager"); info!("Polling interval: {} seconds", self.poll_interval); info!("API endpoint: {}", self.api_url); @@ -118,11 +118,19 @@ impl RemoteConfigManager { let mut interval_timer = interval(Duration::from_secs(self.poll_interval)); loop { - if let Err(e) = self.fetch_and_update_config().await { - error!("Failed to fetch remote configuration: {}", e); + tokio::select! { + _ = token.cancelled() => { + info!("Shutting down remote configuration manager"); + break; + } + _ = interval_timer.tick() => { + if let Err(e) = self.fetch_and_update_config().await { + error!("Failed to fetch remote configuration: {}", e); + } + } } - interval_timer.tick().await; } + Ok(()) } /// Initialize default logging configuration From 7a7578c7492be5232eabd209ceb54ff3bf9f7af6 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 21:30:07 +0800 Subject: [PATCH 3/8] x --- docs/remote-config.md | 239 ------------------------------------------ 1 file changed, 239 deletions(-) delete mode 100644 docs/remote-config.md diff --git a/docs/remote-config.md b/docs/remote-config.md deleted file mode 100644 index 870e6d22..00000000 --- a/docs/remote-config.md +++ /dev/null @@ -1,239 +0,0 @@ -# 远端配置管理功能 - -## 概述 - -远端配置管理功能允许ArkFlow流处理引擎自动从远端API接口拉取配置信息，并动态地创建、修改、删除和启停流处理管道。这个功能特别适用于需要集中管理多个流处理实例的场景。 - -## 功能特性 - -- **自动配置拉取**: 定期从远端API获取最新的管道配置 -- **动态管道管理**: 支持运行时创建、更新、停止和删除管道 -- **版本控制**: 基于配置版本进行变更检测，避免不必要的重启 -- **认证支持**: 支持Bearer Token认证 -- **健康检查**: 保持原有的健康检查功能 -- **错误处理**: 优雅处理网络错误和配置错误 - -## 使用方法 - -### 命令行参数 - -```bash -# 使用远端配置启动 -arkflow --remote-config-url "http://api.example.com/pipelines" \ - --remote-config-interval 30 \ - --remote-config-token "your-auth-token" - -# 参数说明: -# --remote-config-url: 远端配置API的URL地址 -# --remote-config-interval: 轮询间隔（秒），默认30秒 -# --remote-config-token: 认证令牌（可选） -``` - -### 远端API接口规范 - -#### 请求格式 - -``` -GET /pipelines -Authorization: Bearer # 如果提供了token -Content-Type: application/json -``` - -#### 响应格式 - -```json -{ - "version": "v1.0.0", - "pipelines": [ - { - "id": "pipeline-001", - "name": "数据处理管道", - "status": "active", - "version": "1.0.0", - "config": { - "input": { - "type": "generate", - "config": { - "interval": "1s", - "count": 100, - "mapping": "root.message = 'Hello World'" - } - }, - "pipeline": { - "thread_num": 1, - "processors": [ - { - "type": "json", - "config": { - "operator": "select", - "mapping": "root.processed_at = now()" - } - } - ] - }, - "output": { - "type": "stdout", - "config": {} - } - } - } - ] -} -``` - -#### 字段说明 - -- `version`: 配置版本号，用于变更检测 -- `pipelines`: 管道配置数组 - - `id`: 管道唯一标识符 - - `name`: 管道名称 - - `status`: 管道状态 - - `active`: 激活状态，管道将运行 - - `inactive`: 非激活状态，管道将停止 - - `deleted`: 删除状态，管道将被移除 - - `version`: 管道配置版本 - - `config`: 标准的ArkFlow流配置 - -## 管道生命周期管理 - -### 创建管道 - -当远端API返回新的管道配置时，引擎会自动创建并启动该管道。 - -### 更新管道 - -当管道的`version`字段发生变化时，引擎会： -1. 停止当前运行的管道 -2. 使用新配置重新创建管道 -3. 启动新管道 - -### 停止管道 - -当管道状态变为`inactive`时，引擎会停止该管道但保留其配置。 - -### 删除管道 - -当管道状态变为`deleted`或从配置中移除时，引擎会停止并删除该管道。 - -## 配置示例 - -### 本地配置文件 - -当使用远端配置时，本地配置文件主要用于设置日志和健康检查： - -```yaml -logging: - level: info - format: plain - -health_check: - enabled: true - address: "0.0.0.0:8080" - -# 远端配置模式下，streams数组可以为空 -streams: [] -``` - -### 远端API实现示例 - -```python -# Flask示例 -from flask import Flask, jsonify, request - -app = Flask(__name__) - -@app.route('/pipelines') -def get_pipelines(): - # 验证认证令牌 - auth_header = request.headers.get('Authorization') - if auth_header: - token = auth_header.replace('Bearer ', '') - # 验证token逻辑 - - # 返回管道配置 - return jsonify({ - "version": "v1.0.0", - "pipelines": [ - # 管道配置... - ] - }) - -if __name__ == '__main__': - app.run(host='0.0.0.0', port=8000) -``` - -## 监控和日志 - -引擎会记录以下关键事件： - -- 配置拉取成功/失败 -- 管道创建、更新、停止、删除 -- 网络错误和重试 -- 配置解析错误 - -示例日志输出： - -``` -2024-01-15T10:30:00Z INFO Starting remote configuration manager -2024-01-15T10:30:00Z INFO Polling interval: 30 seconds -2024-01-15T10:30:00Z INFO API endpoint: http://api.example.com/pipelines -2024-01-15T10:30:30Z INFO Configuration changed, updating pipelines (version: v1.0.1) -2024-01-15T10:30:30Z INFO Starting new pipeline 'Data Processing Pipeline' -2024-01-15T10:30:30Z INFO Pipeline 'Data Processing Pipeline' started successfully -``` - -## 错误处理 - -### 网络错误 - -当无法连接到远端API时，引擎会： -- 记录错误日志 -- 继续运行现有管道 -- 在下一个轮询周期重试 - -### 配置错误 - -当远端配置格式错误时，引擎会： -- 记录详细错误信息 -- 保持现有管道运行 -- 跳过本次更新 - -### 管道启动失败 - -当管道配置无效导致启动失败时，引擎会： -- 记录错误详情 -- 继续处理其他管道 -- 不影响已运行的管道 - -## 最佳实践 - -1. **版本管理**: 确保每次配置变更都更新版本号 -2. **渐进式部署**: 先在测试环境验证配置，再推送到生产环境 -3. **监控告警**: 监控API可用性和管道健康状态 -4. **备份策略**: 保留配置历史版本以便回滚 -5. **认证安全**: 使用强认证令牌并定期轮换 - -## 故障排除 - -### 常见问题 - -1. **管道无法启动** - - 检查配置格式是否正确 - - 验证输入/输出源的连接性 - - 查看详细错误日志 - -2. **配置未更新** - - 确认API返回了新的版本号 - - 检查网络连接 - - 验证认证令牌 - -3. **性能问题** - - 调整轮询间隔 - - 优化管道配置 - - 监控资源使用情况 - -### 调试技巧 - -- 设置日志级别为`debug`获取详细信息 -- 使用健康检查端点监控状态 -- 检查管道指标和性能数据 \ No newline at end of file From 679fd93a7dafc8486d5eaf663eddb69dfa7987c2 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 23:16:00 +0800 Subject: [PATCH 4/8] x --- crates/arkflow-core/src/cli/mod.rs | 40 ++++++++++++++++-------------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index d0932171..d4f5ed5f 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -47,7 +47,6 @@ impl Cli { .long("config") .value_name("FILE") .help("Specify the profile path.") - .required_unless_present("remote-config-url"), ) .arg( Arg::new("validate") @@ -56,36 +55,39 @@ impl Cli { .help("Only the profile is verified, not the engine is started.") .action(clap::ArgAction::SetTrue), ) - .arg( + .subcommand( + Command::new("remote") + .arg( Arg::new("remote-config-url") .long("remote-config-url") .value_name("URL") .help("Remote configuration API endpoint URL for automatic pipeline management.") - .required_unless_present("config"), - ) - .arg( - Arg::new("remote-config-interval") - .long("remote-config-interval") - .value_name("SECONDS") - .help("Interval in seconds for polling remote configuration (default: 30).") - .default_value("30"), - ) - .arg( - Arg::new("remote-config-token") - .long("remote-config-token") - .value_name("TOKEN") - .help("Authentication token for remote configuration API."), ) + .arg( + Arg::new("remote-config-interval") + .long("remote-config-interval") + .value_name("SECONDS") + .help("Interval in seconds for polling remote configuration (default: 30).") + .default_value("30"), + ) + .arg( + Arg::new("remote-config-token") + .long("remote-config-token") + .value_name("TOKEN") + .help("Authentication token for remote configuration API."), + )) .get_matches(); // Check if using remote configuration - if let Some(remote_url) = matches.get_one::("remote-config-url") { + if let Some(remote) = matches.subcommand_matches("remote") { // Initialize remote configuration manager - let interval = matches + matches.subcommand_matches("remote"); + let interval = remote .get_one::("remote-config-interval") .and_then(|s| s.parse::().ok()) .unwrap_or(30); - let token = matches.get_one::("remote-config-token").cloned(); + let token = remote.get_one::("remote-config-token").cloned(); + let remote_url = remote.get_one::("remote-config-url").expect("Remote configuration URL not found"); let remote_manager = RemoteConfigManager::new(remote_url.clone(), interval, token); From 17da89684fa007ee061ee257a0fd6decff698a52 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 23:16:50 +0800 Subject: [PATCH 5/8] x --- crates/arkflow-core/src/cli/mod.rs | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index d4f5ed5f..b2cfef58 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -58,21 +58,21 @@ impl Cli { .subcommand( Command::new("remote") .arg( - Arg::new("remote-config-url") - .long("remote-config-url") + Arg::new("config-url") + .long("config-url") .value_name("URL") .help("Remote configuration API endpoint URL for automatic pipeline management.") ) .arg( - Arg::new("remote-config-interval") - .long("remote-config-interval") + Arg::new("config-interval") + .long("config-interval") .value_name("SECONDS") .help("Interval in seconds for polling remote configuration (default: 30).") .default_value("30"), ) .arg( - Arg::new("remote-config-token") - .long("remote-config-token") + Arg::new("config-token") + .long("config-token") .value_name("TOKEN") .help("Authentication token for remote configuration API."), )) @@ -83,11 +83,13 @@ impl Cli { // Initialize remote configuration manager matches.subcommand_matches("remote"); let interval = remote - .get_one::("remote-config-interval") + .get_one::("config-interval") .and_then(|s| s.parse::().ok()) .unwrap_or(30); - let token = remote.get_one::("remote-config-token").cloned(); - let remote_url = remote.get_one::("remote-config-url").expect("Remote configuration URL not found"); + let token = remote.get_one::("config-token").cloned(); + let remote_url = remote + .get_one::("config-url") + .expect("Remote configuration URL not found"); let remote_manager = RemoteConfigManager::new(remote_url.clone(), interval, token); From 622819b711e224babf68d20dc5f45c4d45e8d5b1 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 23:22:33 +0800 Subject: [PATCH 6/8] x --- crates/arkflow-core/src/cli/mod.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index b2cfef58..854ebd6a 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -56,32 +56,33 @@ impl Cli { .action(clap::ArgAction::SetTrue), ) .subcommand( - Command::new("remote") + Command::new("remote").about("Use remote configuration for automatic stream management.") .arg( Arg::new("config-url") .long("config-url") .value_name("URL") - .help("Remote configuration API endpoint URL for automatic pipeline management.") + .help("Remote configuration API endpoint URL for automatic stream management.") + .required( true) ) .arg( Arg::new("config-interval") .long("config-interval") .value_name("SECONDS") - .help("Interval in seconds for polling remote configuration (default: 30).") + .help("Interval in seconds for polling remote configuration.") .default_value("30"), ) .arg( Arg::new("config-token") .long("config-token") .value_name("TOKEN") - .help("Authentication token for remote configuration API."), + .help("Authentication token for remote configuration API.") + .required( true), )) .get_matches(); // Check if using remote configuration if let Some(remote) = matches.subcommand_matches("remote") { // Initialize remote configuration manager - matches.subcommand_matches("remote"); let interval = remote .get_one::("config-interval") .and_then(|s| s.parse::().ok()) @@ -93,7 +94,7 @@ impl Cli { let remote_manager = RemoteConfigManager::new(remote_url.clone(), interval, token); - self.remote_config_manager = Some(remote_manager); + self.remote_config_manager.replace(remote_manager); info!("Using remote configuration from: {}", remote_url); } else { // Use local configuration file From 2628e438d8144a96d7b3c01c533927d12d5de8f3 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 23:26:38 +0800 Subject: [PATCH 7/8] x --- crates/arkflow-core/src/cli/mod.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index 854ebd6a..3d73c9cb 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -58,22 +58,22 @@ impl Cli { .subcommand( Command::new("remote").about("Use remote configuration for automatic stream management.") .arg( - Arg::new("config-url") - .long("config-url") + Arg::new("url") + .long("url") .value_name("URL") .help("Remote configuration API endpoint URL for automatic stream management.") .required( true) ) .arg( - Arg::new("config-interval") - .long("config-interval") + Arg::new("interval") + .long("interval") .value_name("SECONDS") .help("Interval in seconds for polling remote configuration.") .default_value("30"), ) .arg( - Arg::new("config-token") - .long("config-token") + Arg::new("token") + .long("token") .value_name("TOKEN") .help("Authentication token for remote configuration API.") .required( true), @@ -84,12 +84,12 @@ impl Cli { if let Some(remote) = matches.subcommand_matches("remote") { // Initialize remote configuration manager let interval = remote - .get_one::("config-interval") + .get_one::("interval") .and_then(|s| s.parse::().ok()) .unwrap_or(30); - let token = remote.get_one::("config-token").cloned(); + let token = remote.get_one::("token").cloned(); let remote_url = remote - .get_one::("config-url") + .get_one::("url") .expect("Remote configuration URL not found"); let remote_manager = RemoteConfigManager::new(remote_url.clone(), interval, token); From 716096481c55560fab45ff30e9b1e3dff56f61f3 Mon Sep 17 00:00:00 2001 From: chenquan Date: Wed, 9 Jul 2025 23:29:42 +0800 Subject: [PATCH 8/8] x --- crates/arkflow-core/src/cli/mod.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/crates/arkflow-core/src/cli/mod.rs b/crates/arkflow-core/src/cli/mod.rs index 3d73c9cb..b51dc071 100644 --- a/crates/arkflow-core/src/cli/mod.rs +++ b/crates/arkflow-core/src/cli/mod.rs @@ -40,42 +40,42 @@ impl Cli { let matches = Command::new("arkflow") .version("0.4.0-rc1") .author("chenquan") - .about("High-performance Rust stream processing engine, providing powerful data stream processing capabilities, supporting multiple input/output sources and processors.") + .about("High-performance Rust stream processing engine, providing powerful data stream processing capabilities, supporting multiple input/output sources and processors") .arg( Arg::new("config") .short('c') .long("config") .value_name("FILE") - .help("Specify the profile path.") + .help("Specify the profile path") ) .arg( Arg::new("validate") .short('v') .long("validate") - .help("Only the profile is verified, not the engine is started.") + .help("Only the profile is verified, not the engine is started") .action(clap::ArgAction::SetTrue), ) .subcommand( - Command::new("remote").about("Use remote configuration for automatic stream management.") + Command::new("remote").about("Use remote configuration for automatic stream management") .arg( Arg::new("url") .long("url") .value_name("URL") - .help("Remote configuration API endpoint URL for automatic stream management.") + .help("Remote configuration API endpoint URL for automatic stream management") .required( true) ) .arg( Arg::new("interval") .long("interval") .value_name("SECONDS") - .help("Interval in seconds for polling remote configuration.") + .help("Interval in seconds for polling remote configuration") .default_value("30"), ) .arg( Arg::new("token") .long("token") .value_name("TOKEN") - .help("Authentication token for remote configuration API.") + .help("Authentication token for remote configuration API") .required( true), )) .get_matches();