From 16ba109aa4a21ccaf2f4b7b9cb9456c0d032ced2 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Thu, 4 Jun 2026 18:25:02 -0400
Subject: [PATCH 01/12] Add return_timestamp parameter to wait_for_nodes_ready

---
 modules/python/clients/aks_client.py          | 203 ++++++++++++++----
 modules/python/clients/kubernetes_client.py   |   9 +-
 .../python/tests/clients/test_aks_client.py   |  86 ++++++--
 .../tests/clients/test_kubernetes_client.py   |  22 ++
 4 files changed, 257 insertions(+), 63 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index 4e571b2840..e50f9de0d2 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -12,11 +12,12 @@
 and troubleshooting.
 """
 
+import asyncio
 import logging
 import os
 import subprocess
 import time
-from typing import Dict, Optional, Any
+from typing import Dict, Optional, Any, Tuple
 
 # Third party imports
 from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
@@ -61,6 +62,85 @@ def _get_operation_context(self):
 
         return OperationContext
 
+    def _run_concurrent_arm_and_readiness(
+        self,
+        poller,
+        node_count: int,
+        label_selector: str,
+        start_time: float
+    ) -> Tuple[Any, list, float, float]:
+        """
+        Run ARM operation and K8s node readiness check concurrently.
+
+        This allows accurate measurement of both ARM completion time and node readiness time
+        independently, enabling identification of which layer is causing latency.
+
+        Args:
+            poller: Azure ARM LROPoller from begin_create_or_update (thread-safe)
+            node_count: Expected number of nodes to be ready
+            label_selector: K8s label selector for filtering nodes
+            start_time: Operation start timestamp for calculating durations
+
+        Returns:
+            Tuple of (arm_result, ready_nodes, node_readiness_time, command_execution_time)
+            - node_readiness_time: seconds from start until K8s nodes were ready
+            - command_execution_time: seconds from start until ARM operation completed
+
+        Raises:
+            Exception: If either ARM or K8s readiness fails. Both tasks run to
+                completion (or failure) so we can capture timing for whichever
+                succeeded, enabling better diagnosis of which layer caused the failure.
+        """
+        def _poll_arm_with_timestamp():
+            """Run ARM poller and return (result, completion_timestamp)."""
+            result = poller.result()
+            return result, time.time()
+
+        async def _run():
+            # Run ARM polling and K8s readiness check concurrently
+            arm_task = asyncio.to_thread(_poll_arm_with_timestamp)
+            k8s_task = asyncio.to_thread(
+                self.k8s_client.wait_for_nodes_ready,
+                node_count=node_count,
+                operation_timeout_in_minutes=self.operation_timeout_minutes,
+                label_selector=label_selector,
+                return_timestamp=True,
+            )
+
+            # Use return_exceptions=True to capture both results even if one fails
+            results = await asyncio.gather(arm_task, k8s_task, return_exceptions=True)
+            arm_result, k8s_result = results
+
+            # Check for failures and build diagnostic message
+            arm_failed = isinstance(arm_result, Exception)
+            k8s_failed = isinstance(k8s_result, Exception)
+
+            if arm_failed or k8s_failed:
+                # Build diagnostic info for logging
+                arm_status = f"FAILED: {arm_result}" if arm_failed else "succeeded"
+                k8s_status = f"FAILED: {k8s_result}" if k8s_failed else "succeeded"
+                elapsed = time.time() - start_time
+                logger.error(
+                    "Concurrent operation failed after %.2fs - ARM: %s, K8s readiness: %s",
+                    elapsed, arm_status, k8s_status
+                )
+                # Raise the first exception encountered
+                if arm_failed:
+                    raise arm_result
+                raise k8s_result
+
+            # Both succeeded - unpack results
+            arm_response, arm_timestamp = arm_result
+            ready_nodes, ready_timestamp = k8s_result
+
+            # Calculate times relative to start
+            node_readiness_time = ready_timestamp - start_time
+            command_execution_time = arm_timestamp - start_time
+
+            return arm_response, ready_nodes, node_readiness_time, command_execution_time
+
+        return asyncio.run(_run())
+
     def __init__(
         self,
         subscription_id: Optional[str] = None,
@@ -508,6 +588,10 @@ def create_node_pool(
                     f"Creating node pool {node_pool_name} in cluster {cluster_name}"
                 )
 
+                # Capture start time for timing measurements
+                start_time = time.time()
+                label_selector = f"agentpool={node_pool_name}"
+
                 if enable_managed_gpu:
                     # Fully managed GPU: use az CLI (aks-preview) since the stable SDK
                     # doesn't expose gpuProfile.nvidia.managementMode
@@ -519,26 +603,39 @@ def create_node_pool(
                         gpu_instance_profile=gpu_instance_profile,
                         gpu_mig_strategy=gpu_mig_strategy,
                     )
+                    command_execution_time = time.time() - start_time
+                    ready_nodes, ready_timestamp = self.k8s_client.wait_for_nodes_ready(
+                        node_count=node_count,
+                        operation_timeout_in_minutes=self.operation_timeout_minutes,
+                        label_selector=label_selector,
+                        return_timestamp=True,
+                    )
+                    node_readiness_time = ready_timestamp - start_time
                 else:
-                    # Create the node pool via SDK
-                    self.aks_client.agent_pools.begin_create_or_update(
-                        resource_group_name=self.resource_group,
-                        resource_name=cluster_name,
-                        agent_pool_name=node_pool_name,
-                        parameters=parameters,
-                    ).result()
-
-                label_selector = f"agentpool={node_pool_name}"
+                    # Run ARM and K8s readiness concurrently to capture both timings
+                    _, ready_nodes, node_readiness_time, command_execution_time = \
+                        self._run_concurrent_arm_and_readiness(
+                            self.aks_client.agent_pools.begin_create_or_update(
+                                resource_group_name=self.resource_group,
+                                resource_name=cluster_name,
+                                agent_pool_name=node_pool_name,
+                                parameters=parameters,
+                            ),
+                            node_count, label_selector, start_time
+                        )
 
-                # Wait for nodes to be ready
-                ready_nodes = self.k8s_client.wait_for_nodes_ready(
-                    node_count=node_count,
-                    operation_timeout_in_minutes=self.operation_timeout_minutes,
-                    label_selector=label_selector,
+                logger.info(
+                    f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
 
+                # Add timing metadata for regression analysis
+                op.add_metadata("node_readiness_time", node_readiness_time)
+                op.add_metadata("command_execution_time", command_execution_time)
+
+                # Log timing - analysis happens in ADX/dashboards
                 logger.info(
-                    f"All {node_count} nodes in pool {node_pool_name} are ready"
+                    "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
+                    node_pool_name, node_readiness_time, command_execution_time
                 )
 
                 # Verify NVIDIA drivers for managed GPU only (fully managed uses systemd)
@@ -688,28 +785,41 @@ def scale_node_pool(
                 node_pool.count = node_count
 
                 logger.info(f"Scaling node pool {node_pool_name} to {node_count} nodes")
-                self._begin_update_with_retry(
-                    node_pool_name=node_pool_name,
-                    cluster_name=cluster_name,
-                    node_pool=node_pool,
-                )
+
+                # Capture start time for timing measurements
+                start_time = time.time()
+                label_selector = f"agentpool={node_pool_name}"
 
                 logger.info(
                     f"Waiting for {node_count} nodes in pool {node_pool_name} to be ready..."
                 )
 
-                # Use agentpool=node_pool_name as default label if not specified
-                label_selector = f"agentpool={node_pool_name}"
+                # Run ARM and K8s readiness concurrently to capture both timings
+                _, ready_nodes, node_readiness_time, command_execution_time = \
+                    self._run_concurrent_arm_and_readiness(
+                        self.aks_client.agent_pools.begin_create_or_update(
+                            resource_group_name=self.resource_group,
+                            resource_name=cluster_name,
+                            agent_pool_name=node_pool_name,
+                            parameters=node_pool,
+                        ),
+                        node_count, label_selector, start_time
+                    )
 
-                ready_nodes = self.k8s_client.wait_for_nodes_ready(
-                    node_count=node_count,
-                    operation_timeout_in_minutes=self.operation_timeout_minutes,
-                    label_selector=label_selector,
-                )
                 logger.info(
                     f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
 
+                # Add timing metadata for regression analysis
+                op.add_metadata("node_readiness_time", node_readiness_time)
+                op.add_metadata("command_execution_time", command_execution_time)
+
+                # Log timing - analysis happens in ADX/dashboards
+                logger.info(
+                    "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
+                    node_pool_name, node_readiness_time, command_execution_time
+                )
+
                 pod_logs = None
                 if gpu_node_pool and not enable_managed_gpu and operation_type == "scale_up" and node_count > 0:
                     logger.info(
@@ -924,24 +1034,35 @@ def _progressive_scale(
                         "cluster_info", self.get_cluster_data(cluster_name)
                     )
                     node_pool.count = step  # Update node count in the node pool object
-                    self._begin_update_with_retry(
-                        node_pool_name=node_pool_name,
-                        cluster_name=cluster_name,
-                        node_pool=node_pool,
-                        label=f"step {step} ",
-                    )
-                    result = node_pool
 
-                    # Use agentpool=node_pool_name as default label if not specified
+                    # Capture start time for timing measurements
+                    start_time = time.time()
                     label_selector = f"agentpool={node_pool_name}"
 
-                    ready_nodes = self.k8s_client.wait_for_nodes_ready(
-                        node_count=step,
-                        operation_timeout_in_minutes=self.operation_timeout_minutes,
-                        label_selector=label_selector,
-                    )
+                    # Run ARM and K8s readiness concurrently to capture both timings
+                    result, ready_nodes, node_readiness_time, command_execution_time = \
+                        self._run_concurrent_arm_and_readiness(
+                            self.aks_client.agent_pools.begin_create_or_update(
+                                resource_group_name=self.resource_group,
+                                resource_name=cluster_name,
+                                agent_pool_name=node_pool_name,
+                                parameters=node_pool,
+                            ),
+                            step, label_selector, start_time
+                        )
+
                     logger.info(f"All {step} nodes in pool {node_pool_name} are ready")
 
+                    # Add timing metadata for regression analysis
+                    op.add_metadata("node_readiness_time", node_readiness_time)
+                    op.add_metadata("command_execution_time", command_execution_time)
+
+                    # Log timing - analysis happens in ADX/dashboards
+                    logger.info(
+                        "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
+                        node_pool_name, node_readiness_time, command_execution_time
+                    )
+
                     if result is None:
                         logger.error(f"Progressive scaling failed at step {step}")
                         op.add_metadata("error", "Scaling operation returned None")
diff --git a/modules/python/clients/kubernetes_client.py b/modules/python/clients/kubernetes_client.py
index 9d7b171353..d5b9eafb6f 100644
--- a/modules/python/clients/kubernetes_client.py
+++ b/modules/python/clients/kubernetes_client.py
@@ -237,15 +237,16 @@ def delete_node(self, node_name):
             else:
                 raise Exception(f"Error deleting Node '{node_name}': {str(e)}") from e
 
-    def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_selector=None):
+    def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_selector=None, return_timestamp=False):
         """
         Waits for a specific number of nodes with a given label to be ready within a specified timeout.
         Raises an exception if the expected number of nodes are not ready within the timeout.
 
-        :param node_label: The label to filter nodes.
         :param node_count: The expected number of nodes to be ready.
         :param operation_timeout_in_minutes: The timeout in minutes to wait for the nodes to be ready.
-        :return: None
+        :param label_selector: The label to filter nodes.
+        :param return_timestamp: If True, returns (ready_nodes, ready_timestamp) tuple for timing measurement.
+        :return: List of ready nodes, or (ready_nodes, timestamp) if return_timestamp=True.
         """
         ready_nodes = []
         ready_node_count = 0
@@ -256,6 +257,8 @@ def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_s
             ready_node_count = len(ready_nodes)
             logger.info(f"Currently {ready_node_count} nodes are ready.")
             if ready_node_count == node_count:
+                if return_timestamp:
+                    return ready_nodes, time.time()
                 return ready_nodes
             logger.info(f"Waiting for {node_count} nodes to be ready.")
             time.sleep(10)
diff --git a/modules/python/tests/clients/test_aks_client.py b/modules/python/tests/clients/test_aks_client.py
index d9f6a6f11f..07f70d832d 100644
--- a/modules/python/tests/clients/test_aks_client.py
+++ b/modules/python/tests/clients/test_aks_client.py
@@ -174,13 +174,17 @@ def test_create_node_pool_success(self, mock_time):
         vm_size = "Standard_DS2_v2"
         node_count = 2
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_operation = mock.MagicMock()
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Mock the node pool that will be retrieved after creation
         mock_created_node_pool = mock.MagicMock()
@@ -217,8 +221,15 @@ def mock_get_node_pool_side_effect(name, cluster=None):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
 
+        # Verify timing measurements are calculated and stored correctly
+        # node_readiness_time = nodes_ready_time - start_time = 130 - 100 = 30
+        # command_execution_time = arm_done_time - start_time = 150 - 100 = 50
+        self.mock_operation.add_metadata.assert_any_call("node_readiness_time", 30)
+        self.mock_operation.add_metadata.assert_any_call("command_execution_time", 50)
+
     @mock.patch("clients.aks_client.time")
     def test_create_node_pool_gpu(self, mock_time):
         """Test creating a GPU node pool with driver verification"""
@@ -227,13 +238,17 @@ def test_create_node_pool_gpu(self, mock_time):
         vm_size = "Standard_NC6s_v3"  # GPU VM size
         node_count = 1
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_operation = mock.MagicMock()
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -275,6 +290,7 @@ def mock_get_node_pool_side_effect(name, cluster=None):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
 
         # Check that NVIDIA verification was performed
@@ -329,7 +345,11 @@ def test_scale_node_pool_up(self, mock_time):
         node_pool_name = "test-pool"
         node_count = 3
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -341,7 +361,7 @@ def test_scale_node_pool_up(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Mock get_cluster_data to return a dictionary for JSON serialization
         self.aks_client.get_cluster_data = mock.MagicMock(
@@ -363,9 +383,16 @@ def test_scale_node_pool_up(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
+        # Verify timing measurements are calculated and stored correctly
+        # node_readiness_time = nodes_ready_time - start_time = 130 - 100 = 30
+        # command_execution_time = arm_done_time - start_time = 150 - 100 = 50
+        self.mock_operation.add_metadata.assert_any_call("node_readiness_time", 30)
+        self.mock_operation.add_metadata.assert_any_call("command_execution_time", 50)
+
     @mock.patch("clients.aks_client.time")
     def test_scale_node_pool_down(self, mock_time):
         """Test scaling a node pool down"""
@@ -373,7 +400,11 @@ def test_scale_node_pool_down(self, mock_time):
         node_pool_name = "test-pool"
         node_count = 1
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 3  # Current count
@@ -390,7 +421,7 @@ def test_scale_node_pool_down(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Mock get_node_pool to return the node pool with as_dict method
         self.aks_client.get_node_pool = mock.MagicMock(return_value=mock_node_pool)
@@ -407,6 +438,7 @@ def test_scale_node_pool_down(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
@@ -454,7 +486,11 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
         # Set VM size that triggers NVIDIA verification
         self.aks_client.vm_size = "Standard_NC40ads_H100_v5"
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -477,7 +513,7 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -498,6 +534,7 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
@@ -514,12 +551,15 @@ def test_scale_gpu_node_pool_up_progressive_final_step(self, mock_time):
         # Set VM size that triggers NVIDIA verification
         self.aks_client.vm_size = "Standard_NC40ads_H100_v5"
 
-        mock_time.time.side_effect = [
-            100,
-            150,
-            200,
-            250,
-        ]  # Start and end times including progressive scaling
+        # Define timestamps for clarity (progressive has multiple steps)
+        # Each step calls time.time() twice: once for start, once for arm completion
+        step1_start = 100
+        step1_nodes_ready = 130
+        step1_arm_done = 150
+        step2_start = 200
+        step2_nodes_ready = 230
+        step2_arm_done = 250
+        mock_time.time.side_effect = [step1_start, step1_arm_done, step2_start, step2_arm_done]
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -548,7 +588,10 @@ def test_scale_gpu_node_pool_up_progressive_final_step(self, mock_time):
             mock.MagicMock(),
             mock.MagicMock(),
         ]  # Second step to 3 nodes
-        self.mock_k8s.wait_for_nodes_ready.side_effect = [ready_nodes1, ready_nodes2]
+        self.mock_k8s.wait_for_nodes_ready.side_effect = [
+            (ready_nodes1, step1_nodes_ready),
+            (ready_nodes2, step2_nodes_ready)
+        ]
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -584,7 +627,11 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
         node_pool_name = "gpu-pool"
         node_count = 1
 
-        mock_time.time.side_effect = [100, 150]  # Start and end times
+        # Define timestamps for clarity
+        start_time = 100
+        nodes_ready_time = 130
+        arm_done_time = 150
+        mock_time.time.side_effect = [start_time, arm_done_time]
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 3  # Current count
@@ -602,7 +649,7 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -623,6 +670,7 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
+            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
diff --git a/modules/python/tests/clients/test_kubernetes_client.py b/modules/python/tests/clients/test_kubernetes_client.py
index 0ab752312e..1467339662 100644
--- a/modules/python/tests/clients/test_kubernetes_client.py
+++ b/modules/python/tests/clients/test_kubernetes_client.py
@@ -1203,6 +1203,28 @@ def test_wait_for_nodes_ready_exception(self, mock_sleep, mock_get_ready_nodes):
         self.assertIn("Only 1 nodes are ready, expected 2 nodes!", str(context.exception))
         mock_sleep.assert_called()
 
+    @patch('time.time')
+    @patch('clients.kubernetes_client.KubernetesClient.get_ready_nodes')
+    @patch("time.sleep", return_value=None)
+    def test_wait_for_nodes_ready_with_timestamp(self, mock_sleep, mock_get_ready_nodes, mock_time):
+        """Test waiting for nodes with return_timestamp=True returns tuple."""
+        mock_get_ready_nodes.side_effect = [[], ["node1", "node2"]]
+        mock_time.return_value = 1234567890.123  # Fixed timestamp for assertion
+        node_count = 2
+        timeout = 0.01
+
+        result = self.client.wait_for_nodes_ready(
+            node_count, timeout, return_timestamp=True
+        )
+
+        # Should return (ready_nodes, timestamp) tuple
+        self.assertIsInstance(result, tuple)
+        self.assertEqual(len(result), 2)
+        ready_nodes, timestamp = result
+        self.assertEqual(ready_nodes, ["node1", "node2"])
+        self.assertEqual(timestamp, 1234567890.123)
+        mock_sleep.assert_called()
+
     @patch('clients.kubernetes_client.KubernetesClient.get_pods_by_namespace')
     @patch('clients.kubernetes_client.KubernetesClient.get_ready_pods_by_namespace')
     @patch("time.sleep", return_value=None)

From b00a147fc45d33cc0b1d8115d526d54ce0c66b1b Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 02:33:57 -0400
Subject: [PATCH 02/12] Replace asyncio with ThreadPoolExecutor for concurrent
 timing

Use ThreadPoolExecutor instead of asyncio.run() to avoid implicit requirement that callers must not be in an existing event loop. Keeps the same method signature and behavior.
---
 modules/python/clients/aks_client.py | 69 +++++++++++++---------------
 1 file changed, 32 insertions(+), 37 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index e50f9de0d2..f5534b9821 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -12,11 +12,11 @@
 and troubleshooting.
 """
 
-import asyncio
 import logging
 import os
 import subprocess
 import time
+from concurrent.futures import ThreadPoolExecutor, wait
 from typing import Dict, Optional, Any, Tuple
 
 # Third party imports
@@ -70,7 +70,7 @@ def _run_concurrent_arm_and_readiness(
         start_time: float
     ) -> Tuple[Any, list, float, float]:
         """
-        Run ARM operation and K8s node readiness check concurrently.
+        Run ARM operation and K8s node readiness check concurrently using threads.
 
         This allows accurate measurement of both ARM completion time and node readiness time
         independently, enabling identification of which layer is causing latency.
@@ -91,55 +91,50 @@ def _run_concurrent_arm_and_readiness(
                 completion (or failure) so we can capture timing for whichever
                 succeeded, enabling better diagnosis of which layer caused the failure.
         """
-        def _poll_arm_with_timestamp():
+        def _poll_arm():
             """Run ARM poller and return (result, completion_timestamp)."""
             result = poller.result()
             return result, time.time()
 
-        async def _run():
-            # Run ARM polling and K8s readiness check concurrently
-            arm_task = asyncio.to_thread(_poll_arm_with_timestamp)
-            k8s_task = asyncio.to_thread(
-                self.k8s_client.wait_for_nodes_ready,
+        def _wait_k8s():
+            """Wait for K8s nodes to become ready and return (nodes, timestamp)."""
+            return self.k8s_client.wait_for_nodes_ready(
                 node_count=node_count,
                 operation_timeout_in_minutes=self.operation_timeout_minutes,
                 label_selector=label_selector,
                 return_timestamp=True,
             )
 
-            # Use return_exceptions=True to capture both results even if one fails
-            results = await asyncio.gather(arm_task, k8s_task, return_exceptions=True)
-            arm_result, k8s_result = results
-
-            # Check for failures and build diagnostic message
-            arm_failed = isinstance(arm_result, Exception)
-            k8s_failed = isinstance(k8s_result, Exception)
-
-            if arm_failed or k8s_failed:
-                # Build diagnostic info for logging
-                arm_status = f"FAILED: {arm_result}" if arm_failed else "succeeded"
-                k8s_status = f"FAILED: {k8s_result}" if k8s_failed else "succeeded"
-                elapsed = time.time() - start_time
-                logger.error(
-                    "Concurrent operation failed after %.2fs - ARM: %s, K8s readiness: %s",
-                    elapsed, arm_status, k8s_status
-                )
-                # Raise the first exception encountered
-                if arm_failed:
-                    raise arm_result
-                raise k8s_result
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            arm_future = executor.submit(_poll_arm)
+            k8s_future = executor.submit(_wait_k8s)
+            wait([arm_future, k8s_future])
+
+        # Check for failures and build diagnostic message
+        arm_exc = arm_future.exception()
+        k8s_exc = k8s_future.exception()
 
-            # Both succeeded - unpack results
-            arm_response, arm_timestamp = arm_result
-            ready_nodes, ready_timestamp = k8s_result
+        if arm_exc or k8s_exc:
+            elapsed = time.time() - start_time
+            arm_status = f"FAILED: {arm_exc}" if arm_exc else "succeeded"
+            k8s_status = f"FAILED: {k8s_exc}" if k8s_exc else "succeeded"
+            logger.error(
+                "Concurrent operation failed after %.2fs - ARM: %s, K8s readiness: %s",
+                elapsed, arm_status, k8s_status
+            )
+            if arm_exc:
+                raise arm_exc
+            raise k8s_exc
 
-            # Calculate times relative to start
-            node_readiness_time = ready_timestamp - start_time
-            command_execution_time = arm_timestamp - start_time
+        # Both succeeded - unpack results
+        arm_response, arm_timestamp = arm_future.result()
+        ready_nodes, ready_timestamp = k8s_future.result()
 
-            return arm_response, ready_nodes, node_readiness_time, command_execution_time
+        # Calculate times relative to start
+        node_readiness_time = ready_timestamp - start_time
+        command_execution_time = arm_timestamp - start_time
 
-        return asyncio.run(_run())
+        return arm_response, ready_nodes, node_readiness_time, command_execution_time
 
     def __init__(
         self,

From 006c87d9287e4f0bfcca2292bf03f1da8c21cb04 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 02:35:32 -0400
Subject: [PATCH 03/12] Internalize ARM call and start_time in concurrent
 method

Move begin_create_or_update call and start_time capture inside the method. Callers now pass node_pool_name, cluster_name, parameters, and node_count instead of a pre-created poller and external timestamp.
---
 modules/python/clients/aks_client.py | 54 ++++++++++------------------
 1 file changed, 19 insertions(+), 35 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index f5534b9821..daa7f35d5b 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -64,10 +64,10 @@ def _get_operation_context(self):
 
     def _run_concurrent_arm_and_readiness(
         self,
-        poller,
+        node_pool_name: str,
+        cluster_name: str,
+        parameters: Any,
         node_count: int,
-        label_selector: str,
-        start_time: float
     ) -> Tuple[Any, list, float, float]:
         """
         Run ARM operation and K8s node readiness check concurrently using threads.
@@ -76,10 +76,10 @@ def _run_concurrent_arm_and_readiness(
         independently, enabling identification of which layer is causing latency.
 
         Args:
-            poller: Azure ARM LROPoller from begin_create_or_update (thread-safe)
+            node_pool_name: Name of the node pool being provisioned
+            cluster_name: Name of the AKS cluster
+            parameters: Parameters for begin_create_or_update (dict or node pool object)
             node_count: Expected number of nodes to be ready
-            label_selector: K8s label selector for filtering nodes
-            start_time: Operation start timestamp for calculating durations
 
         Returns:
             Tuple of (arm_result, ready_nodes, node_readiness_time, command_execution_time)
@@ -91,6 +91,16 @@ def _run_concurrent_arm_and_readiness(
                 completion (or failure) so we can capture timing for whichever
                 succeeded, enabling better diagnosis of which layer caused the failure.
         """
+        start_time = time.time()
+        label_selector = f"agentpool={node_pool_name}"
+
+        poller = self.aks_client.agent_pools.begin_create_or_update(
+            resource_group_name=self.resource_group,
+            resource_name=cluster_name,
+            agent_pool_name=node_pool_name,
+            parameters=parameters,
+        )
+
         def _poll_arm():
             """Run ARM poller and return (result, completion_timestamp)."""
             result = poller.result()
@@ -610,13 +620,7 @@ def create_node_pool(
                     # Run ARM and K8s readiness concurrently to capture both timings
                     _, ready_nodes, node_readiness_time, command_execution_time = \
                         self._run_concurrent_arm_and_readiness(
-                            self.aks_client.agent_pools.begin_create_or_update(
-                                resource_group_name=self.resource_group,
-                                resource_name=cluster_name,
-                                agent_pool_name=node_pool_name,
-                                parameters=parameters,
-                            ),
-                            node_count, label_selector, start_time
+                            node_pool_name, cluster_name, parameters, node_count
                         )
 
                 logger.info(
@@ -781,10 +785,6 @@ def scale_node_pool(
 
                 logger.info(f"Scaling node pool {node_pool_name} to {node_count} nodes")
 
-                # Capture start time for timing measurements
-                start_time = time.time()
-                label_selector = f"agentpool={node_pool_name}"
-
                 logger.info(
                     f"Waiting for {node_count} nodes in pool {node_pool_name} to be ready..."
                 )
@@ -792,13 +792,7 @@ def scale_node_pool(
                 # Run ARM and K8s readiness concurrently to capture both timings
                 _, ready_nodes, node_readiness_time, command_execution_time = \
                     self._run_concurrent_arm_and_readiness(
-                        self.aks_client.agent_pools.begin_create_or_update(
-                            resource_group_name=self.resource_group,
-                            resource_name=cluster_name,
-                            agent_pool_name=node_pool_name,
-                            parameters=node_pool,
-                        ),
-                        node_count, label_selector, start_time
+                        node_pool_name, cluster_name, node_pool, node_count
                     )
 
                 logger.info(
@@ -1030,20 +1024,10 @@ def _progressive_scale(
                     )
                     node_pool.count = step  # Update node count in the node pool object
 
-                    # Capture start time for timing measurements
-                    start_time = time.time()
-                    label_selector = f"agentpool={node_pool_name}"
-
                     # Run ARM and K8s readiness concurrently to capture both timings
                     result, ready_nodes, node_readiness_time, command_execution_time = \
                         self._run_concurrent_arm_and_readiness(
-                            self.aks_client.agent_pools.begin_create_or_update(
-                                resource_group_name=self.resource_group,
-                                resource_name=cluster_name,
-                                agent_pool_name=node_pool_name,
-                                parameters=node_pool,
-                            ),
-                            step, label_selector, start_time
+                            node_pool_name, cluster_name, node_pool, step
                         )
 
                     logger.info(f"All {step} nodes in pool {node_pool_name} are ready")

From 8658356dc69be6aa2b43ee0aaf6b8db5cf5845d0 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 02:38:07 -0400
Subject: [PATCH 04/12] Rename _run_concurrent_arm_and_readiness to
 _instrument_nodepool_provisioning

---
 modules/python/clients/aks_client.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index daa7f35d5b..2a4600667c 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -62,7 +62,7 @@ def _get_operation_context(self):
 
         return OperationContext
 
-    def _run_concurrent_arm_and_readiness(
+    def _instrument_nodepool_provisioning(
         self,
         node_pool_name: str,
         cluster_name: str,
@@ -619,7 +619,7 @@ def create_node_pool(
                 else:
                     # Run ARM and K8s readiness concurrently to capture both timings
                     _, ready_nodes, node_readiness_time, command_execution_time = \
-                        self._run_concurrent_arm_and_readiness(
+                        self._instrument_nodepool_provisioning(
                             node_pool_name, cluster_name, parameters, node_count
                         )
 
@@ -791,7 +791,7 @@ def scale_node_pool(
 
                 # Run ARM and K8s readiness concurrently to capture both timings
                 _, ready_nodes, node_readiness_time, command_execution_time = \
-                    self._run_concurrent_arm_and_readiness(
+                    self._instrument_nodepool_provisioning(
                         node_pool_name, cluster_name, node_pool, node_count
                     )
 
@@ -1026,7 +1026,7 @@ def _progressive_scale(
 
                     # Run ARM and K8s readiness concurrently to capture both timings
                     result, ready_nodes, node_readiness_time, command_execution_time = \
-                        self._run_concurrent_arm_and_readiness(
+                        self._instrument_nodepool_provisioning(
                             node_pool_name, cluster_name, node_pool, step
                         )
 

From d2afe5a78a267bc90ce2d838c227294c08f6a155 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 02:39:55 -0400
Subject: [PATCH 05/12] Enhance timing logs with bottleneck analysis and total
 elapsed

Log now shows which layer (ARM vs K8s) was the bottleneck, the delta between them, and the total elapsed time for the concurrent operation.
---
 modules/python/clients/aks_client.py          | 33 +++++++++++--------
 .../python/tests/clients/test_aks_client.py   |  2 +-
 2 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index 2a4600667c..a4430dd907 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -593,13 +593,10 @@ def create_node_pool(
                     f"Creating node pool {node_pool_name} in cluster {cluster_name}"
                 )
 
-                # Capture start time for timing measurements
-                start_time = time.time()
-                label_selector = f"agentpool={node_pool_name}"
-
                 if enable_managed_gpu:
                     # Fully managed GPU: use az CLI (aks-preview) since the stable SDK
                     # doesn't expose gpuProfile.nvidia.managementMode
+                    start_time = time.time()
                     self.add_managed_gpu_node_pool(
                         node_pool_name=node_pool_name,
                         cluster_name=cluster_name,
@@ -609,6 +606,7 @@ def create_node_pool(
                         gpu_mig_strategy=gpu_mig_strategy,
                     )
                     command_execution_time = time.time() - start_time
+                    label_selector = f"agentpool={node_pool_name}"
                     ready_nodes, ready_timestamp = self.k8s_client.wait_for_nodes_ready(
                         node_count=node_count,
                         operation_timeout_in_minutes=self.operation_timeout_minutes,
@@ -631,10 +629,13 @@ def create_node_pool(
                 op.add_metadata("node_readiness_time", node_readiness_time)
                 op.add_metadata("command_execution_time", command_execution_time)
 
-                # Log timing - analysis happens in ADX/dashboards
+                # Log timing with bottleneck analysis
+                delta = abs(command_execution_time - node_readiness_time)
+                bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
+                total_elapsed = max(command_execution_time, node_readiness_time)
                 logger.info(
-                    "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
-                    node_pool_name, node_readiness_time, command_execution_time
+                    "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
+                    node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
                 )
 
                 # Verify NVIDIA drivers for managed GPU only (fully managed uses systemd)
@@ -803,10 +804,13 @@ def scale_node_pool(
                 op.add_metadata("node_readiness_time", node_readiness_time)
                 op.add_metadata("command_execution_time", command_execution_time)
 
-                # Log timing - analysis happens in ADX/dashboards
+                # Log timing with bottleneck analysis
+                delta = abs(command_execution_time - node_readiness_time)
+                bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
+                total_elapsed = max(command_execution_time, node_readiness_time)
                 logger.info(
-                    "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
-                    node_pool_name, node_readiness_time, command_execution_time
+                    "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
+                    node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
                 )
 
                 pod_logs = None
@@ -1036,10 +1040,13 @@ def _progressive_scale(
                     op.add_metadata("node_readiness_time", node_readiness_time)
                     op.add_metadata("command_execution_time", command_execution_time)
 
-                    # Log timing - analysis happens in ADX/dashboards
+                    # Log timing with bottleneck analysis
+                    delta = abs(command_execution_time - node_readiness_time)
+                    bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
+                    total_elapsed = max(command_execution_time, node_readiness_time)
                     logger.info(
-                        "[%s] Timing: K8s nodes ready in %.2fs, ARM completed in %.2fs",
-                        node_pool_name, node_readiness_time, command_execution_time
+                        "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
+                        node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
                     )
 
                     if result is None:
diff --git a/modules/python/tests/clients/test_aks_client.py b/modules/python/tests/clients/test_aks_client.py
index 07f70d832d..a26f6cbcf1 100644
--- a/modules/python/tests/clients/test_aks_client.py
+++ b/modules/python/tests/clients/test_aks_client.py
@@ -308,7 +308,7 @@ def test_create_node_pool_fully_managed_gpu(self, mock_time, mock_subprocess_run
         mock_subprocess_run.return_value = mock.MagicMock(returncode=0, stderr="")
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
+        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, 130)
         self.mock_k8s.verify_managed_gpu_systemd_services = mock.MagicMock(return_value={})
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock()
 

From 49b64af50af880cd342a2743e71b9c0082dd526f Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 03:17:07 -0400
Subject: [PATCH 06/12] Add pipeline test config (to be reverted before merge)

---
 pipelines/system/new-pipeline-test.yml | 42 ++++++++++++++++----------
 1 file changed, 26 insertions(+), 16 deletions(-)

diff --git a/pipelines/system/new-pipeline-test.yml b/pipelines/system/new-pipeline-test.yml
index 2f864af31c..d2f4c5dcb6 100644
--- a/pipelines/system/new-pipeline-test.yml
+++ b/pipelines/system/new-pipeline-test.yml
@@ -4,25 +4,35 @@ pool:
   name: 'AKS-Telescope-Airlock'
 
 variables:
-  SCENARIO_TYPE: <scenario-type>
-  SCENARIO_NAME: <scenario-name>
+  SCENARIO_TYPE: perf-eval
+  SCENARIO_NAME: k8s-gpu-cluster-crud
 
 stages:
-  - stage: <stage-name> # format: <cloud>[_<region>]+ (e.g. azure_eastus2, aws_eastus_westus)
+  - stage: azure_australiaeast
     dependsOn: []
     jobs:
-      - template: /jobs/competitive-test.yml # must keep as is
+      - template: /jobs/competitive-test.yml
         parameters:
-          cloud: <cloud> # e.g. azure, aws
-          regions: # list of regions
-            - region1 # e.g. eastus2
-          topology: <topology> # e.g. cluster-autoscaler
-          engine: <engine> # e.g. clusterloader2
-          matrix: # list of test parameters to customize the provisioned resources
-            <case-name>:
-              <key1>: <value1>
-              <key2>: <value2>
-          max_parallel: <number of concurrent jobs> # required
-          credential_type: service_connection # required
+          cloud: azure
+          regions:
+            - australiaeast
+          topology: k8s-crud-gpu
+          engine: crud
+          matrix:
+            node_readiness_timing_test:
+              GPU_NODE_POOL: ""
+              pool_name: nrtiming
+              vm_size: Standard_D4s_v3
+              create_node_count: 2
+              scale_node_count: 3
+              scale_step_size: 1
+              step_time_out: 600
+              step_wait_time: 30
+              count: 1
+              replicas: 1
+              completions: 1
+              manifest_dir: ""
+          max_parallel: 1
+          credential_type: service_connection
           ssh_key_enabled: false
-          timeout_in_minutes: 60 # if not specified, default is 60
+          timeout_in_minutes: 60

From 30a6908d071b4929322dbe934848169ad2e306b1 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 12:09:34 -0400
Subject: [PATCH 07/12] Extract _log_timing_metrics helper and trim
 _instrument_nodepool_provisioning

---
 modules/python/clients/aks_client.py | 68 ++++++++--------------------
 1 file changed, 18 insertions(+), 50 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index a4430dd907..f23a37aa59 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -16,7 +16,7 @@
 import os
 import subprocess
 import time
-from concurrent.futures import ThreadPoolExecutor, wait
+from concurrent.futures import ThreadPoolExecutor
 from typing import Dict, Optional, Any, Tuple
 
 # Third party imports
@@ -72,9 +72,6 @@ def _instrument_nodepool_provisioning(
         """
         Run ARM operation and K8s node readiness check concurrently using threads.
 
-        This allows accurate measurement of both ARM completion time and node readiness time
-        independently, enabling identification of which layer is causing latency.
-
         Args:
             node_pool_name: Name of the node pool being provisioned
             cluster_name: Name of the AKS cluster
@@ -83,13 +80,9 @@ def _instrument_nodepool_provisioning(
 
         Returns:
             Tuple of (arm_result, ready_nodes, node_readiness_time, command_execution_time)
-            - node_readiness_time: seconds from start until K8s nodes were ready
-            - command_execution_time: seconds from start until ARM operation completed
 
         Raises:
-            Exception: If either ARM or K8s readiness fails. Both tasks run to
-                completion (or failure) so we can capture timing for whichever
-                succeeded, enabling better diagnosis of which layer caused the failure.
+            Exception: If either the ARM operation or K8s readiness check fails.
         """
         start_time = time.time()
         label_selector = f"agentpool={node_pool_name}"
@@ -118,11 +111,10 @@ def _wait_k8s():
         with ThreadPoolExecutor(max_workers=2) as executor:
             arm_future = executor.submit(_poll_arm)
             k8s_future = executor.submit(_wait_k8s)
-            wait([arm_future, k8s_future])
 
-        # Check for failures and build diagnostic message
         arm_exc = arm_future.exception()
         k8s_exc = k8s_future.exception()
+        k8s_exc = k8s_future.exception()
 
         if arm_exc or k8s_exc:
             elapsed = time.time() - start_time
@@ -146,6 +138,18 @@ def _wait_k8s():
 
         return arm_response, ready_nodes, node_readiness_time, command_execution_time
 
+    def _log_timing_metrics(self, op, node_pool_name, node_readiness_time, command_execution_time):
+        """Log timing metrics with bottleneck analysis and add metadata to operation."""
+        op.add_metadata("node_readiness_time", node_readiness_time)
+        op.add_metadata("command_execution_time", command_execution_time)
+        delta = abs(command_execution_time - node_readiness_time)
+        bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
+        total_elapsed = max(command_execution_time, node_readiness_time)
+        logger.info(
+            "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
+            node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
+        )
+
     def __init__(
         self,
         subscription_id: Optional[str] = None,
@@ -624,19 +628,7 @@ def create_node_pool(
                 logger.info(
                     f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
-
-                # Add timing metadata for regression analysis
-                op.add_metadata("node_readiness_time", node_readiness_time)
-                op.add_metadata("command_execution_time", command_execution_time)
-
-                # Log timing with bottleneck analysis
-                delta = abs(command_execution_time - node_readiness_time)
-                bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
-                total_elapsed = max(command_execution_time, node_readiness_time)
-                logger.info(
-                    "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
-                    node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
-                )
+                self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
 
                 # Verify NVIDIA drivers for managed GPU only (fully managed uses systemd)
                 pod_logs = None
@@ -799,19 +791,7 @@ def scale_node_pool(
                 logger.info(
                     f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
-
-                # Add timing metadata for regression analysis
-                op.add_metadata("node_readiness_time", node_readiness_time)
-                op.add_metadata("command_execution_time", command_execution_time)
-
-                # Log timing with bottleneck analysis
-                delta = abs(command_execution_time - node_readiness_time)
-                bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
-                total_elapsed = max(command_execution_time, node_readiness_time)
-                logger.info(
-                    "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
-                    node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
-                )
+                self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
 
                 pod_logs = None
                 if gpu_node_pool and not enable_managed_gpu and operation_type == "scale_up" and node_count > 0:
@@ -1035,19 +1015,7 @@ def _progressive_scale(
                         )
 
                     logger.info(f"All {step} nodes in pool {node_pool_name} are ready")
-
-                    # Add timing metadata for regression analysis
-                    op.add_metadata("node_readiness_time", node_readiness_time)
-                    op.add_metadata("command_execution_time", command_execution_time)
-
-                    # Log timing with bottleneck analysis
-                    delta = abs(command_execution_time - node_readiness_time)
-                    bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
-                    total_elapsed = max(command_execution_time, node_readiness_time)
-                    logger.info(
-                        "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
-                        node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
-                    )
+                    self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
 
                     if result is None:
                         logger.error(f"Progressive scaling failed at step {step}")

From 0a1b01074370cd2374cef1678468d494a25bbd46 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 24 Jun 2026 13:06:36 -0400
Subject: [PATCH 08/12] Revert "Add pipeline test config (to be reverted before
 merge)"

This reverts commit 7c896e4ecd6c8ebb1277418017f763395b6ef1b4.
---
 pipelines/system/new-pipeline-test.yml | 42 ++++++++++----------------
 1 file changed, 16 insertions(+), 26 deletions(-)

diff --git a/pipelines/system/new-pipeline-test.yml b/pipelines/system/new-pipeline-test.yml
index d2f4c5dcb6..2f864af31c 100644
--- a/pipelines/system/new-pipeline-test.yml
+++ b/pipelines/system/new-pipeline-test.yml
@@ -4,35 +4,25 @@ pool:
   name: 'AKS-Telescope-Airlock'
 
 variables:
-  SCENARIO_TYPE: perf-eval
-  SCENARIO_NAME: k8s-gpu-cluster-crud
+  SCENARIO_TYPE: <scenario-type>
+  SCENARIO_NAME: <scenario-name>
 
 stages:
-  - stage: azure_australiaeast
+  - stage: <stage-name> # format: <cloud>[_<region>]+ (e.g. azure_eastus2, aws_eastus_westus)
     dependsOn: []
     jobs:
-      - template: /jobs/competitive-test.yml
+      - template: /jobs/competitive-test.yml # must keep as is
         parameters:
-          cloud: azure
-          regions:
-            - australiaeast
-          topology: k8s-crud-gpu
-          engine: crud
-          matrix:
-            node_readiness_timing_test:
-              GPU_NODE_POOL: ""
-              pool_name: nrtiming
-              vm_size: Standard_D4s_v3
-              create_node_count: 2
-              scale_node_count: 3
-              scale_step_size: 1
-              step_time_out: 600
-              step_wait_time: 30
-              count: 1
-              replicas: 1
-              completions: 1
-              manifest_dir: ""
-          max_parallel: 1
-          credential_type: service_connection
+          cloud: <cloud> # e.g. azure, aws
+          regions: # list of regions
+            - region1 # e.g. eastus2
+          topology: <topology> # e.g. cluster-autoscaler
+          engine: <engine> # e.g. clusterloader2
+          matrix: # list of test parameters to customize the provisioned resources
+            <case-name>:
+              <key1>: <value1>
+              <key2>: <value2>
+          max_parallel: <number of concurrent jobs> # required
+          credential_type: service_connection # required
           ssh_key_enabled: false
-          timeout_in_minutes: 60
+          timeout_in_minutes: 60 # if not specified, default is 60

From 666c1f1e34038bfcc48163e50e4b5bac4442ec63 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Wed, 1 Jul 2026 11:36:45 -0400
Subject: [PATCH 09/12] Address review: simplify timing, restore retry, remove
 return_timestamp

---
 modules/python/clients/aks_client.py          | 115 ++++++++----------
 modules/python/clients/kubernetes_client.py   |   7 +-
 .../python/tests/clients/test_aks_client.py   |  63 +++++-----
 .../tests/clients/test_kubernetes_client.py   |  22 ----
 4 files changed, 80 insertions(+), 127 deletions(-)

diff --git a/modules/python/clients/aks_client.py b/modules/python/clients/aks_client.py
index f23a37aa59..b943f8317b 100644
--- a/modules/python/clients/aks_client.py
+++ b/modules/python/clients/aks_client.py
@@ -17,7 +17,7 @@
 import subprocess
 import time
 from concurrent.futures import ThreadPoolExecutor
-from typing import Dict, Optional, Any, Tuple
+from typing import Dict, Optional, Any
 
 # Third party imports
 from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
@@ -68,7 +68,9 @@ def _instrument_nodepool_provisioning(
         cluster_name: str,
         parameters: Any,
         node_count: int,
-    ) -> Tuple[Any, list, float, float]:
+        op,
+        use_retry: bool = False,
+    ) -> list:
         """
         Run ARM operation and K8s node readiness check concurrently using threads.
 
@@ -77,9 +79,11 @@ def _instrument_nodepool_provisioning(
             cluster_name: Name of the AKS cluster
             parameters: Parameters for begin_create_or_update (dict or node pool object)
             node_count: Expected number of nodes to be ready
+            op: Operation context for recording timing metadata
+            use_retry: If True, use _begin_update_with_retry for transient error handling
 
         Returns:
-            Tuple of (arm_result, ready_nodes, node_readiness_time, command_execution_time)
+            List of ready nodes
 
         Raises:
             Exception: If either the ARM operation or K8s readiness check fails.
@@ -87,34 +91,35 @@ def _instrument_nodepool_provisioning(
         start_time = time.time()
         label_selector = f"agentpool={node_pool_name}"
 
-        poller = self.aks_client.agent_pools.begin_create_or_update(
-            resource_group_name=self.resource_group,
-            resource_name=cluster_name,
-            agent_pool_name=node_pool_name,
-            parameters=parameters,
-        )
-
-        def _poll_arm():
-            """Run ARM poller and return (result, completion_timestamp)."""
-            result = poller.result()
-            return result, time.time()
-
-        def _wait_k8s():
-            """Wait for K8s nodes to become ready and return (nodes, timestamp)."""
-            return self.k8s_client.wait_for_nodes_ready(
-                node_count=node_count,
-                operation_timeout_in_minutes=self.operation_timeout_minutes,
-                label_selector=label_selector,
-                return_timestamp=True,
-            )
-
         with ThreadPoolExecutor(max_workers=2) as executor:
-            arm_future = executor.submit(_poll_arm)
-            k8s_future = executor.submit(_wait_k8s)
+            if use_retry:
+                arm_future = executor.submit(
+                    lambda: (
+                        self._begin_update_with_retry(node_pool_name, cluster_name, parameters),
+                        time.time(),
+                    )
+                )
+            else:
+                poller = self.aks_client.agent_pools.begin_create_or_update(
+                    resource_group_name=self.resource_group,
+                    resource_name=cluster_name,
+                    agent_pool_name=node_pool_name,
+                    parameters=parameters,
+                )
+                arm_future = executor.submit(lambda: (poller.result(), time.time()))
+            k8s_future = executor.submit(
+                lambda: (
+                    self.k8s_client.wait_for_nodes_ready(
+                        node_count=node_count,
+                        operation_timeout_in_minutes=self.operation_timeout_minutes,
+                        label_selector=label_selector,
+                    ),
+                    time.time(),
+                )
+            )
 
         arm_exc = arm_future.exception()
         k8s_exc = k8s_future.exception()
-        k8s_exc = k8s_future.exception()
 
         if arm_exc or k8s_exc:
             elapsed = time.time() - start_time
@@ -128,28 +133,22 @@ def _wait_k8s():
                 raise arm_exc
             raise k8s_exc
 
-        # Both succeeded - unpack results
-        arm_response, arm_timestamp = arm_future.result()
+        _, arm_timestamp = arm_future.result()
         ready_nodes, ready_timestamp = k8s_future.result()
 
-        # Calculate times relative to start
         node_readiness_time = ready_timestamp - start_time
         command_execution_time = arm_timestamp - start_time
 
-        return arm_response, ready_nodes, node_readiness_time, command_execution_time
-
-    def _log_timing_metrics(self, op, node_pool_name, node_readiness_time, command_execution_time):
-        """Log timing metrics with bottleneck analysis and add metadata to operation."""
         op.add_metadata("node_readiness_time", node_readiness_time)
         op.add_metadata("command_execution_time", command_execution_time)
-        delta = abs(command_execution_time - node_readiness_time)
-        bottleneck = "ARM" if command_execution_time > node_readiness_time else "K8s"
-        total_elapsed = max(command_execution_time, node_readiness_time)
         logger.info(
-            "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Bottleneck: %s (+%.2fs) | Total elapsed: %.2fs",
-            node_pool_name, command_execution_time, node_readiness_time, bottleneck, delta, total_elapsed
+            "[%s] ARM completed in %.2fs, K8s nodes ready in %.2fs | Delta: %.2fs",
+            node_pool_name, command_execution_time, node_readiness_time,
+            abs(command_execution_time - node_readiness_time)
         )
 
+        return ready_nodes
+
     def __init__(
         self,
         subscription_id: Optional[str] = None,
@@ -600,7 +599,6 @@ def create_node_pool(
                 if enable_managed_gpu:
                     # Fully managed GPU: use az CLI (aks-preview) since the stable SDK
                     # doesn't expose gpuProfile.nvidia.managementMode
-                    start_time = time.time()
                     self.add_managed_gpu_node_pool(
                         node_pool_name=node_pool_name,
                         cluster_name=cluster_name,
@@ -609,26 +607,21 @@ def create_node_pool(
                         gpu_instance_profile=gpu_instance_profile,
                         gpu_mig_strategy=gpu_mig_strategy,
                     )
-                    command_execution_time = time.time() - start_time
                     label_selector = f"agentpool={node_pool_name}"
-                    ready_nodes, ready_timestamp = self.k8s_client.wait_for_nodes_ready(
+                    ready_nodes = self.k8s_client.wait_for_nodes_ready(
                         node_count=node_count,
                         operation_timeout_in_minutes=self.operation_timeout_minutes,
                         label_selector=label_selector,
-                        return_timestamp=True,
                     )
-                    node_readiness_time = ready_timestamp - start_time
                 else:
                     # Run ARM and K8s readiness concurrently to capture both timings
-                    _, ready_nodes, node_readiness_time, command_execution_time = \
-                        self._instrument_nodepool_provisioning(
-                            node_pool_name, cluster_name, parameters, node_count
-                        )
+                    ready_nodes = self._instrument_nodepool_provisioning(
+                        node_pool_name, cluster_name, parameters, node_count, op
+                    )
 
                 logger.info(
                     f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
-                self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
 
                 # Verify NVIDIA drivers for managed GPU only (fully managed uses systemd)
                 pod_logs = None
@@ -783,15 +776,14 @@ def scale_node_pool(
                 )
 
                 # Run ARM and K8s readiness concurrently to capture both timings
-                _, ready_nodes, node_readiness_time, command_execution_time = \
-                    self._instrument_nodepool_provisioning(
-                        node_pool_name, cluster_name, node_pool, node_count
-                    )
+                ready_nodes = self._instrument_nodepool_provisioning(
+                    node_pool_name, cluster_name, node_pool, node_count, op,
+                    use_retry=True,
+                )
 
                 logger.info(
                     f"All {node_count} nodes in pool {node_pool_name} are ready"
                 )
-                self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
 
                 pod_logs = None
                 if gpu_node_pool and not enable_managed_gpu and operation_type == "scale_up" and node_count > 0:
@@ -965,7 +957,6 @@ def _progressive_scale(
 
         logger.info(f"Planned scaling steps: {list(steps)}")
 
-        result = None
         completed_steps = []
 
         # Execute scaling operation for each step
@@ -1009,18 +1000,12 @@ def _progressive_scale(
                     node_pool.count = step  # Update node count in the node pool object
 
                     # Run ARM and K8s readiness concurrently to capture both timings
-                    result, ready_nodes, node_readiness_time, command_execution_time = \
-                        self._instrument_nodepool_provisioning(
-                            node_pool_name, cluster_name, node_pool, step
-                        )
+                    ready_nodes = self._instrument_nodepool_provisioning(
+                        node_pool_name, cluster_name, node_pool, step, op,
+                        use_retry=True,
+                    )
 
                     logger.info(f"All {step} nodes in pool {node_pool_name} are ready")
-                    self._log_timing_metrics(op, node_pool_name, node_readiness_time, command_execution_time)
-
-                    if result is None:
-                        logger.error(f"Progressive scaling failed at step {step}")
-                        op.add_metadata("error", "Scaling operation returned None")
-                        return None
 
                     op.add_metadata(
                         "ready_nodes", len(ready_nodes) if ready_nodes else 0
diff --git a/modules/python/clients/kubernetes_client.py b/modules/python/clients/kubernetes_client.py
index d5b9eafb6f..a564c503f9 100644
--- a/modules/python/clients/kubernetes_client.py
+++ b/modules/python/clients/kubernetes_client.py
@@ -237,7 +237,7 @@ def delete_node(self, node_name):
             else:
                 raise Exception(f"Error deleting Node '{node_name}': {str(e)}") from e
 
-    def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_selector=None, return_timestamp=False):
+    def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_selector=None):
         """
         Waits for a specific number of nodes with a given label to be ready within a specified timeout.
         Raises an exception if the expected number of nodes are not ready within the timeout.
@@ -245,8 +245,7 @@ def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_s
         :param node_count: The expected number of nodes to be ready.
         :param operation_timeout_in_minutes: The timeout in minutes to wait for the nodes to be ready.
         :param label_selector: The label to filter nodes.
-        :param return_timestamp: If True, returns (ready_nodes, ready_timestamp) tuple for timing measurement.
-        :return: List of ready nodes, or (ready_nodes, timestamp) if return_timestamp=True.
+        :return: List of ready nodes.
         """
         ready_nodes = []
         ready_node_count = 0
@@ -257,8 +256,6 @@ def wait_for_nodes_ready(self, node_count, operation_timeout_in_minutes, label_s
             ready_node_count = len(ready_nodes)
             logger.info(f"Currently {ready_node_count} nodes are ready.")
             if ready_node_count == node_count:
-                if return_timestamp:
-                    return ready_nodes, time.time()
                 return ready_nodes
             logger.info(f"Waiting for {node_count} nodes to be ready.")
             time.sleep(10)
diff --git a/modules/python/tests/clients/test_aks_client.py b/modules/python/tests/clients/test_aks_client.py
index a26f6cbcf1..44ae48bc6e 100644
--- a/modules/python/tests/clients/test_aks_client.py
+++ b/modules/python/tests/clients/test_aks_client.py
@@ -176,15 +176,15 @@ def test_create_node_pool_success(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
 
         mock_operation = mock.MagicMock()
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Mock the node pool that will be retrieved after creation
         mock_created_node_pool = mock.MagicMock()
@@ -221,7 +221,6 @@ def mock_get_node_pool_side_effect(name, cluster=None):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
 
         # Verify timing measurements are calculated and stored correctly
@@ -240,15 +239,15 @@ def test_create_node_pool_gpu(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
 
         mock_operation = mock.MagicMock()
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -290,7 +289,6 @@ def mock_get_node_pool_side_effect(name, cluster=None):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
 
         # Check that NVIDIA verification was performed
@@ -308,7 +306,7 @@ def test_create_node_pool_fully_managed_gpu(self, mock_time, mock_subprocess_run
         mock_subprocess_run.return_value = mock.MagicMock(returncode=0, stderr="")
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, 130)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
         self.mock_k8s.verify_managed_gpu_systemd_services = mock.MagicMock(return_value={})
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock()
 
@@ -347,9 +345,10 @@ def test_scale_node_pool_up(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -361,7 +360,7 @@ def test_scale_node_pool_up(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Mock get_cluster_data to return a dictionary for JSON serialization
         self.aks_client.get_cluster_data = mock.MagicMock(
@@ -383,7 +382,6 @@ def test_scale_node_pool_up(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
@@ -402,9 +400,10 @@ def test_scale_node_pool_down(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 3  # Current count
@@ -421,7 +420,7 @@ def test_scale_node_pool_down(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Mock get_node_pool to return the node pool with as_dict method
         self.aks_client.get_node_pool = mock.MagicMock(return_value=mock_node_pool)
@@ -438,7 +437,6 @@ def test_scale_node_pool_down(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
@@ -488,9 +486,10 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -513,7 +512,7 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -534,7 +533,6 @@ def test_scale_gpu_node_pool_up_final_target(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
@@ -552,14 +550,9 @@ def test_scale_gpu_node_pool_up_progressive_final_step(self, mock_time):
         self.aks_client.vm_size = "Standard_NC40ads_H100_v5"
 
         # Define timestamps for clarity (progressive has multiple steps)
-        # Each step calls time.time() twice: once for start, once for arm completion
-        step1_start = 100
-        step1_nodes_ready = 130
-        step1_arm_done = 150
-        step2_start = 200
-        step2_nodes_ready = 230
-        step2_arm_done = 250
-        mock_time.time.side_effect = [step1_start, step1_arm_done, step2_start, step2_arm_done]
+        # Each step calls time.time() 3x: start, arm_done, nodes_ready
+        mock_time.time.side_effect = [100, 150, 130, 200, 250, 230]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1  # Current count
@@ -589,8 +582,8 @@ def test_scale_gpu_node_pool_up_progressive_final_step(self, mock_time):
             mock.MagicMock(),
         ]  # Second step to 3 nodes
         self.mock_k8s.wait_for_nodes_ready.side_effect = [
-            (ready_nodes1, step1_nodes_ready),
-            (ready_nodes2, step2_nodes_ready)
+            ready_nodes1,
+            ready_nodes2,
         ]
 
         # Add nvidia-smi verification mock
@@ -629,9 +622,10 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
 
         # Define timestamps for clarity
         start_time = 100
-        nodes_ready_time = 130
         arm_done_time = 150
-        mock_time.time.side_effect = [start_time, arm_done_time]
+        nodes_ready_time = 130
+        mock_time.time.side_effect = [start_time, arm_done_time, nodes_ready_time]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 3  # Current count
@@ -649,7 +643,7 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
         self.mock_agent_pools.begin_create_or_update.return_value = mock_operation
 
         ready_nodes = [mock.MagicMock()]
-        self.mock_k8s.wait_for_nodes_ready.return_value = (ready_nodes, nodes_ready_time)
+        self.mock_k8s.wait_for_nodes_ready.return_value = ready_nodes
 
         # Add nvidia-smi verification mock
         self.mock_k8s.verify_nvidia_smi_on_node = mock.MagicMock(
@@ -670,7 +664,6 @@ def test_scale_gpu_node_pool_down_no_verification(self, mock_time):
             node_count=node_count,
             operation_timeout_in_minutes=10,
             label_selector=f"agentpool={node_pool_name}",
-            return_timestamp=True,
         )
         self.assertEqual(mock_node_pool.count, node_count)
 
diff --git a/modules/python/tests/clients/test_kubernetes_client.py b/modules/python/tests/clients/test_kubernetes_client.py
index 1467339662..0ab752312e 100644
--- a/modules/python/tests/clients/test_kubernetes_client.py
+++ b/modules/python/tests/clients/test_kubernetes_client.py
@@ -1203,28 +1203,6 @@ def test_wait_for_nodes_ready_exception(self, mock_sleep, mock_get_ready_nodes):
         self.assertIn("Only 1 nodes are ready, expected 2 nodes!", str(context.exception))
         mock_sleep.assert_called()
 
-    @patch('time.time')
-    @patch('clients.kubernetes_client.KubernetesClient.get_ready_nodes')
-    @patch("time.sleep", return_value=None)
-    def test_wait_for_nodes_ready_with_timestamp(self, mock_sleep, mock_get_ready_nodes, mock_time):
-        """Test waiting for nodes with return_timestamp=True returns tuple."""
-        mock_get_ready_nodes.side_effect = [[], ["node1", "node2"]]
-        mock_time.return_value = 1234567890.123  # Fixed timestamp for assertion
-        node_count = 2
-        timeout = 0.01
-
-        result = self.client.wait_for_nodes_ready(
-            node_count, timeout, return_timestamp=True
-        )
-
-        # Should return (ready_nodes, timestamp) tuple
-        self.assertIsInstance(result, tuple)
-        self.assertEqual(len(result), 2)
-        ready_nodes, timestamp = result
-        self.assertEqual(ready_nodes, ["node1", "node2"])
-        self.assertEqual(timestamp, 1234567890.123)
-        mock_sleep.assert_called()
-
     @patch('clients.kubernetes_client.KubernetesClient.get_pods_by_namespace')
     @patch('clients.kubernetes_client.KubernetesClient.get_ready_pods_by_namespace')
     @patch("time.sleep", return_value=None)

From d286aaf92472c1d9c2483aecdc950032244278f9 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Mon, 27 Jul 2026 16:51:15 -0400
Subject: [PATCH 10/12] Fix test: add time.time() side_effect values for
 concurrent timing

---
 modules/python/tests/clients/test_aks_client.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/modules/python/tests/clients/test_aks_client.py b/modules/python/tests/clients/test_aks_client.py
index 44ae48bc6e..c621cae849 100644
--- a/modules/python/tests/clients/test_aks_client.py
+++ b/modules/python/tests/clients/test_aks_client.py
@@ -742,7 +742,8 @@ def test_scale_node_pool_records_gpu_mode_metadata(self, mock_time):
         node_pool_name = "h100fullmgd"
         node_count = 3
 
-        mock_time.time.side_effect = [100, 150]
+        mock_time.time.side_effect = [100, 150, 130]
+        mock_time.sleep = mock.MagicMock()
 
         mock_node_pool = mock.MagicMock()
         mock_node_pool.count = 1

From 94bf515e4e9d8f6d1ac4f376825bf313be9188d9 Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Mon, 27 Jul 2026 17:12:08 -0400
Subject: [PATCH 11/12] Add pipeline test config (to be reverted before merge)

---
 pipelines/system/new-pipeline-test.yml        | 42 ++++++++++++-------
 .../terraform-inputs/azure.tfvars             |  2 +-
 2 files changed, 27 insertions(+), 17 deletions(-)

diff --git a/pipelines/system/new-pipeline-test.yml b/pipelines/system/new-pipeline-test.yml
index 2f864af31c..d2f4c5dcb6 100644
--- a/pipelines/system/new-pipeline-test.yml
+++ b/pipelines/system/new-pipeline-test.yml
@@ -4,25 +4,35 @@ pool:
   name: 'AKS-Telescope-Airlock'
 
 variables:
-  SCENARIO_TYPE: <scenario-type>
-  SCENARIO_NAME: <scenario-name>
+  SCENARIO_TYPE: perf-eval
+  SCENARIO_NAME: k8s-gpu-cluster-crud
 
 stages:
-  - stage: <stage-name> # format: <cloud>[_<region>]+ (e.g. azure_eastus2, aws_eastus_westus)
+  - stage: azure_australiaeast
     dependsOn: []
     jobs:
-      - template: /jobs/competitive-test.yml # must keep as is
+      - template: /jobs/competitive-test.yml
         parameters:
-          cloud: <cloud> # e.g. azure, aws
-          regions: # list of regions
-            - region1 # e.g. eastus2
-          topology: <topology> # e.g. cluster-autoscaler
-          engine: <engine> # e.g. clusterloader2
-          matrix: # list of test parameters to customize the provisioned resources
-            <case-name>:
-              <key1>: <value1>
-              <key2>: <value2>
-          max_parallel: <number of concurrent jobs> # required
-          credential_type: service_connection # required
+          cloud: azure
+          regions:
+            - australiaeast
+          topology: k8s-crud-gpu
+          engine: crud
+          matrix:
+            node_readiness_timing_test:
+              GPU_NODE_POOL: ""
+              pool_name: nrtiming
+              vm_size: Standard_D4s_v3
+              create_node_count: 2
+              scale_node_count: 3
+              scale_step_size: 1
+              step_time_out: 600
+              step_wait_time: 30
+              count: 1
+              replicas: 1
+              completions: 1
+              manifest_dir: ""
+          max_parallel: 1
+          credential_type: service_connection
           ssh_key_enabled: false
-          timeout_in_minutes: 60 # if not specified, default is 60
+          timeout_in_minutes: 60
diff --git a/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars b/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
index d947aaf42e..5cfb259d34 100644
--- a/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
+++ b/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
@@ -8,7 +8,7 @@ aks_cli_config_list = [
     role                          = "gpu"
     aks_name                      = "gpu-cluster"
     sku_tier                      = "standard"
-    kubernetes_version            = "1.33"
+    kubernetes_version            = "1.34"
     use_aks_preview_cli_extension = true
     default_node_pool = {
       name       = "default"

From 1b699f24193336e62f5df67b8fe185218d14c61f Mon Sep 17 00:00:00 2001
From: Diamond Powell <32712461+diamondpowell@users.noreply.github.com>
Date: Tue, 28 Jul 2026 17:20:08 -0400
Subject: [PATCH 12/12] Revert "Add pipeline test config (to be reverted before
 merge)"

This reverts commit 94bf515e4e9d8f6d1ac4f376825bf313be9188d9.
---
 pipelines/system/new-pipeline-test.yml        | 42 +++++++------------
 .../terraform-inputs/azure.tfvars             |  2 +-
 2 files changed, 17 insertions(+), 27 deletions(-)

diff --git a/pipelines/system/new-pipeline-test.yml b/pipelines/system/new-pipeline-test.yml
index d2f4c5dcb6..2f864af31c 100644
--- a/pipelines/system/new-pipeline-test.yml
+++ b/pipelines/system/new-pipeline-test.yml
@@ -4,35 +4,25 @@ pool:
   name: 'AKS-Telescope-Airlock'
 
 variables:
-  SCENARIO_TYPE: perf-eval
-  SCENARIO_NAME: k8s-gpu-cluster-crud
+  SCENARIO_TYPE: <scenario-type>
+  SCENARIO_NAME: <scenario-name>
 
 stages:
-  - stage: azure_australiaeast
+  - stage: <stage-name> # format: <cloud>[_<region>]+ (e.g. azure_eastus2, aws_eastus_westus)
     dependsOn: []
     jobs:
-      - template: /jobs/competitive-test.yml
+      - template: /jobs/competitive-test.yml # must keep as is
         parameters:
-          cloud: azure
-          regions:
-            - australiaeast
-          topology: k8s-crud-gpu
-          engine: crud
-          matrix:
-            node_readiness_timing_test:
-              GPU_NODE_POOL: ""
-              pool_name: nrtiming
-              vm_size: Standard_D4s_v3
-              create_node_count: 2
-              scale_node_count: 3
-              scale_step_size: 1
-              step_time_out: 600
-              step_wait_time: 30
-              count: 1
-              replicas: 1
-              completions: 1
-              manifest_dir: ""
-          max_parallel: 1
-          credential_type: service_connection
+          cloud: <cloud> # e.g. azure, aws
+          regions: # list of regions
+            - region1 # e.g. eastus2
+          topology: <topology> # e.g. cluster-autoscaler
+          engine: <engine> # e.g. clusterloader2
+          matrix: # list of test parameters to customize the provisioned resources
+            <case-name>:
+              <key1>: <value1>
+              <key2>: <value2>
+          max_parallel: <number of concurrent jobs> # required
+          credential_type: service_connection # required
           ssh_key_enabled: false
-          timeout_in_minutes: 60
+          timeout_in_minutes: 60 # if not specified, default is 60
diff --git a/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars b/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
index 5cfb259d34..d947aaf42e 100644
--- a/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
+++ b/scenarios/perf-eval/k8s-gpu-cluster-crud/terraform-inputs/azure.tfvars
@@ -8,7 +8,7 @@ aks_cli_config_list = [
     role                          = "gpu"
     aks_name                      = "gpu-cluster"
     sku_tier                      = "standard"
-    kubernetes_version            = "1.34"
+    kubernetes_version            = "1.33"
     use_aks_preview_cli_extension = true
     default_node_pool = {
       name       = "default"