fix(asap-tools): serialize fake exporter container batches (#686)

* fix(asap-tools): serialize fake exporter container batches

* test(ci): run experiment utility tests

Author: milindsrivastava1997

.github/workflows/python.yml

@@ -125,6 +125,24 @@ jobs:
         # Exit-zero treats all errors as warnings
         flake8 . --config=../.flake8 --count --exit-zero --max-complexity=10 --statistics
 
+  test-experiment-utilities:
+    needs: detect-changes
+    if: needs.detect-changes.outputs.utilities == 'true'
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.11'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install Jinja2==3.1.2 PyYAML==6.0.2 omegaconf==2.3.0
+    - name: Run experiment tests
+      working-directory: asap-tools/experiments
+      run: python -m unittest discover -s tests -p 'test_*.py' -v
+
   test-prometheus-exporters:
     needs: detect-changes
     if: needs.detect-changes.outputs.prometheus_exporters == 'true'

asap-tools/experiments/experiment_utils/services/fake_exporters.py

@@ -531,29 +531,31 @@ def _start_containerized(
 
         # Start containers in batches to avoid overwhelming Docker daemon.
         BATCH_SIZE = 5
-        node_batch_commands: List[Tuple[int, str]] = []
-        for node_idx in target_nodes:
-            node_commands = [
-                cmd
-                for command_node_idx, cmd in docker_run_cmds
-                if command_node_idx == node_idx
-            ]
-            for i in range(0, len(node_commands), BATCH_SIZE):
-                batch = node_commands[i : i + BATCH_SIZE]
-                # Combine docker run commands in batch for one node.
-                batch_cmd = "; ".join(batch)
-                node_batch_commands.append((node_idx, batch_cmd))
-
-        # Start each worker's batches concurrently.
-        execute_fake_exporter_commands_in_parallel(
-            self.provider,
-            node_batch_commands,
-            cmd_dir="",
-            nohup=False,
-            popen=True,
-            redirect=True,
-            wait=True,  # Wait for batch to complete
-        )
+        node_commands_by_node: Dict[int, List[str]] = {
+            node_idx: [] for node_idx in target_nodes
+        }
+        for node_idx, command in docker_run_cmds:
+            node_commands_by_node[node_idx].append(command)
+
+        # Run one batch round at a time: nodes run concurrently within a round,
+        # while batches for the same node remain sequential.
+        for batch_start in range(0, num_ports, BATCH_SIZE):
+            node_batch_commands: List[Tuple[int, str]] = []
+            batch_end = batch_start + BATCH_SIZE
+            for node_idx in target_nodes:
+                batch = node_commands_by_node[node_idx][batch_start:batch_end]
+                if batch:
+                    node_batch_commands.append((node_idx, "; ".join(batch)))
+
+            execute_fake_exporter_commands_in_parallel(
+                self.provider,
+                node_batch_commands,
+                cmd_dir="",
+                nohup=False,
+                popen=True,
+                redirect=True,
+                wait=True,  # Wait for this batch round before starting the next.
+            )
 
         return
 

asap-tools/experiments/tests/test_fake_exporters.py

@@ -0,0 +1,78 @@
+"""Regression tests for fake exporter service orchestration."""
+
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from experiment_utils.services.fake_exporters import RustExporterService
+
+
+class FakeExporterArgs:
+    def get_node_range(self, include_coordinator=True):
+        return [1, 2] if not include_coordinator else [0, 1, 2]
+
+    def get_coordinator_node(self):
+        return 0
+
+
+class FakeExporterProvider:
+    def get_home_dir(self):
+        return "/tmp"
+
+
+class RustExporterContainerBatchingTest(unittest.TestCase):
+    def test_container_batches_are_sequential_per_node(self):
+        """Keep Docker startup bounded per node while retaining cross-node fan-out."""
+        service = RustExporterService(
+            provider=FakeExporterProvider(),
+            args=FakeExporterArgs(),
+            use_container=True,
+        )
+        config = {
+            "num_ports_per_server": 8,
+            "dataset": "demo",
+            "start_port": 50000,
+            "synthetic_data_value_scale": 1,
+            "num_labels": 1,
+            "num_values_per_label": 2,
+            "metric_type": "gauge",
+        }
+        rounds = []
+
+        def record_round(_provider, node_commands, **kwargs):
+            rounds.append((list(node_commands), kwargs))
+
+        # Regression coverage for Roborev 139: submitting all per-node batches
+        # together allows multiple batches to overlap on the same Docker daemon.
+        with patch(
+            "experiment_utils.services.fake_exporters."
+            "execute_fake_exporter_commands_in_parallel",
+            side_effect=record_round,
+        ):
+            with tempfile.TemporaryDirectory() as local_experiment_dir:
+                service._start_containerized(
+                    config=config,
+                    experiment_output_dir="",
+                    local_experiment_dir=local_experiment_dir,
+                )
+
+        self.assertEqual(len(rounds), 2)
+        self.assertEqual(
+            [
+                [node_idx for node_idx, _ in node_commands]
+                for node_commands, _ in rounds
+            ],
+            [[1, 2], [1, 2]],
+        )
+        self.assertEqual(
+            [
+                [command.count("docker run") for _, command in node_commands]
+                for node_commands, _ in rounds
+            ],
+            [[5, 5], [3, 3]],
+        )
+        self.assertTrue(all(kwargs["wait"] for _, kwargs in rounds))
+
+
+if __name__ == "__main__":
+    unittest.main()