Convert OpenSearch's index-stats merge wait to Solr's wait-for-merges

docs/converter/what-converts.md

@@ -23,6 +23,7 @@ The table below summarizes which OpenSearch Benchmark constructs are automatical
 | `terms` aggregations | Converted to Solr `terms` facet |
 | `date_histogram` aggregations | Converted to Solr `range` facet with calendar gap |
 | `avg` / `sum` / `min` / `max` aggregations | Converted to Solr function query stats |
+| `index-stats` retried until `merges.current` reaches zero | → `wait-for-merges`, with the polling expressed as `retry-wait-period` and `max-wait-seconds` |
 
 ## Requires manual review
 

solrorbit/conversion/workload_converter.py

@@ -209,6 +209,18 @@ def _load_workload_json(workload_path: str) -> dict:
     "delete-pipeline",
 }
 
+# An index-stats operation polling this condition path is a wait-for-merges in Solr
+_MERGE_WAIT_CONDITION_PATH = "merges.current"
+_MERGE_WAIT_RETRY_WAIT_PERIOD = 2.0
+_MERGE_WAIT_MAX_WAIT_SECONDS = 600
+
+_MERGE_WAIT_OBJECT = re.compile(
+    r'\{[^{}]*?"operation-type"\s*:\s*"index-stats"\s*,'
+    r'[^{}]*?"condition"\s*:\s*\{[^{}]*?"path"\s*:\s*"[^"]*'
+    + re.escape(_MERGE_WAIT_CONDITION_PATH)
+    + r'"[^{}]*?\}[^{}]*?\}'
+)
+
 
 def detect_workload_format_from_file(workload_dir: str) -> bool:
     """
@@ -455,7 +467,12 @@ def _process_collected_files(source_dir: str, output_dir: str, issues: list, ski
                 ops_list, tokens = _parse_jinja_fragment(raw, wrap_array=True)
             except ValueError as exc:
                 issues.append(f"{subdir}/{filename}: cannot parse as JSON fragment ({exc}); copied verbatim")
-                shutil.copy2(src_path, dst_path)
+                merges_converted = _convert_merge_wait_text(raw)
+                if merges_converted == raw:
+                    shutil.copy2(src_path, dst_path)
+                else:
+                    with open(dst_path, "w", encoding="utf-8") as f:
+                        f.write(merges_converted)
                 continue
 
             # Convert each operation in the fragment; filter out skipped ones
@@ -552,6 +569,73 @@ def _has_auto_date_histogram(aggs: dict) -> bool:
     return False
 
 
+def _is_merge_wait(op: dict) -> bool:
+    """
+    Return True if *op* is OpenSearch Benchmark's "wait until merges finish" idiom:
+    an index-stats operation retried until the active merge count reaches zero.
+    """
+    condition = op.get("condition")
+    if not isinstance(condition, dict):
+        return False
+    path = condition.get("path")
+    if not isinstance(path, str) or not path.endswith(_MERGE_WAIT_CONDITION_PATH):
+        return False
+    return condition.get("expected-value") == 0
+
+
+def _convert_merge_wait(op: dict) -> None:
+    """
+    Rewrite *op* in-place as a Solr ``wait-for-merges`` operation.
+
+    The polling OpenSearch expresses with ``condition`` and ``retry-until-success`` is
+    the runner's own loop in Solr, so those keys are replaced by its wait parameters.
+    """
+    converted = {}
+    if "name" in op:
+        converted["name"] = op["name"]
+    converted["operation-type"] = "wait-for-merges"
+    converted["retry-wait-period"] = _MERGE_WAIT_RETRY_WAIT_PERIOD
+    converted["max-wait-seconds"] = _MERGE_WAIT_MAX_WAIT_SECONDS
+    if "include-in-reporting" in op:
+        converted["include-in-reporting"] = op["include-in-reporting"]
+
+    replaced = {"name", "operation-type", "type", "index", "indices", "condition",
+                "retry-until-success", "include-in-reporting"}
+    converted.update({key: value for key, value in op.items() if key not in replaced})
+
+    op.clear()
+    op.update(converted)
+
+
+def _convert_merge_wait_text(text: str) -> str:
+    """
+    Rewrite every merge-wait operation object in *text* as ``wait-for-merges``.
+
+    Used for fragments whose Jinja2 directives keep them from being parsed as JSON,
+    where the operation object itself is still plain JSON.
+    """
+    def replace(match):
+        matched = match.group(0)
+        try:
+            op = json.loads(matched)
+        except json.JSONDecodeError:
+            return matched
+        if not isinstance(op, dict) or not _is_merge_wait(op):
+            return matched
+
+        _convert_merge_wait(op)
+        line_start = text.rfind("\n", 0, match.start()) + 1
+        indent = text[line_start:match.start()]
+        indent = indent[:len(indent) - len(indent.lstrip())]
+
+        first_key = re.match(r"\{\s*\n([ \t]*)", matched)
+        step = len(first_key.group(1)) - len(indent) if first_key else 4
+        rendered = json.dumps(op, indent=max(step, 1))
+        return rendered.replace("\n", "\n" + indent)
+
+    return _MERGE_WAIT_OBJECT.sub(replace, text)
+
+
 def _convert_operation(op, issues, skipped, source_dir, output_dir):
     """Convert an operation definition dict in-place.
 
@@ -560,6 +644,10 @@ def _convert_operation(op, issues, skipped, source_dir, output_dir):
     op_type = op.get("operation-type") or op.get("type", "")
     op_name = op.get("name", op_type)
 
+    if op_type == "index-stats" and _is_merge_wait(op):
+        _convert_merge_wait(op)
+        return True
+
     if op_type in _UNSUPPORTED_OPS:
         logger.warning("Skipping unsupported operation '%s' (type: %s)", op_name, op_type)
         skipped.append(op_name)
@@ -752,7 +840,7 @@ def _convert_fragment_text(raw: str, issues: list, skipped: list) -> str:
             return _serialise_jinja_fragment(ops_list, tokens, wrap_array=True)
         except ValueError:
             # Complex Jinja2 — fall back to text substitution for known op-type strings
-            result = raw
+            result = _convert_merge_wait_text(raw)
             for old_op, new_op in _OP_MAP.items():
                 if old_op != new_op:
                     result = re.sub(

tests/unit/solr/test_workload_converter.py

@@ -24,6 +24,9 @@
 
 from solrorbit.conversion.workload_converter import (
     CONVERTED_MARKER,
+    _convert_merge_wait,
+    _convert_merge_wait_text,
+    _convert_operation,
     convert_opensearch_workload,
     detect_workload_format_from_file,
     is_already_converted,
@@ -353,5 +356,166 @@ def test_case_insensitive(self):
         self.assertEqual("+1MONTH", _calendar_interval_to_solr_gap("MONTH"))
 
 
+class TestMergeWaitConversion(unittest.TestCase):
+    """Tests for OpenSearch's index-stats merge wait becoming Solr's wait-for-merges."""
+
+    def _osb_merge_wait(self):
+        return {
+            "name": "wait-until-merges-finish",
+            "operation-type": "index-stats",
+            "index": "_all",
+            "condition": {"path": "_all.total.merges.current", "expected-value": 0},
+            "retry-until-success": True,
+            "include-in-reporting": False,
+        }
+
+    def test_converted_operation_is_the_solr_operation(self):
+        op = self._osb_merge_wait()
+        _convert_merge_wait(op)
+        self.assertEqual({
+            "name": "wait-until-merges-finish",
+            "operation-type": "wait-for-merges",
+            "retry-wait-period": 2.0,
+            "max-wait-seconds": 600,
+            "include-in-reporting": False,
+        }, op)
+
+    def test_convert_operation_leaves_other_index_stats_alone(self):
+        op = {"name": "index-stats", "operation-type": "index-stats"}
+        _convert_operation(op, [], [], "", "")
+        self.assertEqual("index-stats", op["operation-type"])
+
+    def test_convert_operation_leaves_another_condition_path_alone(self):
+        op = self._osb_merge_wait()
+        op["condition"]["path"] = "_all.total.docs.count"
+        _convert_operation(op, [], [], "", "")
+        self.assertEqual("index-stats", op["operation-type"])
+
+    def test_text_conversion_of_a_jinja_fragment(self):
+        raw = """{
+    "operation": {
+        "operation-type": "force-merge",
+        "request-timeout": {{ request_timeout | default(60) | tojson }}
+    }
+},
+{
+    "name": "wait-until-merges-finish",
+    "operation": {
+        "operation-type": "index-stats",
+        "index": "_all",
+        "condition": {
+            "path": "_all.total.merges.current",
+            "expected-value": 0
+        },
+        "retry-until-success": true,
+        "include-in-reporting": false
+    }
+}"""
+        converted = _convert_merge_wait_text(raw)
+        self.assertNotIn("index-stats", converted)
+        self.assertIn('"operation-type": "wait-for-merges"', converted)
+        self.assertIn('"retry-wait-period": 2.0', converted)
+        self.assertIn('"max-wait-seconds": 600', converted)
+        self.assertIn("{{ request_timeout | default(60) | tojson }}", converted)
+
+    def test_text_conversion_follows_the_fragment_indentation(self):
+        template = """{{
+{i}"name": "wait-until-merges-finish",
+{i}"operation": {{
+{i}{i}"operation-type": "index-stats",
+{i}{i}"index": "_all",
+{i}{i}"condition": {{
+{i}{i}{i}"path": "_all.total.merges.current",
+{i}{i}{i}"expected-value": 0
+{i}{i}}},
+{i}{i}"include-in-reporting": false
+{i}}}
+}}"""
+        for indent in ("    ", "  "):
+            with self.subTest(indent=len(indent)):
+                converted = _convert_merge_wait_text(template.format(i=indent))
+                self.assertIn(f'\n{indent * 2}"operation-type": "wait-for-merges",', converted)
+                self.assertIn(f"\n{indent}}}\n}}", converted)
+
+    def test_text_conversion_leaves_another_condition_alone(self):
+        raw = """{
+    "operation": {
+        "operation-type": "index-stats",
+        "index": "_all",
+        "condition": {
+            "path": "_all.total.docs.count",
+            "expected-value": 0
+        }
+    }
+}"""
+        self.assertEqual(raw, _convert_merge_wait_text(raw))
+
+    def test_conversion_of_an_external_fragment_end_to_end(self):
+        """The merge wait reaches the output even from a fragment Jinja2 keeps unparseable."""
+        fragment = """{
+    "operation": {
+        "operation-type": "force-merge",
+        "request-timeout": {{ request_timeout | default(60) | tojson }}{%- if max_num_segments is defined %},
+        "max-num-segments": {{ max_num_segments | tojson }}
+        {%- endif %}
+    }
+},
+{
+    "name": "wait-until-merges-finish",
+    "operation": {
+        "operation-type": "index-stats",
+        "index": "_all",
+        "condition": {
+            "path": "_all.total.merges.current",
+            "expected-value": 0
+        },
+        "retry-until-success": true,
+        "include-in-reporting": false
+    }
+}"""
+        with tempfile.TemporaryDirectory() as tmpdir, tempfile.TemporaryDirectory() as dst:
+            shared = os.path.join(tmpdir, "common_operations")
+            os.makedirs(shared)
+            with open(os.path.join(shared, "force_merge.json"), "w") as f:
+                f.write(fragment)
+
+            src = os.path.join(tmpdir, "my_workload")
+            os.makedirs(os.path.join(src, "test_procedures"))
+            with open(os.path.join(src, "workload.json"), "w") as f:
+                json.dump({"indices": [], "challenges": []}, f)
+            with open(os.path.join(src, "test_procedures", "default.json"), "w") as f:
+                f.write('{\n  "name": "default",\n  "schedule": [\n'
+                        '    {{ benchmark.collect(parts="../../common_operations/force_merge.json") }}\n'
+                        "  ]\n}")
+
+            convert_opensearch_workload(src, dst)
+
+            with open(os.path.join(dst, "common_operations", "force_merge.json")) as f:
+                out = f.read()
+            self.assertNotIn("index-stats", out)
+            self.assertIn('"operation-type": "wait-for-merges"', out)
+            self.assertIn('"max-wait-seconds": 600', out)
+
+    def test_conversion_end_to_end(self):
+        with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as dst:
+            with open(os.path.join(src, "workload.json"), "w") as f:
+                json.dump({
+                    "indices": [],
+                    "challenges": [
+                        {
+                            "name": "default",
+                            "schedule": [{"operation": self._osb_merge_wait()}],
+                        }
+                    ],
+                }, f)
+            convert_opensearch_workload(src, dst)
+            with open(os.path.join(dst, "workload.json")) as f:
+                out = json.load(f)
+            op = out["challenges"][0]["schedule"][0]["operation"]
+            self.assertEqual("wait-for-merges", op["operation-type"])
+            self.assertEqual(600, op["max-wait-seconds"])
+            self.assertNotIn("condition", op)
+
+
 if __name__ == "__main__":
     unittest.main()