diff --git a/docs/converter/what-converts.md b/docs/converter/what-converts.md index f7b064ef..fdaeeed2 100644 --- a/docs/converter/what-converts.md +++ b/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 diff --git a/solrorbit/conversion/workload_converter.py b/solrorbit/conversion/workload_converter.py index b99dcb6d..025fbd8b 100644 --- a/solrorbit/conversion/workload_converter.py +++ b/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( diff --git a/tests/unit/solr/test_workload_converter.py b/tests/unit/solr/test_workload_converter.py index 423dc9c7..983a8625 100644 --- a/tests/unit/solr/test_workload_converter.py +++ b/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()