From a109fb5184d54e4e82c5d63ad508a711c0f0a528 Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Mon, 10 Aug 2026 21:56:05 +0000 Subject: [PATCH 1/2] chore(function-autoscaler): remove history database writes Signed-off-by: Bora Oztekin --- .../server/src/cassandra/cassandra_service.rs | 631 ++++-------------- .../src/cassandra/cassandra_settings.rs | 4 - .../crates/server/src/cassandra/statements.rs | 98 --- .../crates/server/src/models/mod.rs | 4 - .../crates/server/src/nvcf_api/nvcf_client.rs | 46 +- .../crates/server/src/work/discovery.rs | 6 - 6 files changed, 137 insertions(+), 652 deletions(-) diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs index cc7fb3f54..80f0856a4 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs @@ -37,7 +37,7 @@ use scylla::statement::{Consistency, SerialConsistency, Statement}; use std::num::NonZero; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tracing; +use tracing::{self, Instrument}; use uuid::Uuid; use super::cassandra_settings::CassandraSettings; @@ -45,21 +45,55 @@ use super::statements::*; use crate::secrets::secrets_file_watcher::SecretFileWatcher; pub const CASSANDRA_TOKEN_RANGE: [i64; 2] = [i64::MIN, i64::MAX]; -/// Execute any async operation with automatic timing and tracing -async fn with_cassandra_timing(operation_name: &str, operation: F) -> T +/// Execute a Cassandra operation in a structured span and record its outcome. +async fn with_cassandra_timing( + operation_name: &'static str, + operation: F, +) -> std::result::Result where F: FnOnce() -> Fut, - Fut: std::future::Future, + Fut: std::future::Future>, + E: std::fmt::Display, { - let start = std::time::Instant::now(); - let result = operation().await; - let duration = start.elapsed(); - - tracing::trace!( + let span = tracing::debug_span!( + "cassandra.operation", cassandra.operation = operation_name, - cassandra.duration_ms = duration.as_millis(), - "Cassandra operation completed" + cassandra.duration_ms = tracing::field::Empty, + cassandra.status = tracing::field::Empty, + error = tracing::field::Empty, + otel.status_code = tracing::field::Empty, ); + let start = std::time::Instant::now(); + let result = operation().instrument(span.clone()).await; + let duration_ms = start.elapsed().as_millis() as u64; + + span.record("cassandra.duration_ms", duration_ms); + match &result { + Ok(_) => { + span.record("cassandra.status", "ok"); + span.record("otel.status_code", "OK"); + tracing::trace!( + parent: &span, + cassandra.operation = operation_name, + cassandra.duration_ms = duration_ms, + cassandra.status = "ok", + "Cassandra operation completed" + ); + } + Err(error) => { + span.record("cassandra.status", "error"); + span.record("error", tracing::field::display(error)); + span.record("otel.status_code", "ERROR"); + tracing::warn!( + parent: &span, + cassandra.operation = operation_name, + cassandra.duration_ms = duration_ms, + cassandra.status = "error", + error = %error, + "Cassandra operation failed" + ); + } + } result } @@ -378,21 +412,24 @@ impl CassandraServiceManager { } }; - let mut prepared_statement = session.prepare(stmt).await?; - prepared_statement.set_is_idempotent(true); let nca_id = function.nca_id_or_nil(); - session - .execute_unpaged( - &prepared_statement, - ( - &function.function_id, - &function.function_version_id, - &nca_id, - &function.last_updated_at, - ), - ) - .await?; - Ok(()) + with_cassandra_timing("insert_to_active_functions", || async { + let mut prepared_statement = session.prepare(stmt).await?; + prepared_statement.set_is_idempotent(true); + session + .execute_unpaged( + &prepared_statement, + ( + &function.function_id, + &function.function_version_id, + &nca_id, + &function.last_updated_at, + ), + ) + .await?; + Ok(()) + }) + .await } // Not instrumented: return value is Vec and would be captured in the span (large debug output). @@ -448,107 +485,6 @@ impl CassandraServiceManager { .await } - #[tracing::instrument(skip(self))] - pub async fn get_active_function_history_by_id( - &self, - function_id: &Uuid, - function_version_id: &Uuid, - table: ActiveFunctionTable, - ) -> Result> { - let session = self.get_session().await?; - let stmt = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_select_recently_invoked_function_history_by_id_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_select_running_function_without_invocations_history_by_id_stmt( - &self.config.keyspace, - ) - } - }; - with_cassandra_timing("get_active_function_history_by_id", || async { - let mut prepared_statement = session.prepare(stmt).await?; - prepared_statement.set_tracing(true); - let mut iter = session - .execute_iter(prepared_statement, (function_id, function_version_id)) - .await? - .rows_stream::()?; - Ok(iter.try_next().await?) - }) - .await - } - - #[tracing::instrument(skip(self))] - pub async fn add_new_active_function( - &self, - function: &ActiveFunctionDetails, - table: ActiveFunctionTable, - ) -> Result<()> { - let session = self.get_session().await?; - let stmt_active_function = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_stmt_insert_to_recently_invoked_functions( - &self.config.keyspace, - self.config.recently_invoked_ttl_seconds, - ) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_stmt_insert_to_running_functions_without_invocations( - &self.config.keyspace, - self.config.recently_invoked_ttl_seconds, - ) - } - }; - let stmt_active_function_history = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_insert_recently_invoked_functions_history_pk_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_insert_running_functions_without_invocations_history_pk_stmt( - &self.config.keyspace, - ) - } - }; - let mut batch = scylla::statement::batch::Batch::default(); - batch.set_consistency(scylla::statement::Consistency::Quorum); - batch.append_statement(Statement::new(stmt_active_function)); - batch.append_statement(Statement::new(stmt_active_function_history)); - let nca_id = function.nca_id_or_nil(); - let values = ( - ( - &function.function_id, - &function.function_version_id, - &nca_id, - function.last_updated_at, - ), - ( - &function.function_id, - &function.function_version_id, - &nca_id, - &function.num_workers.unwrap_or(-1), - ), - ); - match session.batch(&batch, values).await { - Ok(_) => { - tracing::debug!( - "Successfully inserted function {}:{} to Cassandra", - function.function_id, - function.function_version_id - ); - } - Err(e) => { - tracing::error!( - "Failed to insert function {}:{} to Cassandra: {}", - function.function_id, - function.function_version_id, - e - ); - return Err(e.into()); - } - } - Ok(()) - } - /// Upserts a function into recently_invoked_functions with a fresh TTL. /// Called by the scaling loop when desired_instance_count > 0 to keep the /// function alive in the active set without touching the history table. @@ -603,49 +539,36 @@ impl CassandraServiceManager { ) } }; - let stmt_active_function_history = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_insert_recently_invoked_functions_history_pk_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_insert_running_functions_without_invocations_history_pk_stmt( - &self.config.keyspace, - ) - } - }; - - let prepared_active_function = session.prepare(stmt_active_function).await?; - let prepared_active_function_history = - session.prepare(stmt_active_function_history).await?; - - execute_chunked(functions, 200, |function| { - let session = session.clone(); - let prepared_active_function = prepared_active_function.clone(); - let prepared_active_function_history = prepared_active_function_history.clone(); - let function_id = function.function_id; - let function_version_id = function.function_version_id; - let nca_id = function.nca_id_or_nil(); - let last_updated_at = function.last_updated_at; - let num_workers = function.num_workers.unwrap_or(-1); - async move { - let mut batch = scylla::statement::batch::Batch::default(); - batch.set_consistency(scylla::statement::Consistency::Quorum); - batch.append_statement(prepared_active_function); - batch.append_statement(prepared_active_function_history); - let values = ( - (&function_id, &function_version_id, &nca_id, last_updated_at), - (&function_id, &function_version_id, &nca_id, &num_workers), - ); - session.batch(&batch, values).await.map_err(|e| { - tracing::error!( - "Failed to insert function {}:{} to Cassandra: {}", - function_id, - function_version_id, - e - ); - anyhow::Error::from(e) - }) - } + let mut prepared_active_function = session.prepare(stmt_active_function).await?; + prepared_active_function.set_consistency(scylla::statement::Consistency::Quorum); + + with_cassandra_timing("add_new_active_functions_batch", || async { + execute_chunked(functions, 200, |function| { + let session = session.clone(); + let prepared_active_function = prepared_active_function.clone(); + let function_id = function.function_id; + let function_version_id = function.function_version_id; + let nca_id = function.nca_id_or_nil(); + let last_updated_at = function.last_updated_at; + async move { + session + .execute_unpaged( + &prepared_active_function, + (&function_id, &function_version_id, &nca_id, last_updated_at), + ) + .await + .map_err(|e| { + tracing::error!( + "Failed to insert function {}:{} to Cassandra: {}", + function_id, + function_version_id, + e + ); + anyhow::Error::from(e) + }) + } + }) + .await }) .await?; @@ -668,26 +591,15 @@ impl CassandraServiceManager { get_delete_running_function_without_invocations_stmt(&self.config.keyspace) } }; - let stmt_active_function_history = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_delete_recently_invoked_function_history_pk_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_delete_running_function_without_invocations_history_pk_stmt( - &self.config.keyspace, - ) - } - }; - let mut batch = scylla::statement::batch::Batch::default(); - batch.set_consistency(scylla::statement::Consistency::Quorum); - - batch.append_statement(Statement::new(stmt_active_function)); - batch.append_statement(Statement::new(stmt_active_function_history)); - let values: ((&Uuid, &Uuid), (&Uuid, &Uuid)) = ( - (function_id, function_version_id), - (function_id, function_version_id), - ); - match session.batch(&batch, values).await { + let mut prepared = session.prepare(stmt_active_function).await?; + prepared.set_consistency(scylla::statement::Consistency::Quorum); + let result = with_cassandra_timing("delete_active_function", || async { + session + .execute_unpaged(&prepared, (function_id, function_version_id)) + .await + }) + .await; + match result { Ok(_) => { tracing::debug!( "Successfully deleted function {}:{} from Cassandra", @@ -708,171 +620,6 @@ impl CassandraServiceManager { Ok(()) } - #[tracing::instrument(skip(self, functions), fields(functions_len = functions.len()))] - pub async fn transition_functions_between_tables_batch( - &self, - functions: &[ActiveFunctionDetails], - from_table: ActiveFunctionTable, - to_table: ActiveFunctionTable, - ) -> Result<()> { - if functions.is_empty() { - return Ok(()); - } - - let session = self.get_session().await?; - - // Prepare delete statements for source table - let stmt_delete_active = match from_table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_delete_recently_invoked_function_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_delete_running_function_without_invocations_stmt(&self.config.keyspace) - } - }; - let stmt_delete_history = match from_table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_delete_recently_invoked_function_history_pk_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_delete_running_function_without_invocations_history_pk_stmt( - &self.config.keyspace, - ) - } - }; - - // Prepare insert statements for destination table - let stmt_insert_active = match to_table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_stmt_insert_to_recently_invoked_functions( - &self.config.keyspace, - self.config.recently_invoked_ttl_seconds, - ) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_stmt_insert_to_running_functions_without_invocations( - &self.config.keyspace, - self.config.recently_invoked_ttl_seconds, - ) - } - }; - let stmt_insert_history = match to_table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_insert_recently_invoked_functions_history_pk_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_insert_running_functions_without_invocations_history_pk_stmt( - &self.config.keyspace, - ) - } - }; - - let prepared_delete_active = session.prepare(stmt_delete_active).await?; - let prepared_delete_history = session.prepare(stmt_delete_history).await?; - let prepared_insert_active = session.prepare(stmt_insert_active).await?; - let prepared_insert_history = session.prepare(stmt_insert_history).await?; - - execute_chunked(functions, 200, |function| { - let session = session.clone(); - let prepared_delete_active = prepared_delete_active.clone(); - let prepared_delete_history = prepared_delete_history.clone(); - let prepared_insert_active = prepared_insert_active.clone(); - let prepared_insert_history = prepared_insert_history.clone(); - let function_id = function.function_id; - let function_version_id = function.function_version_id; - let nca_id = function.nca_id_or_nil(); - let last_updated_at = function.last_updated_at; - let num_workers = function.num_workers.unwrap_or(-1); - async move { - let mut batch = scylla::statement::batch::Batch::default(); - batch.set_consistency(scylla::statement::Consistency::Quorum); - batch.append_statement(prepared_delete_active); - batch.append_statement(prepared_delete_history); - batch.append_statement(prepared_insert_active); - batch.append_statement(prepared_insert_history); - let values = ( - (&function_id, &function_version_id), - (&function_id, &function_version_id), - (&function_id, &function_version_id, &nca_id, last_updated_at), - (&function_id, &function_version_id, &nca_id, &num_workers), - ); - session.batch(&batch, values).await.map_err(|e| { - tracing::error!( - "Failed to transition function {}:{}: {}", - function_id, - function_version_id, - e - ); - anyhow::Error::from(e) - }) - } - }) - .await?; - - Ok(()) - } - - #[tracing::instrument(skip(self))] - pub async fn insert_to_active_function_history_prediction_row( - &self, - function: &ActiveFunctionDetails, - table: ActiveFunctionTable, - ) -> Result<()> { - let session = self.get_session().await?; - let stmt = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_stmt_str_insert_to_recently_invoked_functions_history_prediction_row( - &self.config.keyspace, - self.config.history_prediction_ttl_seconds, - ) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_stmt_str_insert_to_running_functions_without_invocations_history_prediction_row( - &self.config.keyspace, - self.config.history_prediction_ttl_seconds, - ) - } - }; - let error_code = function.last_predicted_error_code.clone(); - let nca_id = function.nca_id_or_nil(); - - let mut prepared_statement = session.prepare(stmt).await?; - prepared_statement.set_is_idempotent(true); - match session - .execute_unpaged( - &prepared_statement, - ( - &function.function_id, - &function.function_version_id, - &nca_id, - &function.num_workers, - &function.last_predicted_desired_instance_count.unwrap_or(0), - &error_code, - &function.last_updated_at, - ), - ) - .await - { - Ok(_) => { - tracing::debug!( - "Successfully inserted prediction row for function {}:{} to Cassandra", - function.function_id, - function.function_version_id - ); - } - Err(e) => { - tracing::error!( - "Failed to insert prediction row for function {}:{} to Cassandra: {}", - function.function_id, - function.function_version_id, - e - ); - return Err(e.into()); - } - } - Ok(()) - } - // Returns true if the lock was acquired, false if it was already held by another node #[tracing::instrument(skip(self))] pub async fn put_lock(&self, lock: &DistributedLock, ttl_seconds: i32) -> Result { @@ -987,10 +734,13 @@ impl CassandraServiceManager { .prepare(get_delete_locks_stmt(&self.config.keyspace)) .await?; prepared_statement.set_is_idempotent(true); - match session - .execute_unpaged(&prepared_statement, (lock_name,)) - .await - { + let result = with_cassandra_timing("delete_lock", || async { + session + .execute_unpaged(&prepared_statement, (lock_name,)) + .await + }) + .await; + match result { Ok(_) => { tracing::debug!("Successfully deleted lock {} from Cassandra", lock_name); } @@ -1012,10 +762,13 @@ impl CassandraServiceManager { )) .await?; prepared_statement.set_is_idempotent(true); - match session - .execute_unpaged(&prepared_statement, (&node.node_id, &node.last_updated_at)) - .await - { + let result = with_cassandra_timing("insert_to_nodes", || async { + session + .execute_unpaged(&prepared_statement, (&node.node_id, &node.last_updated_at)) + .await + }) + .await; + match result { Ok(_) => { tracing::debug!("Successfully inserted node {} to Cassandra", node.node_id); } @@ -1052,10 +805,13 @@ impl CassandraServiceManager { .prepare(get_delete_node_stmt(&self.config.keyspace)) .await?; prepared_statement.set_is_idempotent(true); - match session - .execute_unpaged(&prepared_statement, (node_id,)) - .await - { + let result = with_cassandra_timing("delete_node", || async { + session + .execute_unpaged(&prepared_statement, (node_id,)) + .await + }) + .await; + match result { Ok(_) => { tracing::debug!("Successfully deleted node {} from Cassandra", node_id); } @@ -1219,7 +975,6 @@ mod tests { pool: PoolSettings { local_size: 1 }, execution_profile: ExecutionProfileSettings::default(), is_development: true, - history_prediction_ttl_seconds: 300, ..Default::default() } } @@ -1230,8 +985,6 @@ mod tests { function_version_id: Uuid::new_v4(), nca_id: Some("test-nca-id".to_string()), num_workers: Some(1), - last_predicted_desired_instance_count: Some(1), - last_predicted_error_code: None, last_updated_at: Some(Utc::now()), } } @@ -1279,55 +1032,21 @@ mod tests { .await .unwrap(); - // Test both table types for table_type in [ ActiveFunctionTable::RecentlyInvokedFunctions, ActiveFunctionTable::RunningFunctionsWithoutInvocations, ] { let function = create_test_active_function_details(); - // Test insert operation - let insert_result = manager.add_new_active_function(&function, table_type).await; - assert!( - insert_result.is_ok(), - "Insert failed for {:?}: {:?}", - table_type, - insert_result.err() - ); - - // Test get operation - let get_result = manager - .get_active_function_history_by_id( - &function.function_id, - &function.function_version_id, - table_type, - ) - .await; - assert!( - get_result.is_ok(), - "Get failed for {:?}: {:?}", - table_type, - get_result.err() - ); - let result = get_result.unwrap(); - assert!(result.is_some()); - let function_details = result.unwrap(); - assert_eq!(function_details.function_id, function.function_id); - assert_eq!( - function_details.function_version_id, - function.function_version_id - ); - assert_eq!(function_details.num_workers, Some(1)); - assert_eq!(function_details.last_predicted_desired_instance_count, None); - assert_eq!(function_details.last_predicted_error_code, None); + manager + .add_new_active_functions_batch(std::slice::from_ref(&function), table_type) + .await + .unwrap(); - // Test get operation with token range let token_range = CASSANDRA_TOKEN_RANGE; - let get_result = manager + let functions = manager .get_active_functions_with_token_range(&token_range, 100, table_type) - .await; - assert!(get_result.is_ok()); - let functions = get_result.unwrap(); - assert!(!functions.is_empty()); + .await + .unwrap(); assert_eq!(functions.len(), 1); assert_eq!(functions[0].function_id, function.function_id); assert_eq!( @@ -1335,97 +1054,19 @@ mod tests { function.function_version_id ); - // Test insert details with some fields - let insert_result = manager - .insert_to_active_function_history_prediction_row(&function, table_type) - .await; - assert!(insert_result.is_ok()); - - // Test get operation - let get_result = manager - .get_active_function_history_by_id( - &function_details.function_id, - &function_details.function_version_id, - table_type, - ) - .await; - assert!(get_result.is_ok()); - let result = get_result.unwrap(); - assert!(result.is_some()); - let mut modified_function_details = result.unwrap(); - assert_eq!(function.function_id, modified_function_details.function_id); - assert_eq!( - function.function_version_id, - modified_function_details.function_version_id - ); - let expected_num_workers = function.num_workers; - assert_eq!(function.num_workers, modified_function_details.num_workers); - assert_eq!(modified_function_details.num_workers, expected_num_workers); - // Expect the value from the fixture (Some(1) for first insert) - assert_eq!( - modified_function_details.last_predicted_desired_instance_count, - Some(1) - ); - assert_eq!(modified_function_details.last_predicted_error_code, None); - - // Insert again with some fields modified - modified_function_details.last_predicted_desired_instance_count = Some(10); - modified_function_details.last_predicted_error_code = None; - let insert_result = manager - .insert_to_active_function_history_prediction_row( - &modified_function_details, - table_type, - ) - .await; - assert!(insert_result.is_ok()); - - // Test get operation again - let get_result = manager - .get_active_function_history_by_id( - &function_details.function_id, - &function_details.function_version_id, - table_type, - ) - .await; - assert!(get_result.is_ok()); - let result = get_result.unwrap(); - assert!(result.is_some()); - let function_details = result.unwrap(); - assert_eq!( - function_details.function_id, - modified_function_details.function_id - ); - assert_eq!( - function_details.function_version_id, - modified_function_details.function_version_id - ); - assert_eq!( - function_details.num_workers, - modified_function_details.num_workers - ); - assert_eq!(function_details.num_workers, expected_num_workers); - // Expect the last value written (Some(10) for second insert) - assert_eq!( - function_details.last_predicted_desired_instance_count, - Some(10) - ); - assert_eq!(function_details.last_predicted_error_code, None); - - // Test delete operation - let delete_result = manager + manager .delete_active_function( &function.function_id, &function.function_version_id, table_type, ) - .await; - assert!(delete_result.is_ok()); - // Test get operation again- it should be empty - let get_result = manager + .await + .unwrap(); + let functions = manager .get_active_functions_with_token_range(&token_range, 100, table_type) - .await; - assert!(get_result.is_ok()); - assert!(get_result.unwrap().is_empty()); + .await + .unwrap(); + assert!(functions.is_empty()); } } diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs index 2819d5f46..78596e86e 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs @@ -29,8 +29,6 @@ const DEFAULT_SERIAL_CONSISTENCY: &str = "LOCAL_SERIAL"; const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_millis(10000); // 10 seconds const DEFAULT_MAX_RETRY_COUNT: u32 = 3; const DEFAULT_RETRY_INTERVAL: Duration = Duration::from_millis(1000); // 1 second -const DEFAULT_HISTORY_PREDICTION_TTL_SECONDS: i32 = 300; - #[serde_as] #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(default)] @@ -47,7 +45,6 @@ pub struct CassandraSettings { pub execution_profile: ExecutionProfileSettings, #[serde(default)] pub is_development: bool, - pub history_prediction_ttl_seconds: i32, #[serde(default = "default_node_health_ttl")] pub node_health_ttl_seconds: i32, #[serde(default = "default_recently_invoked_ttl")] @@ -78,7 +75,6 @@ impl Default for CassandraSettings { pool: PoolSettings::default(), execution_profile: ExecutionProfileSettings::default(), is_development: true, - history_prediction_ttl_seconds: DEFAULT_HISTORY_PREDICTION_TTL_SECONDS, node_health_ttl_seconds: default_node_health_ttl(), recently_invoked_ttl_seconds: default_recently_invoked_ttl(), health_check_cache_ttl_seconds: default_health_check_cache_ttl(), diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/statements.rs b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/statements.rs index 0cf468b7a..6e97d26eb 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/statements.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/statements.rs @@ -87,70 +87,6 @@ pub(crate) fn get_delete_running_function_without_invocations_stmt(keyspace: &st ) } -// recently_invoked_functions_history Table -// account_id is a regular column, not part of PK -pub(crate) fn get_select_recently_invoked_function_history_by_id_stmt(keyspace: &str) -> String { - format!( - "SELECT function_id, function_version_id, account_id, num_workers, \ - last_predicted_desired_instance_count, \ - last_predicted_error_code, last_updated_at \ - FROM {}.recently_invoked_functions_history \ - WHERE function_id = ? AND function_version_id = ? LIMIT 1;", - keyspace - ) -} - -pub(crate) fn get_delete_recently_invoked_function_history_pk_stmt(keyspace: &str) -> String { - format!( - "DELETE FROM {}.recently_invoked_functions_history \ - WHERE function_id = ? AND function_version_id = ?;", - keyspace - ) -} - -pub(crate) fn get_insert_recently_invoked_functions_history_pk_stmt(keyspace: &str) -> String { - format!( - "INSERT INTO {}.recently_invoked_functions_history (function_id, function_version_id, account_id, num_workers) \ - VALUES (?, ?, ?, ?)", - keyspace - ) -} - -// running_functions_without_invocations_history Table -// account_id is a regular column, not part of PK -pub(crate) fn get_select_running_function_without_invocations_history_by_id_stmt( - keyspace: &str, -) -> String { - format!( - "SELECT function_id, function_version_id, account_id, num_workers, \ - last_predicted_desired_instance_count, \ - last_predicted_error_code, last_updated_at \ - FROM {}.running_functions_without_invocations_history \ - WHERE function_id = ? AND function_version_id = ? LIMIT 1;", - keyspace - ) -} - -pub(crate) fn get_delete_running_function_without_invocations_history_pk_stmt( - keyspace: &str, -) -> String { - format!( - "DELETE FROM {}.running_functions_without_invocations_history \ - WHERE function_id = ? AND function_version_id = ?;", - keyspace - ) -} - -pub(crate) fn get_insert_running_functions_without_invocations_history_pk_stmt( - keyspace: &str, -) -> String { - format!( - "INSERT INTO {}.running_functions_without_invocations_history (function_id, function_version_id, account_id, num_workers) \ - VALUES (?, ?, ?, ?)", - keyspace - ) -} - pub(crate) fn get_health_check_query_stmt(keyspace: &str) -> String { format!("SELECT now() from {}.healthy_nodes LIMIT 1;", keyspace) } @@ -210,40 +146,6 @@ pub(crate) fn get_stmt_insert_to_running_functions_without_invocations( ) } -// Inserts to the recently_invoked_functions_history table must be done with a row TTL of 180 seconds. -// If function discovery logic doesn't report the function as active, the row is pruned automatically after 180 seconds. -// The table itself has no default TTL and is kept for historical context if needed. -// We want to add a configurable job to prune the table periodically for inactive rows. -pub(crate) fn get_stmt_str_insert_to_recently_invoked_functions_history_prediction_row( - keyspace: &str, - ttl_seconds: i32, -) -> String { - format!( - "INSERT INTO {}.recently_invoked_functions_history (function_id, function_version_id, account_id, \ - num_workers, last_predicted_desired_instance_count, \ - last_predicted_error_code, last_updated_at) \ - VALUES (?, ?, ?, ?, ?, ?, ?) USING TTL {}", - keyspace, ttl_seconds - ) -} - -// Inserts to the running_functions_without_invocations_history table must be done with a row TTL of 300 seconds. -// If function discovery logic doesn't report the function as active, the row is pruned automatically after 300 seconds. -// The table itself has no default TTL and is kept for historical context if needed. -// We want to add a configurable job to prune the table periodically for inactive rows. -pub(crate) fn get_stmt_str_insert_to_running_functions_without_invocations_history_prediction_row( - keyspace: &str, - ttl_seconds: i32, -) -> String { - format!( - "INSERT INTO {}.running_functions_without_invocations_history (function_id, function_version_id, account_id, \ - num_workers, last_predicted_desired_instance_count, \ - last_predicted_error_code, last_updated_at) \ - VALUES (?, ?, ?, ?, ?, ?, ?) USING TTL {}", - keyspace, ttl_seconds - ) -} - #[cfg(test)] mod tests { use super::get_stmt_refresh_lock; diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/models/mod.rs b/src/control-plane-services/function-autoscaler/crates/server/src/models/mod.rs index 324c8120c..910329f2d 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/models/mod.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/models/mod.rs @@ -28,8 +28,6 @@ pub struct ActiveFunctionDetails { pub nca_id: Option, pub last_updated_at: Option>, pub num_workers: Option, - pub last_predicted_desired_instance_count: Option, - pub last_predicted_error_code: Option, } impl ActiveFunctionDetails { @@ -40,8 +38,6 @@ impl ActiveFunctionDetails { nca_id: Some(nca_id), last_updated_at: Some(Utc::now()), num_workers: None, - last_predicted_desired_instance_count: None, - last_predicted_error_code: None, } } diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/nvcf_api/nvcf_client.rs b/src/control-plane-services/function-autoscaler/crates/server/src/nvcf_api/nvcf_client.rs index c2dadac5b..aacdec85b 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/nvcf_api/nvcf_client.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/nvcf_api/nvcf_client.rs @@ -17,15 +17,12 @@ use crate::cassandra::cassandra_service::CassandraServiceManager; use crate::cassandra::distributed_lock::DistributedLockManager; -use crate::cassandra::statements::ActiveFunctionTable; use crate::metrics; -use crate::models::ActiveFunctionDetails; use crate::nvcf_api::oauth2_client; use crate::nvcf_api::{AutoscalerResponse, DeploymentInfo, FunctionStatus, NvcfApiError}; use crate::secrets::secrets_file_watcher::SecretFileWatcher; use crate::work::bucket::{NodeBucketManager, BUCKET_COUNT}; use crate::work::{FunctionCachedState, FunctionStateCache}; -use chrono::Utc; use leaky_bucket::RateLimiter; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -163,7 +160,6 @@ struct ProcessRequestCtx<'a> { rate_limiter: &'a Arc, nvcf_api_channel: &'a Channel, oauth2_client: Option<&'a oauth2_client::OAuth2Client>, - cassandra_service: Option<&'a CassandraServiceManager>, function_state_cache: Option<&'a FunctionStateCache>, dry_run: bool, } @@ -316,7 +312,7 @@ impl NvcfApiService { return; } - if let Some(cassandra_service) = &cassandra_service { + if cassandra_service.is_some() { let lock_name = format!("{}_{}", NVCF_API_BUCKET_LOCK_PREFIX, bucket_index); match lock_manager.try_acquire( lock_name, @@ -346,7 +342,6 @@ impl NvcfApiService { rate_limiter: &rate_limiter, nvcf_api_channel: &nvcf_api_channel, oauth2_client: oauth2_client.as_ref(), - cassandra_service: Some(cassandra_service.as_ref()), function_state_cache: function_state_cache.as_deref(), dry_run, }, @@ -577,7 +572,6 @@ impl NvcfApiService { let rate_limiter = ctx.rate_limiter; let nvcf_api_channel = ctx.nvcf_api_channel; let oauth2_client = ctx.oauth2_client; - let cassandra_service = ctx.cassandra_service; let function_state_cache = ctx.function_state_cache; let dry_run = ctx.dry_run; // Check if request is stale (older than 15 seconds) @@ -614,7 +608,6 @@ impl NvcfApiService { .await; // Log result and record metrics - let mut num_workers_from_api: Option = None; match result { Ok(response) => { // Record autoscaling status @@ -624,9 +617,6 @@ impl NvcfApiService { 0_f64, ); - // Capture active_instances from API response for feedback loop - num_workers_from_api = Some(response.active_instances); - tracing::debug!( "Successfully processed scaling request - Active: {}, Pending: {}, Allocating: {}, Terminating: {}, Status: {}", response.active_instances, @@ -649,16 +639,6 @@ impl NvcfApiService { } } - let active_function_details = ActiveFunctionDetails { - function_id: info.function_id, - function_version_id: info.function_version_id, - nca_id: Some(info.nca_id), - last_updated_at: Some(Utc::now()), - num_workers: num_workers_from_api, - last_predicted_desired_instance_count: Some(info.required_number_of_instances), - last_predicted_error_code: error_code.clone(), - }; - // Update in-memory cache with the latest prediction result if let Some(cache) = function_state_cache { cache.insert( @@ -671,30 +651,6 @@ impl NvcfApiService { }, ); } - - // Handle Cassandra operations if available - if let Some(cassandra_service) = &cassandra_service { - let table = if info.recently_invoked { - ActiveFunctionTable::RecentlyInvokedFunctions - } else { - ActiveFunctionTable::RunningFunctionsWithoutInvocations - }; - - if let Err(cassandra_error) = cassandra_service - .insert_to_active_function_history_prediction_row( - &active_function_details, - table, - ) - .await - { - tracing::error!( - "Failed to report error to Cassandra for function {} version {}: {}", - info.function_id, - info.function_version_id, - cassandra_error, - ); - } - } } Ok(()) diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/work/discovery.rs b/src/control-plane-services/function-autoscaler/crates/server/src/work/discovery.rs index 4a034855b..d6166349f 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/work/discovery.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/work/discovery.rs @@ -628,8 +628,6 @@ async fn get_recently_invoked_functions_with_semaphore( nca_id: Some(nca_id.clone()), last_updated_at: Some(end_time), num_workers: None, // Recently invoked functions start with unknown worker count - last_predicted_desired_instance_count: None, - last_predicted_error_code: None, }; tracing::debug!( @@ -778,8 +776,6 @@ avg by(function_id, function_version_id, nca_id) (nvcf_function_instances_curren nca_id: Some(nca_id), last_updated_at: Some(end_time), num_workers, - last_predicted_desired_instance_count: None, - last_predicted_error_code: None, }; let existing = by_key.get(&key).and_then(|d| d.num_workers); @@ -891,8 +887,6 @@ pub async fn get_functions_with_active_instances( nca_id: Some(nca_id.clone()), last_updated_at: Some(end_time), num_workers: Some(-1), // BYOC functions have num_workers = -1 - last_predicted_desired_instance_count: None, - last_predicted_error_code: None, }; tracing::debug!( From 1eb76fef37e7bdba9d7ea099db9288664967f753 Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Tue, 18 Aug 2026 11:56:59 -0700 Subject: [PATCH 2/2] fix(function-autoscaler): address Cassandra review feedback Signed-off-by: Bora Oztekin --- .../server/src/cassandra/cassandra_service.rs | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs index 80f0856a4..08ee7f633 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs @@ -84,14 +84,6 @@ where span.record("cassandra.status", "error"); span.record("error", tracing::field::display(error)); span.record("otel.status_code", "ERROR"); - tracing::warn!( - parent: &span, - cassandra.operation = operation_name, - cassandra.duration_ms = duration_ms, - cassandra.status = "error", - error = %error, - "Cassandra operation failed" - ); } } @@ -541,6 +533,7 @@ impl CassandraServiceManager { }; let mut prepared_active_function = session.prepare(stmt_active_function).await?; prepared_active_function.set_consistency(scylla::statement::Consistency::Quorum); + prepared_active_function.set_is_idempotent(true); with_cassandra_timing("add_new_active_functions_batch", || async { execute_chunked(functions, 200, |function| { @@ -593,6 +586,7 @@ impl CassandraServiceManager { }; let mut prepared = session.prepare(stmt_active_function).await?; prepared.set_consistency(scylla::statement::Consistency::Quorum); + prepared.set_is_idempotent(true); let result = with_cassandra_timing("delete_active_function", || async { session .execute_unpaged(&prepared, (function_id, function_version_id)) @@ -1047,12 +1041,10 @@ mod tests { .get_active_functions_with_token_range(&token_range, 100, table_type) .await .unwrap(); - assert_eq!(functions.len(), 1); - assert_eq!(functions[0].function_id, function.function_id); - assert_eq!( - functions[0].function_version_id, - function.function_version_id - ); + assert!(functions.iter().any(|active| { + active.function_id == function.function_id + && active.function_version_id == function.function_version_id + })); manager .delete_active_function( @@ -1066,7 +1058,10 @@ mod tests { .get_active_functions_with_token_range(&token_range, 100, table_type) .await .unwrap(); - assert!(functions.is_empty()); + assert!(!functions.iter().any(|active| { + active.function_id == function.function_id + && active.function_version_id == function.function_version_id + })); } }