fix(function-autoscaler): align control-plane metric timestamps (#1919)

Signed-off-by: Bora Oztekin <boztekin@nvidia.com>

Author: borao

src/control-plane-services/function-autoscaler/crates/server/src/work/mod.rs

@@ -29,7 +29,7 @@ use crate::{
     timeseries_db::timeseries_db_client::TimeseriesDbClient,
 };
 use anyhow::{Context, Result};
-use chrono::{Duration, Utc};
+use chrono::{DateTime, Duration, Utc};
 use moka::sync::Cache;
 
 use std::sync::Arc;
@@ -90,6 +90,17 @@ const TIMESERIES_DB_QUERY_STEP: StdDuration = StdDuration::from_secs(60); // 1 m
 pub const CALCULATE_UTILIZATION_LOCK_PREFIX: &str = "util_lock";
 const ACTIVE_FUNCTION_SET_NAME: &str = "RecentlyInvokedFunctions";
 
+/// Return the previous complete TimeseriesDb step.
+///
+/// The current step may contain only a subset of the latest scrape cycle.
+fn settled_timeseries_end_time(now: DateTime<Utc>) -> DateTime<Utc> {
+    let step_seconds = TIMESERIES_DB_QUERY_STEP.as_secs() as i64;
+    let current_boundary = now.timestamp().div_euclid(step_seconds) * step_seconds;
+    let settled_timestamp = current_boundary.saturating_sub(step_seconds);
+
+    DateTime::from_timestamp(settled_timestamp, 0).unwrap_or(now)
+}
+
 fn scaling_lock_name(bucket_index: usize) -> String {
     format!(
         "{}_{}_{}",
@@ -131,10 +142,10 @@ async fn get_function_utilization_history(
     env: &str,
     metric_source: MetricSource,
     ignore_env: bool,
+    end_time: DateTime<Utc>,
     lookback: StdDuration,
     utilization_window_seconds: u64,
 ) -> Result<Vec<(i64, String)>> {
-    let end_time = Utc::now();
     let start_time = end_time - Duration::from_std(lookback)?;
     let step = TIMESERIES_DB_QUERY_STEP;
 
@@ -268,6 +279,7 @@ async fn get_byoc_instance_count(
     function_version_id: &Uuid,
     env: &str,
     ignore_env: bool,
+    end_time: DateTime<Utc>,
 ) -> Result<Option<usize>> {
     let env_suffix = if ignore_env {
         String::new()
@@ -283,14 +295,7 @@ async fn get_byoc_instance_count(
         env = env_suffix
     );
 
-    // Align end to the previous fully-settled step boundary (one step back from now).
-    // Reading the bleeding edge can pick up a partial scrape cycle and report a wrong count.
-    const STEP_SECS: i64 = 60;
-    let now_secs = Utc::now().timestamp();
-    let end_secs = (now_secs / STEP_SECS) * STEP_SECS - STEP_SECS;
-    let end_time = chrono::DateTime::from_timestamp(end_secs, 0).unwrap_or_else(Utc::now);
-    let start_time = end_time - chrono::Duration::seconds(STEP_SECS);
-    let step = std::time::Duration::from_secs(STEP_SECS as u64);
+    let start_time = end_time - Duration::from_std(TIMESERIES_DB_QUERY_STEP)?;
 
     tracing::info!(
         "BYOC instance count query for {}:{}: {}",
@@ -300,7 +305,7 @@ async fn get_byoc_instance_count(
     );
 
     let response = timeseries_db_client
-        .query_range(&query, start_time, end_time, step)
+        .query_range(&query, start_time, end_time, TIMESERIES_DB_QUERY_STEP)
         .await?;
 
     tracing::info!(
@@ -636,6 +641,12 @@ async fn gather_scaling_inputs(
         }
     }
 
+    // Control-plane instance count and utilization must describe the same settled snapshot.
+    // Otherwise a running-instance transition can combine an old count with new utilization
+    // and produce contradictory targets.
+    let control_plane_query_end_time = (metric_source == MetricSource::ControlPlane)
+        .then(|| settled_timeseries_end_time(Utc::now()));
+
     match metric_source {
         MetricSource::WorkerThreads => {}
         MetricSource::LlmGateway => {
@@ -657,6 +668,7 @@ async fn gather_scaling_inputs(
                 function_version_id,
                 env,
                 ignore_env,
+                control_plane_query_end_time.expect("control-plane query end time must be present"),
             )
             .await
             .unwrap_or_else(|error| {
@@ -694,6 +706,7 @@ async fn gather_scaling_inputs(
         env,
         metric_source,
         ignore_env,
+        control_plane_query_end_time.unwrap_or_else(Utc::now),
         scaling_settings.lookback,
         scaling_settings.utilization_window_seconds,
     )
@@ -1156,6 +1169,17 @@ mod tests {
         assert_eq!(scaling_lock_name(7), "util_lock_7_RecentlyInvokedFunctions");
     }
 
+    #[test]
+    fn settled_timeseries_end_time_uses_previous_complete_step() {
+        let now = DateTime::from_timestamp(1_700_000_123, 456_000_000)
+            .expect("valid timestamp");
+
+        assert_eq!(
+            settled_timeseries_end_time(now),
+            DateTime::from_timestamp(1_700_000_040, 0).expect("valid timestamp"),
+        );
+    }
+
     #[test]
     fn utilization_data_age_never_goes_negative() {
         assert_eq!(
@@ -1325,6 +1349,7 @@ mod tests {
             "stg",
             MetricSource::ControlPlane,
             true,
+            Utc::now(),
             StdDuration::from_secs(5 * 60),
             60,
         )
@@ -1367,6 +1392,7 @@ mod tests {
                 configured_env,
                 MetricSource::ControlPlane,
                 false,
+                Utc::now(),
                 StdDuration::from_secs(5 * 60),
                 60,
             )
@@ -1521,6 +1547,7 @@ mod tests {
                 "prod",
                 MetricSource::LlmGateway,
                 false,
+                Utc::now(),
                 StdDuration::from_secs(5 * 60),
                 70,
             )
@@ -1638,13 +1665,13 @@ mod tests {
         let client = ts_client(server.url());
 
         assert_eq!(
-            get_byoc_instance_count(&client, &fid, &fvid, "prd", false)
+            get_byoc_instance_count(&client, &fid, &fvid, "prd", false, Utc::now())
                 .await
                 .expect("prod cp instance count"),
             Some(3)
         );
         assert_eq!(
-            get_byoc_instance_count(&client, &fid, &fvid, "stg", false)
+            get_byoc_instance_count(&client, &fid, &fvid, "stg", false, Utc::now())
                 .await
                 .expect("stage cp instance count"),
             Some(3)
@@ -1675,6 +1702,7 @@ mod tests {
                 &fvid,
                 "stg",
                 true,
+                Utc::now(),
             )
             .await
             .expect("missing cp instance count"),
@@ -1703,6 +1731,7 @@ mod tests {
                 &fvid,
                 "stg",
                 true,
+                Utc::now(),
             )
             .await
             .expect("zero cp instance count"),