diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 54c84d0..707579a 100644 --- a/.github/workflows/python.yml +++ b/.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' diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index bbd3f94..ade2eae 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/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 diff --git a/asap-tools/experiments/tests/test_fake_exporters.py b/asap-tools/experiments/tests/test_fake_exporters.py new file mode 100644 index 0000000..0285138 --- /dev/null +++ b/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()