diff --git a/migrations/cassandra/keyspaces/nvcf_autoscaler/04_drop_obsolete_function_tables.up.sql b/migrations/cassandra/keyspaces/nvcf_autoscaler/04_drop_obsolete_function_tables.up.sql new file mode 100644 index 000000000..a331207bf --- /dev/null +++ b/migrations/cassandra/keyspaces/nvcf_autoscaler/04_drop_obsolete_function_tables.up.sql @@ -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; 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 08ee7f633..9f8d94cf8 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 @@ -20,7 +20,7 @@ use crate::models::{ ActiveFunction, ActiveFunctionDetails, DistributedLock, DistributedLockResult, NodeHealth, }; use crate::secrets::secrets_config::CassandraSslCertificates; -use anyhow::Result; +use anyhow::{Context, Result}; use async_trait::async_trait; use base64::Engine; use chrono::Utc; @@ -381,75 +381,21 @@ impl CassandraServiceManager { } } - #[tracing::instrument(skip(self, function), fields(function_id = %function.function_id, function_version_id = %function.function_version_id))] - pub async fn insert_to_active_functions( - &self, - function: &ActiveFunctionDetails, - table: ActiveFunctionTable, - ) -> Result<()> { - let session = self.get_session().await?; - - let stmt = 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 nca_id = function.nca_id_or_nil(); - 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). pub async fn get_active_functions_with_token_range( &self, token_range: &[i64], page_size: i32, - table: ActiveFunctionTable, ) -> Result> { let session = self.get_session().await?; - let stmt = match table { - ActiveFunctionTable::RecentlyInvokedFunctions => { - get_select_recently_invoked_functions_in_token_range_stmt(&self.config.keyspace) - } - ActiveFunctionTable::RunningFunctionsWithoutInvocations => { - get_select_running_functions_without_invocations_in_token_range_stmt( - &self.config.keyspace, - ) - } - }; + let stmt = get_select_recently_invoked_functions_in_token_range_stmt(&self.config.keyspace); with_cassandra_timing("get_active_functions_with_token_range", || async { let mut prepared_statement = session.prepare(stmt).await?; prepared_statement.set_page_size(page_size); // Use LOCAL_QUORUM to match the QUORUM write consistency used in add_new_active_functions_batch. // LOCAL_ONE (execution profile default) can read from a replica that hasn't received a - // recent QUORUM write, causing a second pod to see a function as new and overwrite the - // history row with num_workers=-1 even after a prior pod already wrote it. + // recent write, causing a second pod to see the function as new. prepared_statement.set_consistency(Consistency::LocalQuorum); let mut results = Vec::new(); let token_range_min = token_range[0]; @@ -479,7 +425,7 @@ impl CassandraServiceManager { /// 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. + /// function alive in the active set. #[tracing::instrument(skip(self))] pub async fn refresh_active_function_ttl(&self, function: &ActiveFunction) -> Result<()> { let session = self.get_session().await?; @@ -510,54 +456,40 @@ impl CassandraServiceManager { pub async fn add_new_active_functions_batch( &self, functions: &[ActiveFunctionDetails], - table: ActiveFunctionTable, ) -> Result<()> { if functions.is_empty() { return Ok(()); } 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 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); + let stmt = get_stmt_insert_to_recently_invoked_functions( + &self.config.keyspace, + self.config.recently_invoked_ttl_seconds, + ); + let mut prepared = session.prepare(stmt).await?; + prepared.set_consistency(Consistency::Quorum); + prepared.set_is_idempotent(true); 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 prepared = prepared.clone(); let function_id = function.function_id; let function_version_id = function.function_version_id; - let nca_id = function.nca_id_or_nil(); + let nca_id = function.nca_id.clone().unwrap_or_default(); let last_updated_at = function.last_updated_at; async move { session .execute_unpaged( - &prepared_active_function, + &prepared, (&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) + .with_context(|| { + format!( + "inserting function {}:{} into recently_invoked_functions", + function_id, function_version_id + ) }) } }) @@ -568,52 +500,6 @@ impl CassandraServiceManager { Ok(()) } - #[tracing::instrument(skip(self))] - pub async fn delete_active_function( - &self, - function_id: &Uuid, - function_version_id: &Uuid, - table: ActiveFunctionTable, - ) -> Result<()> { - let session = self.get_session().await?; - let stmt_active_function = match 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 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)) - .await - }) - .await; - match result { - Ok(_) => { - tracing::debug!( - "Successfully deleted function {}:{} from Cassandra", - function_id, - function_version_id - ); - } - Err(e) => { - tracing::error!( - "Failed to delete function {}:{} from Cassandra: {}", - function_id, - 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 { @@ -973,16 +859,6 @@ mod tests { } } - fn create_test_active_function_details() -> ActiveFunctionDetails { - ActiveFunctionDetails { - function_id: Uuid::new_v4(), - function_version_id: Uuid::new_v4(), - nca_id: Some("test-nca-id".to_string()), - num_workers: Some(1), - last_updated_at: Some(Utc::now()), - } - } - fn create_test_lock() -> DistributedLock { DistributedLock { lock_name: "test_lock".to_string(), @@ -1012,59 +888,6 @@ mod tests { assert!(result.is_ok()); } - #[tokio::test] - #[ignore = "Requires running Cassandra"] - async fn test_active_function_operations() { - let settings = create_test_settings().await; - let secrets_path = get_test_secrets_path(); - let secrets_watcher = Arc::new( - SecretFileWatcher::new(Path::new(&secrets_path)) - .await - .unwrap(), - ); - let manager = CassandraServiceManager::new(&settings, secrets_watcher) - .await - .unwrap(); - - for table_type in [ - ActiveFunctionTable::RecentlyInvokedFunctions, - ActiveFunctionTable::RunningFunctionsWithoutInvocations, - ] { - let function = create_test_active_function_details(); - manager - .add_new_active_functions_batch(std::slice::from_ref(&function), table_type) - .await - .unwrap(); - - let token_range = CASSANDRA_TOKEN_RANGE; - let functions = manager - .get_active_functions_with_token_range(&token_range, 100, table_type) - .await - .unwrap(); - assert!(functions.iter().any(|active| { - active.function_id == function.function_id - && active.function_version_id == function.function_version_id - })); - - manager - .delete_active_function( - &function.function_id, - &function.function_version_id, - table_type, - ) - .await - .unwrap(); - let functions = manager - .get_active_functions_with_token_range(&token_range, 100, table_type) - .await - .unwrap(); - assert!(!functions.iter().any(|active| { - active.function_id == function.function_id - && active.function_version_id == function.function_version_id - })); - } - } - #[tokio::test] #[ignore = "Requires running Cassandra"] async fn test_lock_operations() { 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 78596e86e..8195417c3 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,6 +29,7 @@ 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 + #[serde_as] #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(default)] 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 6e97d26eb..47cc7dc90 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 @@ -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!( @@ -58,35 +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 - ) -} - pub(crate) fn get_health_check_query_stmt(keyspace: &str) -> String { format!("SELECT now() from {}.healthy_nodes LIMIT 1;", keyspace) } @@ -133,22 +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 - ) -} - #[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() { @@ -157,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")); + } } 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 910329f2d..b9f086c8f 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 @@ -19,12 +19,10 @@ 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, pub last_updated_at: Option>, pub num_workers: Option, @@ -40,16 +38,6 @@ impl ActiveFunctionDetails { num_workers: 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)] 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 f23a1fcf1..233ad8adc 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 @@ -15,9 +15,7 @@ * limitations under the License. */ -use crate::cassandra::{ - cassandra_service::CassandraServiceManager, statements::ActiveFunctionTable, -}; +use crate::cassandra::cassandra_service::CassandraServiceManager; use crate::metrics; use crate::models::ActiveFunctionDetails; use crate::timeseries_db::timeseries_db_client::TimeseriesDbClient; @@ -277,11 +275,7 @@ async fn fetch_function_state( let page_size = 2000; let db_recently_invoked = cassandra_service - .get_active_functions_with_token_range( - &range, - page_size, - ActiveFunctionTable::RecentlyInvokedFunctions, - ) + .get_active_functions_with_token_range(&range, page_size) .await?; let timeseries_db_active_functions = @@ -429,10 +423,7 @@ async fn execute_function_actions( ); cassandra_service - .add_new_active_functions_batch( - &actions.add_recently_invoked, - ActiveFunctionTable::RecentlyInvokedFunctions, - ) + .add_new_active_functions_batch(&actions.add_recently_invoked) .await?; for function in &actions.add_recently_invoked { diff --git a/src/control-plane-services/function-autoscaler/crates/server/src/work/mod.rs b/src/control-plane-services/function-autoscaler/crates/server/src/work/mod.rs index 05b51225e..ba73cfd45 100644 --- a/src/control-plane-services/function-autoscaler/crates/server/src/work/mod.rs +++ b/src/control-plane-services/function-autoscaler/crates/server/src/work/mod.rs @@ -16,7 +16,6 @@ */ use crate::cassandra::distributed_lock::DistributedLockManager; -use crate::cassandra::statements::ActiveFunctionTable; use crate::health::HealthStatus; use crate::metrics; use crate::models::NodeHealth; @@ -88,6 +87,14 @@ use discovery::get_recently_invoked_functions; const TIMESERIES_DB_QUERY_STEP: StdDuration = StdDuration::from_secs(60); // 1 minute step for TimeseriesDb queries pub const CALCULATE_UTILIZATION_LOCK_PREFIX: &str = "util_lock"; +const ACTIVE_FUNCTION_SET_NAME: &str = "RecentlyInvokedFunctions"; + +fn scaling_lock_name(bucket_index: usize) -> String { + format!( + "{}_{}_{}", + CALCULATE_UTILIZATION_LOCK_PREFIX, bucket_index, ACTIVE_FUNCTION_SET_NAME + ) +} #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct MetricEnvironments { @@ -497,10 +504,6 @@ async fn llm_gateway_recently_invoked( } /// Get current worker count for non-BYOC functions from TimeseriesDb. -/// We use this instead of details.num_workers because the history table is only updated when -/// discovery adds or moves a function; if a function stays in the same table, num_workers is -/// never refreshed and scaling would report stale counts (e.g. 2 when TimeseriesDb shows 3). -/// /// Returns `Ok(None)` when no worker series matched the query at all. Callers use that as the /// signal to try gateway metrics before control-plane metrics. `Ok(Some(0))` would mean /// "series exists but count parsed as 0," which we don't @@ -751,12 +754,10 @@ pub async fn run_autoscaling_logic_p0( ) -> Result<()> { tracing::info!("Starting P0 autoscaling logic for env: {}", env); - // Process recently invoked functions (P0 priority) - make_scaling_requests_for_table( + make_scaling_requests( cassandra_service, timeseries_db_client, nvcf_api_service, - ActiveFunctionTable::RecentlyInvokedFunctions, scaling_settings, env, ignore_env, @@ -770,13 +771,11 @@ pub async fn run_autoscaling_logic_p0( Ok(()) } -// Function that makes scaling requests for a specific table #[allow(clippy::too_many_arguments)] -async fn make_scaling_requests_for_table( +async fn make_scaling_requests( cassandra_service: Arc, timeseries_db_client: Arc, nvcf_api_service: Arc, - table: ActiveFunctionTable, scaling_settings: Arc, env: &str, ignore_env: bool, @@ -799,14 +798,13 @@ async fn make_scaling_requests_for_table( for (bucket_index, (start_token, end_token)) in bucket_ranges.iter() { let token_range = [*start_token, *end_token]; let functions_in_bucket = cassandra_service - .get_active_functions_with_token_range(&token_range, page_size, table) + .get_active_functions_with_token_range(&token_range, page_size) .await?; tracing::info!( - "Processing {} functions in bucket {} for table {:?}", + "Processing {} recently invoked functions in bucket {}", functions_in_bucket.len(), - bucket_index, - table + bucket_index ); if functions_in_bucket.is_empty() { @@ -814,10 +812,7 @@ async fn make_scaling_requests_for_table( } // Try to acquire distributed lock for this bucket - let lock_name = format!( - "{}_{}_{:?}", - CALCULATE_UTILIZATION_LOCK_PREFIX, bucket_index, table - ); + let lock_name = scaling_lock_name(*bucket_index); if let Ok(Some(_lock_guard)) = lock_manager .try_acquire( lock_name, @@ -1049,7 +1044,7 @@ async fn make_scaling_requests_for_table( if let Some((bucket_index, error)) = first_task_error { Err(error.context(format!( - "{task_failures} per-function scaling task(s) failed for table {table:?}; first failure was in bucket {bucket_index}" + "{task_failures} per-function scaling task(s) failed; first failure was in bucket {bucket_index}" ))) } else { Ok(()) @@ -1121,6 +1116,11 @@ mod tests { use crate::timeseries_db::TimeseriesDbSettings; use uuid::Uuid; + #[test] + fn scaling_lock_name_preserves_existing_coordination_key() { + assert_eq!(scaling_lock_name(7), "util_lock_7_RecentlyInvokedFunctions"); + } + // ---- Helpers for the metric-acquisition tests ---- /// Tiny retry budget so error paths resolve in milliseconds, not seconds. diff --git a/src/control-plane-services/function-autoscaler/local_env/cassandra/schema/0001_initial_schema.cql b/src/control-plane-services/function-autoscaler/local_env/cassandra/schema/0001_initial_schema.cql index 806fec835..5e2c5a412 100644 --- a/src/control-plane-services/function-autoscaler/local_env/cassandra/schema/0001_initial_schema.cql +++ b/src/control-plane-services/function-autoscaler/local_env/cassandra/schema/0001_initial_schema.cql @@ -25,44 +25,6 @@ CREATE TABLE IF NOT EXISTS nvcf_autoscaler.recently_invoked_functions ( PRIMARY KEY ((function_id, function_version_id)) ) WITH default_time_to_live = 600; -CREATE TABLE IF NOT EXISTS nvcf_autoscaler.recently_invoked_functions_history ( - function_id uuid, - function_version_id uuid, - nca_id_string text STATIC, -- STATIC: same for all rows in partition (text, not uuid) - last_updated_at timestamp, - num_workers int STATIC, -- Cache the number of workers available (>=0) - last_predicted_desired_instance_count int, -- Cache last result - last_predicted_error_code TEXT, -- will be a code that provides result code. - PRIMARY KEY ((function_id, function_version_id), last_updated_at) -) WITH CLUSTERING ORDER BY (last_updated_at desc) - AND default_time_to_live = 172800; - -CREATE TABLE IF NOT EXISTS nvcf_autoscaler.running_functions_without_invocations ( - function_id uuid, - function_version_id uuid, - nca_id_string text, -- Regular column, not part of PK (text, not uuid) - last_updated_at timestamp, - PRIMARY KEY ((function_id, function_version_id)) -) WITH default_time_to_live = 600; - --- The table doesn't have a default TTL (Time To Live) setting. --- Instead, we'll implement a configurable cleanup task that runs hourly to --- remove entries for inactive functions. This cleanup job will be adjustable --- - we can modify its frequency or disable it entirely. In contrast, the --- prediction history entries will automatically expire after 10 minutes due --- to their TTL setting. -CREATE TABLE IF NOT EXISTS nvcf_autoscaler.running_functions_without_invocations_history ( - function_id uuid, - function_version_id uuid, - nca_id_string text STATIC, -- STATIC: same for all rows in partition (text, not uuid) - last_updated_at timestamp, - num_workers int STATIC, -- Cache the number of workers available (>0) - last_predicted_desired_instance_count int, -- Cache last result - last_predicted_error_code TEXT, -- will be a code that provides result code. - PRIMARY KEY ((function_id, function_version_id), last_updated_at) -) WITH CLUSTERING ORDER BY (last_updated_at desc) - AND default_time_to_live = 172800; - CREATE TABLE IF NOT EXISTS nvcf_autoscaler.locks ( lock_name text PRIMARY KEY, node_id text,