Skip to content

chore(function-autoscaler): remove obsolete Cassandra state #1012

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
borao wants to merge 1 commit into
base: main
Choose a base branch
from
Open

chore(function-autoscaler): remove obsolete Cassandra state #1012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions migrations/cassandra/keyspaces/nvcf_autoscaler/04_drop_obsolete_function_tables.up.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
DROP TABLE IF EXISTS nvcf_autoscaler.recently_invoked_functions_history;
DROP TABLE IF EXISTS nvcf_autoscaler.running_functions_without_invocations;
DROP TABLE IF EXISTS nvcf_autoscaler.running_functions_without_invocations_history;
628 changes: 26 additions & 602 deletions ...ntrol-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 0 additions & 3 deletions ...trol-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +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)]
Expand All @@ -47,7 +46,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")]
Expand Down Expand Up @@ -78,7 +76,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(),
Expand Down
163 changes: 14 additions & 149 deletions src/control-plane-services/function-autoscaler/crates/server/src/cassandra/statements.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,6 @@
* limitations under the License.
*/

// Default TTL values — now configurable via CassandraSettings

#[derive(Debug, Clone, Copy)]
pub enum ActiveFunctionTable {
RecentlyInvokedFunctions,
RunningFunctionsWithoutInvocations,
}

// locks table
pub(crate) fn get_select_locks_stmt(keyspace: &str) -> String {
format!(
Expand Down Expand Up @@ -58,99 +50,6 @@ pub(crate) fn get_select_recently_invoked_functions_in_token_range_stmt(keyspace
)
}

pub(crate) fn get_delete_recently_invoked_function_stmt(keyspace: &str) -> String {
format!(
"DELETE FROM {}.recently_invoked_functions \
WHERE function_id = ? AND function_version_id = ?;",
keyspace
)
}

// running_functions_without_invocations Table
// account_id is a regular column, not part of PK
pub(crate) fn get_select_running_functions_without_invocations_in_token_range_stmt(
keyspace: &str,
) -> String {
format!(
"SELECT function_id, function_version_id, account_id \
FROM {}.running_functions_without_invocations \
WHERE token(function_id, function_version_id) >= ? AND token(function_id, function_version_id) <= ?;",
keyspace
)
}

pub(crate) fn get_delete_running_function_without_invocations_stmt(keyspace: &str) -> String {
format!(
"DELETE FROM {}.running_functions_without_invocations \
WHERE function_id = ? AND function_version_id = ?;",
keyspace
)
}

// 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)
}
Expand Down Expand Up @@ -197,56 +96,12 @@ pub(crate) fn get_stmt_insert_to_recently_invoked_functions(
)
}

// Inserts to the running_functions_without_invocations table with a configurable row TTL (from CassandraSettings.recently_invoked_ttl_seconds, default 1800s).
// If function discovery logic doesn't report the function as active, the row is pruned automatically after the TTL expires.
pub(crate) fn get_stmt_insert_to_running_functions_without_invocations(
keyspace: &str,
ttl_seconds: i32,
) -> String {
format!(
"INSERT INTO {}.running_functions_without_invocations (function_id, function_version_id, account_id, last_updated_at) VALUES (?, ?, ?, ?) USING TTL {}",
keyspace,
ttl_seconds
)
}

// 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;
use super::{
get_select_recently_invoked_functions_in_token_range_stmt,
get_stmt_insert_to_recently_invoked_functions, get_stmt_refresh_lock,
};

#[test]
fn refresh_lock_renews_every_non_key_column() {
Expand All @@ -255,4 +110,14 @@ mod tests {
assert!(statement.contains("SET node_id = ?, acquired_at = ?"));
assert!(statement.contains("IF node_id = ?"));
}

#[test]
fn active_function_statements_preserve_table_and_ttl() {
let select = get_select_recently_invoked_functions_in_token_range_stmt("test_keyspace");
let insert = get_stmt_insert_to_recently_invoked_functions("test_keyspace", 1800);

assert!(select.contains("FROM test_keyspace.recently_invoked_functions"));
assert!(insert.contains("INTO test_keyspace.recently_invoked_functions"));
assert!(insert.ends_with("USING TTL 1800"));
}
}
18 changes: 1 addition & 17 deletions src/control-plane-services/function-autoscaler/crates/server/src/models/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,13 @@ use chrono::{DateTime, Utc};
use scylla::DeserializeRow;
use uuid::Uuid;

#[derive(Debug, Clone, DeserializeRow)]
#[derive(Debug, Clone)]
pub struct ActiveFunctionDetails {
pub function_id: Uuid,
pub function_version_id: Uuid,
/// Optional for backwards compatibility with existing Cassandra rows that predate this column.
#[scylla(rename = "account_id")]
pub nca_id: Option<String>,
pub last_updated_at: Option<DateTime<Utc>>,
pub num_workers: Option<i32>,
pub last_predicted_desired_instance_count: Option<i32>,
pub last_predicted_error_code: Option<String>,
}

impl ActiveFunctionDetails {
Expand All @@ -40,20 +36,8 @@ 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,
}
}

/// Returns nca_id or empty string if not set (for backwards compatibility)
pub fn nca_id_or_empty(&self) -> &str {
self.nca_id.as_deref().unwrap_or("")
}

/// Alias for nca_id_or_empty for backwards compatibility
pub fn nca_id_or_nil(&self) -> String {
self.nca_id.clone().unwrap_or_default()
}
}

#[derive(Debug, Clone, DeserializeRow)]
Expand Down
46 changes: 1 addition & 45 deletions src/control-plane-services/function-autoscaler/crates/server/src/nvcf_api/nvcf_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -163,7 +160,6 @@ struct ProcessRequestCtx<'a> {
rate_limiter: &'a Arc<RateLimiter>,
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,
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
},
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -614,7 +608,6 @@ impl NvcfApiService {
.await;

// Log result and record metrics
let mut num_workers_from_api: Option<i32> = None;
match result {
Ok(response) => {
// Record autoscaling status
Expand All @@ -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,
Expand All @@ -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(
Expand All @@ -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(())
Expand Down
Loading