From 304aab5b21544fb74fcba08568f495a1a8c7a69c Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 22 Jun 2026 18:45:32 +0200 Subject: [PATCH] drop: removed runtime usage service --- crates/taurus/src/app/mod.rs | 12 +----- crates/taurus/src/app/worker.rs | 25 ++---------- crates/taurus/src/client/mod.rs | 1 - crates/taurus/src/client/runtime_usage.rs | 48 ----------------------- docs/dev.md | 6 +-- 5 files changed, 6 insertions(+), 86 deletions(-) delete mode 100644 crates/taurus/src/client/runtime_usage.rs diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs index 552e3d9..78a341b 100644 --- a/crates/taurus/src/app/mod.rs +++ b/crates/taurus/src/app/mod.rs @@ -17,7 +17,6 @@ use tucana::shared::module_status::StatusVariant; use crate::client::runtime_execution::TaurusRuntimeExecutionService; use crate::client::runtime_status::TaurusRuntimeStatusService; -use crate::client::runtime_usage::TaurusRuntimeUsageService; use crate::config::Config; pub async fn run() { @@ -32,7 +31,6 @@ pub async fn run() { let ( runtime_status_service, runtime_execution_service, - runtime_usage_service, mut runtime_status_heartbeat_task, ) = setup_dynamic_services_if_needed(&config).await; @@ -44,7 +42,6 @@ pub async fn run() { nats_remote, runtime_emitter, runtime_execution_service, - runtime_usage_service, ); wait_for_shutdown(&mut worker_task, &mut health_task).await; @@ -112,11 +109,10 @@ async fn setup_dynamic_services_if_needed( ) -> ( Option>, Option, - Option, Option>, ) { if config.mode != DYNAMIC { - return (None, None, None, None); + return (None, None, None); } push_definitions_until_success(config).await; @@ -129,11 +125,6 @@ async fn setup_dynamic_services_if_needed( .await, ); - let runtime_usage_service = Some( - TaurusRuntimeUsageService::from_url(config.aquila_url.clone(), config.aquila_token.clone()) - .await, - ); - let runtime_status_service = Some(Arc::new( TaurusRuntimeStatusService::from_url( config.aquila_url.clone(), @@ -184,7 +175,6 @@ async fn setup_dynamic_services_if_needed( ( runtime_status_service, runtime_execution_service, - runtime_usage_service, runtime_status_heartbeat_task, ) } diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 1dc095e..f45211e 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -11,10 +11,9 @@ use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter; use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; use tokio::task::JoinHandle; use tucana::shared::execution_result; -use tucana::shared::{ExecutionFlow, ExecutionResult, NodeExecutionResult, RuntimeUsage, Value}; +use tucana::shared::{ExecutionFlow, ExecutionResult, NodeExecutionResult, Value}; use crate::client::runtime_execution::TaurusRuntimeExecutionService; -use crate::client::runtime_usage::TaurusRuntimeUsageService; pub fn spawn_worker( client: async_nats::Client, @@ -22,7 +21,6 @@ pub fn spawn_worker( nats_remote: NATSRemoteRuntime, runtime_emitter: NATSRespondEmitter, mut runtime_execution_service: Option, - runtime_usage_service: Option, ) -> JoinHandle<()> { tokio::spawn(async move { let mut execution_subscription = match client @@ -52,7 +50,6 @@ pub fn spawn_worker( &nats_remote, &runtime_emitter, runtime_execution_service.as_mut(), - runtime_usage_service.as_ref(), ).await; } None => { @@ -74,7 +71,6 @@ async fn process_execution_message( nats_remote: &NATSRemoteRuntime, runtime_emitter: &NATSRespondEmitter, mut runtime_execution_service: Option<&mut TaurusRuntimeExecutionService>, - runtime_usage_service: Option<&TaurusRuntimeUsageService>, ) { let requested_execution_id = parse_execution_id_from_subject(&message.subject, "execution") .unwrap_or_else(|| { @@ -138,12 +134,6 @@ async fn process_execution_message( .update_runtime_execution(execution_result) .await; } - - if let Some(usage_service) = runtime_usage_service { - usage_service - .update_runtime_usage(run_result.runtime_usage) - .await; - } } #[derive(Clone)] @@ -155,7 +145,6 @@ struct FlowRunResult { input: Option, signal: Signal, node_execution_results: Vec, - runtime_usage: RuntimeUsage, } async fn execute_flow( @@ -166,7 +155,6 @@ async fn execute_flow( respond_emitter: Option<&dyn RespondEmitter>, ) -> FlowRunResult { let started_at = now_unix_micros(); - let start = Instant::now(); let flow_id = flow.flow_id; let input = flow.input_value.clone(); let report = engine @@ -179,7 +167,6 @@ async fn execute_flow( ) .await; let finished_at = now_unix_micros(); - let duration_micros = start.elapsed().as_micros() as i64; FlowRunResult { execution_id, @@ -189,11 +176,7 @@ async fn execute_flow( input, signal: report.signal, node_execution_results: report.node_execution_results, - runtime_usage: RuntimeUsage { - flow_id, - duration: duration_micros, - }, - } + } } fn parse_execution_id_from_subject( @@ -347,17 +330,15 @@ mod tests { let run_result = execute_flow(execution_id, flow, &engine, None, None).await; println!( - "started_at={} finished_at={} delta={} runtime_usage.duration={}", + "started_at={} finished_at={} delta={}", run_result.started_at, run_result.finished_at, run_result.finished_at - run_result.started_at, - run_result.runtime_usage.duration ); assert_eq!(run_result.execution_id, execution_id); assert!(run_result.started_at >= 1_000_000_000_000_000); assert!(run_result.finished_at >= run_result.started_at); - assert!(run_result.runtime_usage.duration > 0); assert!( run_result .node_execution_results diff --git a/crates/taurus/src/client/mod.rs b/crates/taurus/src/client/mod.rs index 8a5cec6..a953406 100644 --- a/crates/taurus/src/client/mod.rs +++ b/crates/taurus/src/client/mod.rs @@ -1,3 +1,2 @@ pub mod runtime_execution; pub mod runtime_status; -pub mod runtime_usage; diff --git a/crates/taurus/src/client/runtime_usage.rs b/crates/taurus/src/client/runtime_usage.rs deleted file mode 100644 index 58b2360..0000000 --- a/crates/taurus/src/client/runtime_usage.rs +++ /dev/null @@ -1,48 +0,0 @@ -use code0_flow::flow_service::{ - auth::get_authorization_metadata, retry::create_channel_with_retry, -}; -use tonic::{Extensions, Request, transport::Channel}; -use tucana::{ - aquila::{RuntimeUsageRequest, runtime_usage_service_client::RuntimeUsageServiceClient}, - shared::RuntimeUsage, -}; - -pub struct TaurusRuntimeUsageService { - channel: Channel, - aquila_token: String, -} - -impl TaurusRuntimeUsageService { - pub async fn from_url(aquila_url: String, aquila_token: String) -> Self { - let channel = create_channel_with_retry("Aquila", aquila_url).await; - TaurusRuntimeUsageService { - channel, - aquila_token, - } - } - - pub async fn update_runtime_usage(&self, runtime_usage: RuntimeUsage) { - log::info!("Updating the current Runtime Usage!"); - let mut client = RuntimeUsageServiceClient::new(self.channel.clone()); - - let request = Request::from_parts( - get_authorization_metadata(&self.aquila_token), - Extensions::new(), - RuntimeUsageRequest { - runtime_usage: vec![runtime_usage], - }, - ); - - match client.update(request).await { - Ok(response) => { - log::info!( - "Was the update of the RuntimeUsage accepted by Sagittarius? {}", - response.into_inner().success - ); - } - Err(err) => { - log::error!("Failed to update RuntimeUsage: {:?}", err); - } - } - } -} diff --git a/docs/dev.md b/docs/dev.md index 47e9c4f..24dfa82 100644 --- a/docs/dev.md +++ b/docs/dev.md @@ -13,7 +13,7 @@ Taurus is the execution runtime in the CodeZero execution block. - Executes flow graphs via `taurus-core::runtime::engine::ExecutionEngine` - Emits lifecycle events to NATS (`runtime.emitter.`) - Delegates remote nodes to external services over NATS (`action..`) -- Reports runtime status, runtime usage, and execution results to Aquila in dynamic mode +- Reports runtime status and execution results to Aquila in dynamic mode ## Workspace Layout @@ -50,7 +50,7 @@ graph TD Core --> Remote Remote -->|action..| NATS NATS --> Service - Taurus -->|runtime status + usage + execution result| Aquila + Taurus -->|runtime status + execution result| Aquila ``` ### Execution details @@ -73,7 +73,6 @@ Taurus mode is controlled by `MODE`. - Sends definitions to Aquila (retry loop until success) - Starts runtime status reporting (including heartbeat) -- Sends runtime usage updates after each flow run - Sends execution result updates after each flow run ### `static` @@ -83,7 +82,6 @@ Taurus mode is controlled by `MODE`. - Taurus still executes flows from NATS - No definition push - No runtime status updates -- No runtime usage updates - No execution result updates ## Environment Variables