@@ -65,8 +65,10 @@ def get_all_timeseries(self) -> Dict[frozenset, TimeSeries]:
 
         for k in keys:
             for repetition_idx, result in enumerate(self.query_results):
+                # A key absent from one repetition (e.g. a series that appears
+                # mid-run) stays None for that repetition.
                 if result.result:
-                    intermediate_ret[k][repetition_idx] = result.result[k]
+                    intermediate_ret[k][repetition_idx] = result.result.get(k)
 
             ret[k] = TimeSeries(k, intermediate_ret[k])
 
 
@@ -13,6 +13,10 @@
 from experiment_utils.providers.base import InfrastructureProvider
 
 
+# Experiment-config data_type -> the exporter's --data-type value.
+ALIBABA_DATA_TYPE_CLI_VALUES = {"node": "node", "msresource": "ms-resource"}
+
+
 class ClusterDataExporterService(BaseService):
     """
     Service for managing cluster_data_exporter via Docker.
@@ -71,10 +75,11 @@ def start(
         # Get number of nodes from provider (assuming it has this info)
         num_nodes = kwargs.get("num_nodes", 1)
 
-        # Assert that we have exactly 2 nodes
-        assert num_nodes == 1, (
-            f"cluster_data_exporter requires exactly 1 node (num_nodes==1), "
-            f"got {num_nodes}"
+        # One worker node next to the coordinator; local mode reports 0 nodes
+        # and runs everything on this machine.
+        assert num_nodes in (0, 1), (
+            f"cluster_data_exporter requires one worker node (num_nodes==1) "
+            f"or local mode (num_nodes==0), got {num_nodes}"
         )
 
         # Extract configuration
@@ -258,7 +263,8 @@ def _build_docker_command(
 
         elif provider == "alibaba":
             if "data_type" in config:
-                cmd_parts.append(f"--data-type={config['data_type']}")
+                data_type = ALIBABA_DATA_TYPE_CLI_VALUES[config["data_type"]]
+                cmd_parts.append(f"--data-type={data_type}")
             if "data_year" in config:
                 cmd_parts.append(f"--data-year={config['data_year']}")
 
@@ -400,23 +406,18 @@ def _validate_alibaba_data(
         Raises:
             ValueError: If required files are missing
         """
-        # Determine expected file pattern based on data type and year
-        if data_type == "node":
-            if data_year == 2021 or data_year == 2022:
-                pattern = "Node_*.csv.gz"
-            else:
-                raise ValueError(
-                    f"Invalid data_year for Alibaba node data: {data_year}"
-                )
-        elif data_type == "msresource":
-            if data_year == 2021 or data_year == 2022:
-                pattern = "MsResource_*.csv.gz"
-            else:
-                raise ValueError(
-                    f"Invalid data_year for Alibaba msresource data: {data_year}"
-                )
-        else:
-            raise ValueError(f"Invalid data_type for Alibaba: {data_type}")
+        # File names the exporter reads (alibaba_metrics/{node,ms_resource}.rs).
+        patterns = {
+            ("node", 2021): "Node_*.csv.gz",
+            ("node", 2022): "NodeMetrics_*.csv.gz",
+            ("msresource", 2021): "MSResource_*.csv.gz",
+            ("msresource", 2022): "MSMetrics_*.csv.gz",
+        }
+        pattern = patterns.get((data_type, int(data_year)))
+        if pattern is None:
+            raise ValueError(
+                f"Invalid Alibaba data_type/data_year: {data_type}/{data_year}"
+            )
 
         # Check for data files on remote node
         target_node = self.node_offset + 1
@@ -497,12 +498,15 @@ def _wait_for_health(
         while time.time() - start_time < timeout:
             # Run curl from the remote node to check health
             check_cmd = f"curl -s -o /dev/null -w '%{{http_code}}' {url}"
+            # curl exits non-zero while the exporter is still starting; the
+            # local provider raises on that, so keep polling instead.
             result = self.provider.execute_command(
                 node_idx=node_idx,
                 cmd=check_cmd,
                 cmd_dir="",
                 nohup=False,
                 popen=False,
+                ignore_errors=True,
             )
 
             try:
 
@@ -7,7 +7,7 @@
 import shlex
 import time
 import subprocess
-from typing import Optional
+from typing import List, Optional
 
 import constants
 from .base import BaseService
@@ -100,13 +100,15 @@ def start(
                 else:
                     keywords.append(constants.QUERY_ENGINE_RS_PROCESS_KEYWORD)
 
+        keywords_arg = self._keywords_arg(keywords)
+
         if use_timed_mode:
             # Build command for timed mode (skip_querying)
             cmd = (
                 "python3 -u remote_monitor.py "
                 "--execution_mode timed "
                 "--experiment_mode {} "
-                r"--keywords \"{}\" "
+                "--keywords {} "
                 "--config_file {} "
                 "--experiment_output_dir {} "
                 "--monitor_output_file {} "
@@ -116,7 +118,7 @@ def start(
                 "--monitor_interval_seconds {} "
             ).format(
                 experiment_mode,
-                ",".join(keywords),
+                keywords_arg,
                 os.path.join(
                     os.path.dirname(experiment_output_dir),
                     "controller_client_configs",
@@ -162,7 +164,7 @@ def start(
             "python3 -u remote_monitor.py "
             "--execution_mode prometheus_client "
             "--experiment_mode {} "
-            r"--keywords \"{}\" "
+            "--keywords {} "
             "--config_file {} "
             "--experiment_output_dir {} "
             "--monitor_output_file {} "
@@ -172,7 +174,7 @@ def start(
             "--monitor_interval_seconds {} "
         ).format(
             experiment_mode,
-            ",".join(keywords),
+            keywords_arg,
             os.path.join(
                 os.path.dirname(experiment_output_dir),
                 "controller_client_configs",
@@ -261,7 +263,7 @@ def start_clickhouse_ingest_monitor(
             "python3 -u remote_monitor.py "
             "--execution_mode ingest "
             "--experiment_mode {} "
-            r"--keywords \"{}\" "
+            "--keywords {} "
             "--config_file {} "
             "--experiment_output_dir {} "
             "--monitor_output_file {} "
@@ -271,7 +273,7 @@ def start_clickhouse_ingest_monitor(
             "--backend_protocol clickhouse "
         ).format(
             constants.BASELINE_EXPERIMENT_NAME,
-            ",".join(keywords),
+            self._keywords_arg(keywords),
             controller_client_config,
             experiment_output_dir,
             monitor_output_file,
@@ -313,6 +315,13 @@ def start_clickhouse_ingest_monitor(
             popen=False,
         )
 
+    def _keywords_arg(self, keywords: List[str]) -> str:
+        # Over SSH the command is parsed by two shells, so the quotes that keep
+        # the keyword list one argument must be escaped once more.
+        if self.provider.is_remote():
+            return r"\"{}\"".format(",".join(keywords))
+        return '"{}"'.format(",".join(keywords))
+
     @staticmethod
     def _remote_monitor_pgrep_pattern(
         execution_mode: Optional[str] = None,
 
@@ -0,0 +1,70 @@
+"""Tests for ClusterDataExporterService data-file validation and node checks."""
+
+import subprocess
+import unittest
+
+from experiment_utils.services.cluster_data_exporter import ClusterDataExporterService
+
+
+class RecordingProvider:
+    """Answers every command with one matching file and records the commands."""
+
+    def __init__(self):
+        self.commands = []
+
+    def execute_command(self, node_idx, cmd, cmd_dir, nohup, popen):
+        self.commands.append(cmd)
+        return subprocess.CompletedProcess([], 0, "1\n", "")
+
+
+class ValidateAlibabaDataTest(unittest.TestCase):
+    def _counted_pattern(self, data_type, data_year):
+        provider = RecordingProvider()
+        service = ClusterDataExporterService(provider, 0, "/traces")
+        service._validate_alibaba_data(data_type, data_year)
+        return [c for c in provider.commands if "wc -l" in c][0]
+
+    def test_patterns_match_exporter_file_names(self):
+        # The check used to look for MsResource_*.csv.gz for both years, a
+        # name the exporter never reads, so valid data was rejected.
+        cases = {
+            ("node", 2021): "Node_*.csv.gz",
+            ("node", 2022): "NodeMetrics_*.csv.gz",
+            ("msresource", 2021): "MSResource_*.csv.gz",
+            ("msresource", 2022): "MSMetrics_*.csv.gz",
+        }
+        for (data_type, data_year), pattern in cases.items():
+            with self.subTest(data_type=data_type, data_year=data_year):
+                self.assertIn(
+                    f"/traces/{pattern}", self._counted_pattern(data_type, data_year)
+                )
+
+    def test_unknown_year_raises(self):
+        service = ClusterDataExporterService(RecordingProvider(), 0, "/traces")
+        with self.assertRaises(ValueError):
+            service._validate_alibaba_data("msresource", 2020)
+
+
+class DockerCommandTest(unittest.TestCase):
+    def test_msresource_uses_exporter_cli_value(self):
+        # The exporter's clap enum spells it ms-resource; passing the config
+        # value through made the container exit before serving metrics.
+        service = ClusterDataExporterService(RecordingProvider(), 0, "/traces")
+        service.container_name = "cde"
+        cmd = service._build_docker_command(
+            {"provider": "alibaba", "data_type": "msresource", "data_year": 2022},
+            port=40000,
+            output_dir="/out",
+        )
+        self.assertIn("--data-type=ms-resource", cmd)
+
+
+class NodeCountTest(unittest.TestCase):
+    def test_more_than_one_worker_node_is_rejected(self):
+        service = ClusterDataExporterService(RecordingProvider(), 0, "/traces")
+        with self.assertRaises(AssertionError):
+            service.start({"provider": "google"}, "/out", "/local", num_nodes=2)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,84 @@
+"""Tests that remote_monitor.py receives its keyword list as one clean argument."""
+
+import shlex
+import unittest
+
+from experiment_utils.services.remote_monitor_service import RemoteMonitorService
+
+
+class RecordingProvider:
+    def __init__(self, remote):
+        self.remote = remote
+        self.commands = []
+
+    def is_remote(self):
+        return self.remote
+
+    def get_home_dir(self):
+        return "/home"
+
+    def execute_command(self, **kwargs):
+        self.commands.append(kwargs["cmd"])
+
+
+def keywords_seen_by_monitor(remote):
+    provider = RecordingProvider(remote)
+    RemoteMonitorService(provider, 0).start(
+        controller_client_config="/out/controller_client_configs/baseline.yaml",
+        experiment_output_dir="/out/baseline",
+        experiment_mode="baseline",
+        profile_query_engine=False,
+        profile_prometheus_time=None,
+        manual_mode=False,
+        streaming_engine="precompute",
+        query_engine_service=None,
+        controller_remote_output_dir="/out/controller_output",
+        use_container_prometheus_client=True,
+        prometheus_client_parallel=False,
+        backend_protocol="prometheus",
+        pre_query_wait_seconds=0,
+        monitor_interval_seconds=1.0,
+        timed_duration=10,
+    )
+    args = shlex.split(provider.commands[0])
+    if remote:
+        # The SSH provider wraps the command in one more shell.
+        args = shlex.split(" ".join(args))
+    return args[args.index("--keywords") + 1]
+
+
+def clickhouse_ingest_keywords_seen_by_monitor(remote):
+    provider = RecordingProvider(remote)
+    RemoteMonitorService(provider, 0).start_clickhouse_ingest_monitor(
+        controller_client_config="/out/controller_client_configs/baseline.yaml",
+        experiment_output_dir="/out/baseline",
+    )
+    # The first command clears the stop file; the last launches the monitor.
+    args = shlex.split(provider.commands[-1])
+    if remote:
+        args = shlex.split(" ".join(args))
+    return args[args.index("--keywords") + 1]
+
+
+class KeywordQuotingTest(unittest.TestCase):
+    def test_local_provider_gets_unescaped_keywords(self):
+        # Escaped quotes meant for SSH used to reach remote_monitor.py
+        # verbatim in local mode, so no process matched and no queries ran.
+        self.assertEqual(keywords_seen_by_monitor(remote=False), "prometheus.yml")
+
+    def test_remote_provider_keeps_escaped_quotes_for_ssh(self):
+        self.assertEqual(keywords_seen_by_monitor(remote=True), "prometheus.yml")
+
+    def test_clickhouse_ingest_local_provider_gets_unescaped_keywords(self):
+        self.assertEqual(
+            clickhouse_ingest_keywords_seen_by_monitor(remote=False), "clickhouse"
+        )
+
+    def test_clickhouse_ingest_remote_provider_keeps_escaped_quotes_for_ssh(self):
+        self.assertEqual(
+            clickhouse_ingest_keywords_seen_by_monitor(remote=True), "clickhouse"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()