diff --git a/Cargo.lock b/Cargo.lock index 39dca822..4fc3c7ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -604,7 +604,6 @@ dependencies = [ "cpex-tool-namespace", "http", "jsonwebtoken", - "num_cpus", "openport", "opentelemetry", "opentelemetry-otlp", @@ -619,7 +618,6 @@ dependencies = [ "tokio", "tonic", "tracing", - "tracing-appender", "tracing-opentelemetry", "tracing-subscriber", ] @@ -831,21 +829,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "crossbeam-channel" -version = "0.5.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" -dependencies = [ - "crossbeam-utils", -] - -[[package]] -name = "crossbeam-utils" -version = "0.8.22" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" - [[package]] name = "crypto-common" version = "0.1.7" @@ -1234,12 +1217,6 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "hermit-abi" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" - [[package]] name = "http" version = "1.5.0" @@ -1795,16 +1772,6 @@ dependencies = [ "autocfg", ] -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", -] - [[package]] name = "oauth2" version = "5.0.0" @@ -2887,12 +2854,6 @@ dependencies = [ "rand 0.8.5", ] -[[package]] -name = "symlink" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" - [[package]] name = "syn" version = "2.0.119" @@ -3197,6 +3158,7 @@ dependencies = [ "socket2", "sync_wrapper", "tokio", + "tokio-rustls", "tokio-stream", "tower", "tower-layer", @@ -3318,19 +3280,6 @@ dependencies = [ "tracing-core", ] -[[package]] -name = "tracing-appender" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" -dependencies = [ - "crossbeam-channel", - "symlink", - "thiserror 2.0.19", - "time", - "tracing-subscriber", -] - [[package]] name = "tracing-attributes" version = "0.1.31" diff --git a/crates/contextforge-data-plane-cpex/src/cmf.rs b/crates/contextforge-data-plane-cpex/src/cmf.rs index 0f24a6a7..aa39e33c 100644 --- a/crates/contextforge-data-plane-cpex/src/cmf.rs +++ b/crates/contextforge-data-plane-cpex/src/cmf.rs @@ -8,7 +8,7 @@ use cpex::cpex_core::{ }; use rmcp::{ErrorData, model::ErrorCode}; -#[derive(Clone, Copy)] +#[derive(Clone, Debug, Copy)] pub(crate) enum Operation { Tool, Prompt, diff --git a/crates/contextforge-data-plane-cpex/src/resources/mod.rs b/crates/contextforge-data-plane-cpex/src/resources/mod.rs index 190a16fc..069adc08 100644 --- a/crates/contextforge-data-plane-cpex/src/resources/mod.rs +++ b/crates/contextforge-data-plane-cpex/src/resources/mod.rs @@ -128,7 +128,7 @@ impl GatewayPluginRuntimeHandle { .current()? .before( Operation::Resource, - "", + resource_uri, |id| resource_request_payload(resource_uri, id), |payload, _| { let [ContentPart::ResourceRef { content }] = payload.message.content.as_slice() else { diff --git a/crates/contextforge-data-plane-cpex/src/runtime.rs b/crates/contextforge-data-plane-cpex/src/runtime.rs index d82792b6..b9f8d543 100644 --- a/crates/contextforge-data-plane-cpex/src/runtime.rs +++ b/crates/contextforge-data-plane-cpex/src/runtime.rs @@ -14,6 +14,7 @@ use cpex::cpex_core::{ manager::PluginManager, }; use rmcp::ErrorData; +use tracing::instrument; use crate::{ cmf::{CmfResponse, Operation, modified_message_payload, plugin_denied_error}, @@ -61,6 +62,7 @@ impl GatewayPluginRuntime { Ok(Self { manager, hooks }) } + #[instrument(name = "cmf_plugin_before", level = "info", skip(self, payload, update))] pub(crate) async fn before( self: &Arc, operation: Operation, @@ -92,6 +94,7 @@ impl GatewayPluginRuntime { Ok((update, state)) } + #[instrument(name = "cmf_plugin_invoke", level = "info", skip(self, payload, context_table))] async fn invoke( &self, hook: &'static str, diff --git a/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs index b3de29c4..0f1066af 100644 --- a/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs +++ b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs @@ -6,6 +6,7 @@ use std::fmt; use std::net::IpAddr; use std::path::{Path, PathBuf}; use std::time::Duration; +use tracing::instrument; use url::Url; const JWKS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); @@ -50,6 +51,7 @@ impl fmt::Debug for JwtAuthorizationService { #[async_trait] impl AuthorizationService for JwtAuthorizationService { + #[instrument(name = "jwt_authorization_service", level = "info", skip_all)] async fn authorize(&self, authorization_token: &http::HeaderValue) -> Option { let token = authorization_token.as_bytes().strip_prefix(b"Bearer ")?; let token = str::from_utf8(token).ok()?; diff --git a/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/cel_principal_extractor.rs b/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/cel_principal_extractor.rs index aaf9d8f0..6c56ea46 100644 --- a/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/cel_principal_extractor.rs +++ b/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/cel_principal_extractor.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use cel::{Context, Program, objects::Key}; use serde_json::Value as JsonValue; use thiserror::Error; -use tracing::debug; +use tracing::{debug, instrument}; use crate::authorization::{AuthorizedPrincipal, PrincipalExtractor}; @@ -51,7 +51,7 @@ pub enum CelPrincipalExtractorError { /// "scopes": [] /// } /// ``` -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct CelPrincipalExtractor { program: Arc, } @@ -71,6 +71,7 @@ impl CelPrincipalExtractor { } impl PrincipalExtractor for CelPrincipalExtractor { + #[instrument(name = "principal_extract", level = "info", skip_all)] fn extract( &self, claims: &serde_json::Value, diff --git a/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/default_principal_extractor.rs b/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/default_principal_extractor.rs index a8a9488a..7cae3c4d 100644 --- a/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/default_principal_extractor.rs +++ b/crates/contextforge-data-plane-lib/src/authorization/principal_extractor/default_principal_extractor.rs @@ -1,9 +1,12 @@ +use tracing::instrument; + use crate::authorization::{AuthorizedPrincipal, PrincipalExtractor}; #[derive(Debug, Clone)] pub struct DefaultPrincipalExtractor {} impl PrincipalExtractor for DefaultPrincipalExtractor { + #[instrument(name = "principal_extract", level = "info", skip_all)] fn extract( &self, claims: &serde_json::Value, diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/mod.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/mod.rs index 501bbccd..9f4f6675 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/mod.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/mod.rs @@ -24,6 +24,12 @@ pub struct McpService { plugin_runtime: Option, } +#[allow(clippy::missing_fields_in_debug)] +impl std::fmt::Debug for McpService { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("McpService").field("http_client", &self.http_client).finish() + } +} #[allow(clippy::unused_async_trait_impl)] impl ServerHandler for McpService { async fn initialize( diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs index 0260ce7f..306139f5 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs @@ -4,7 +4,7 @@ use rmcp::{ model::{ErrorCode, GetPromptRequestParams, GetPromptResponse}, service::RequestContext, }; -use tracing::info; +use tracing::{info, instrument}; use super::McpService; use crate::gateway::{ @@ -12,6 +12,7 @@ use crate::gateway::{ routing_error::backend_forward_error, }; +#[instrument(name = "get_prompt", level = "info", skip_all)] pub(super) async fn get_prompt( mcp_service: &McpService, request: GetPromptRequestParams, diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs index 73ce227b..b3471ba4 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs @@ -5,7 +5,7 @@ use rmcp::{ model::{CallToolRequestParams, CallToolResponse, ErrorCode, ProtocolVersion}, service::RequestContext, }; -use tracing::{info, warn}; +use tracing::{info, instrument, warn}; use super::McpService; use crate::gateway::{ @@ -14,6 +14,7 @@ use crate::gateway::{ }; use crate::mcp_standard_headers; +#[instrument(name = "call_tool", level = "info", skip_all)] pub(super) async fn call_tool( mcp_service: &McpService, request: CallToolRequestParams, diff --git a/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs b/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs index 87e18c1c..a1c37a35 100644 --- a/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs +++ b/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs @@ -5,19 +5,6 @@ use tracing::debug; use crate::{AuthorizationClaims, authorization::PrincipalExtractor, errors::unauthorized_response}; -// pub async fn principal_extractor_layer(request: http::Request, next: Next) -> Response { -// let maybe_claims = request.extensions().get::(); -// let Some(Ok(authorized_principal)) = maybe_claims.map(|claims| { -// AuthorizedPrincipal::try_from(claims).inspect_err(|e| debug!("Can't extract the principal {e:?}")) -// }) else { -// return unauthorized_response("Invalid token. Unable to extract the principal from claims"); -// }; -// let (mut parts, body) = request.into_parts(); -// parts.extensions.insert(authorized_principal); -// let request = Request::from_parts(parts, body); -// next.run(request).await -// } - use std::task::{Context, Poll}; use tower::{Layer, Service}; diff --git a/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs b/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs index 442675bb..7c07e0cb 100644 --- a/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs +++ b/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs @@ -12,7 +12,7 @@ use redis::{ cmd, }; use tokio::sync::Mutex; -use tracing::{debug, warn}; +use tracing::{debug, instrument, warn}; use super::{ConfigStoreError, UserConfigStore}; use crate::{ @@ -67,6 +67,7 @@ impl RedisUserConfigStore { #[async_trait] impl UserConfigStore for RedisUserConfigStore { + #[instrument(name = "user_config_store_get_config", level = "info", skip(self))] async fn get_config<'a>(&self, user_key: &'a User) -> Result { let subject = user_key.key(); @@ -136,6 +137,7 @@ impl UserConfigStore for RedisUserConfigStore { Ok(user_config) } + #[instrument(name = "user_config_store_set_config", level = "info", skip(self, config))] async fn set_config<'a>(&self, user_key: &'a User, config: &'a UserConfig) -> Result<(), ConfigStoreError> { let subject = user_key.key(); diff --git a/crates/contextforge-data-plane/Cargo.toml b/crates/contextforge-data-plane/Cargo.toml index e8096061..1bc19c56 100644 --- a/crates/contextforge-data-plane/Cargo.toml +++ b/crates/contextforge-data-plane/Cargo.toml @@ -21,15 +21,13 @@ cpex-tool-namespace = { git = "https://github.com/contextforge-gateway-rs/cpex-p cpex-secrets-detection = { workspace = true, optional = true } clap.workspace = true tracing.workspace = true -tracing-appender = "0.2.3" tracing-opentelemetry.workspace = true tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio.workspace = true opentelemetry.workspace = true -opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics"] } +opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics","tls-aws-lc"] } opentelemetry_sdk.workspace = true tonic = "0.14" -num_cpus = "1.17.0" rmcp.workspace = true tikv-jemallocator = "0.7.0" rustls.workspace = true diff --git a/crates/contextforge-data-plane/src/logging.rs b/crates/contextforge-data-plane/src/logging.rs index 08d5b8e3..225f00cf 100644 --- a/crates/contextforge-data-plane/src/logging.rs +++ b/crates/contextforge-data-plane/src/logging.rs @@ -1,13 +1,12 @@ use std::collections::HashMap; -use contextforge_data_plane_lib::{Config, LogRotation, OtlpProtocol}; +use contextforge_data_plane_lib::{Config, OtlpProtocol}; use opentelemetry::global; use opentelemetry::trace::TracerProvider; use opentelemetry_otlp::{MetricExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler}; use tonic::metadata::{MetadataKey, MetadataMap, MetadataValue}; -use tracing_appender::non_blocking::WorkerGuard; use tracing_subscriber::{ Layer, Registry, filter, fmt::{self, format::FmtSpan}, @@ -15,14 +14,8 @@ use tracing_subscriber::{ util::SubscriberInitExt, }; -/// Holds RAII handles whose lifetime must match the process so background -/// telemetry tasks keep running. The file appender's worker thread needs -/// the [`WorkerGuard`] to stay alive to flush logs on shutdown, and the -/// metrics [`SdkMeterProvider`] needs to stay alive so its -/// [`PeriodicReader`] task keeps exporting at the configured interval. #[allow(dead_code)] pub struct Guard { - appender: WorkerGuard, meter_provider: Option, } @@ -33,24 +26,11 @@ const DEFAULT_HTTP_TRACES_ENDPOINT: &str = "http://127.0.0.1:4318/v1/traces"; const DEFAULT_HTTP_METRICS_ENDPOINT: &str = "http://127.0.0.1:4318/v1/metrics"; const METRICS_EXPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30); -const DEFAULT_LOGGING: &str = "debug,hyper_util=OFF,tower_http=OFF,rmcp=warn,reqwest=warn,rustls=WARN"; +const DEFAULT_LOGGING: &str = "debug,hyper_util=OFF,tower_http=OFF,rmcp=warn,reqwest=warn,rustls=WARN,h2=WARN,opentelemetry_sdk=WARN,opentelemetry-otlp=WARN"; pub fn init_tracing_logging(configuration: &Config) -> Result> { let registry = Registry::default(); - let log_name = configuration.log_name.clone().unwrap_or("contextforge-data-plane.log".to_owned()); - - let file_appender = match configuration.log_rotation.clone().unwrap_or_default() { - LogRotation::Minutely => tracing_appender::rolling::minutely(".", log_name), - LogRotation::Hourly => tracing_appender::rolling::hourly(".", log_name), - LogRotation::Daily => tracing_appender::rolling::daily(".", log_name), - LogRotation::Never => tracing_appender::rolling::never(".", log_name), - }; - - let (non_blocking_appender, guard) = tracing_appender::non_blocking(file_appender); - let file_filter = tracing_subscriber::EnvFilter::new( - std::env::var("RUST_FILE_LOG").unwrap_or_else(|_| DEFAULT_LOGGING.to_owned()), - ); let console_filter = tracing_subscriber::EnvFilter::new(std::env::var("RUST_LOG").unwrap_or_else(|_| DEFAULT_LOGGING.to_owned())); let tracing_filter = tracing_subscriber::EnvFilter::new( @@ -65,14 +45,6 @@ pub fn init_tracing_logging(configuration: &Config) -> Result Result Result<(), Box> { +#[tokio::main] +async fn main() -> Result<(), Box> { let provider = crypto::ring::default_provider(); _ = provider.install_default(); @@ -25,8 +27,6 @@ fn main() -> Result<(), Box> { let _guard = logging::init_tracing_logging(&config)?; info!("starting contextforge-data-plane {config:?}"); - let runtime = runtime::Runtime::from(&config); - let plugin_registry = if config.runtime_plugins_enabled.unwrap_or(false) { Some(Arc::new(plugin_runtime_from_config(&config)?)) } else { @@ -44,7 +44,8 @@ fn main() -> Result<(), Box> { .with_authorization_service(authorization_service) .build(); - runtime.execute(gateway, plugin_registry) + let _cpex_watcher = initialize_cpex_runtime(plugin_registry).await?; + run_gateway(gateway).await } fn plugin_runtime_from_config( @@ -74,3 +75,35 @@ fn register_builtin_factories( } Ok(plugin_runtime) } + +pub async fn initialize_cpex_runtime( + cpex_runtime: Option>, +) -> contextforge_data_plane_lib::Result>> { + let Some(cpex_runtime) = cpex_runtime else { + return Ok(None); + }; + match cpex_runtime.initialize().await { + Ok(Some(handle)) => { + debug!("CPEX Plugins initialization successful"); + Ok(Some(handle)) + }, + Ok(None) => { + debug!("CPEX Plugins initialization skipped"); + Ok(None) + }, + Err(e) => { + error!("CPEX Plugins initialization failed {e:?}"); + Err(e) + }, + } +} + +pub async fn run_gateway(gateway: Gateway) -> contextforge_data_plane_lib::Result<()> { + let res = gateway.run_gateway().await; + if res.is_ok() { + debug!("Gateway process terminated"); + } else { + error!("Gateway process terminated {res:?}"); + } + Ok(()) +} diff --git a/crates/contextforge-data-plane/src/runtime.rs b/crates/contextforge-data-plane/src/runtime.rs deleted file mode 100644 index 71e87b55..00000000 --- a/crates/contextforge-data-plane/src/runtime.rs +++ /dev/null @@ -1,180 +0,0 @@ -use std::{ - sync::{Arc, mpsc}, - thread, -}; - -use contextforge_data_plane_cpex::CpexRuntimeRegistry; -use contextforge_data_plane_lib::{Config, Gateway}; -use tokio::runtime::{Builder, LocalOptions}; -use tokio::task::JoinHandle; -use tracing::{debug, error, info, warn}; - -#[derive(Debug, Clone)] -#[allow(clippy::struct_field_names)] -pub struct Runtime { - single_runtime: bool, - number_of_threads: usize, - global_queue_interval: Option, - event_interval: Option, - max_io_events_per_tick: Option, - thread_name: String, -} - -impl<'b> From<&'b Config> for Runtime { - fn from(config: &'b Config) -> Self { - Self { - single_runtime: config.single_runtime.unwrap_or(true), - number_of_threads: config.number_of_cpus.unwrap_or(num_cpus::get()), - ..Default::default() - } - } -} - -impl Default for Runtime { - fn default() -> Self { - Self { - single_runtime: true, - number_of_threads: num_cpus::get(), - global_queue_interval: Option::default(), - event_interval: Option::default(), - max_io_events_per_tick: Option::default(), - thread_name: "contextforge-data-plane-runtime".to_owned(), - } - } -} - -impl Runtime { - fn configure_builder(&self, builder: &mut Builder, thread_name: String) { - let builder = builder.enable_all().name(thread_name); - let builder = if let Some(global_queue_interval) = self.global_queue_interval { - builder.global_queue_interval(global_queue_interval) - } else { - builder - }; - - if let Some(event_interval) = self.event_interval { - builder.event_interval(event_interval); - } - - if let Some(max_io_events_per_tick) = self.max_io_events_per_tick { - builder.max_io_events_per_tick(max_io_events_per_tick); - } - } - - fn configure_single_thread_builder(builder: &mut Builder, thread_name: String) { - builder.enable_all().name(thread_name).global_queue_interval(1024).max_io_events_per_tick(4); - } - - pub fn execute( - self, - gateway: Gateway, - cpex_runtime: Option>, - ) -> contextforge_data_plane_lib::Result<()> { - if self.single_runtime { - let mut builder = Builder::new_multi_thread(); - self.configure_builder(&mut builder, self.thread_name.clone()); - let runtime = builder.build()?; - - runtime.block_on(async { - let _cpex_watcher = Self::initialize_cpex_runtime(cpex_runtime).await?; - Self::run_gateway(gateway).await - }) - } else { - let (init_sender, init_receiver) = mpsc::channel(); - let mut handles = vec![Self::spawn_gateway_thread( - format!("{}0", self.thread_name), - gateway.clone(), - cpex_runtime, - Some(init_sender), - )?]; - match init_receiver - .recv() - .map_err(|error| format!("CPEX plugin initialization result unavailable: {error}"))? - { - Ok(()) => {}, - Err(error) => return Err(error.into()), - } - - for i in 1..self.number_of_threads { - match Self::spawn_gateway_thread(format!("{}{i}", self.thread_name), gateway.clone(), None, None) { - Ok(handle) => handles.push(handle), - Err(error) => warn!("Thread terminated at start with {error:?}"), - } - } - - for handle in handles { - let res = handle.join(); - info!("Thread terminated with {res:?}"); - } - Ok(()) - } - } - - fn spawn_gateway_thread( - thread_name: String, - gateway: Gateway, - cpex_runtime: Option>, - init_sender: Option>>, - ) -> std::io::Result>> { - thread::Builder::new().name(thread_name.clone()).spawn(move || { - let mut builder = Builder::new_current_thread(); - Self::configure_single_thread_builder(&mut builder, thread_name); - let runtime = match builder.build_local(LocalOptions::default()) { - Ok(runtime) => runtime, - Err(error) => { - warn!("Can't build thread {error:?}"); - return Err::<(), contextforge_data_plane_lib::Error>(error.into()); - }, - }; - - runtime.block_on(async { - let Some(init_sender) = init_sender else { - return Self::run_gateway(gateway).await; - }; - match Self::initialize_cpex_runtime(cpex_runtime).await { - Ok(cpex_watcher) => { - let _ = init_sender.send(Ok(())); - let _cpex_watcher = cpex_watcher; - Self::run_gateway(gateway).await - }, - Err(error) => { - let _ = init_sender.send(Err(error.to_string())); - Err(error) - }, - } - }) - }) - } - - async fn initialize_cpex_runtime( - cpex_runtime: Option>, - ) -> contextforge_data_plane_lib::Result>> { - let Some(cpex_runtime) = cpex_runtime else { - return Ok(None); - }; - match cpex_runtime.initialize().await { - Ok(Some(handle)) => { - debug!("CPEX Plugins initialization successful"); - Ok(Some(handle)) - }, - Ok(None) => { - debug!("CPEX Plugins initialization skipped"); - Ok(None) - }, - Err(e) => { - error!("CPEX Plugins initialization failed {e:?}"); - Err(e) - }, - } - } - - async fn run_gateway(gateway: Gateway) -> contextforge_data_plane_lib::Result<()> { - let res = gateway.run_gateway().await; - if res.is_ok() { - debug!("Gateway process terminated"); - } else { - error!("Gateway process terminated {res:?}"); - } - Ok(()) - } -}