From 8e3d51886db9b5732ef14dcad4caf9e63b7edd90 Mon Sep 17 00:00:00 2001 From: Santanu Sinha Date: Tue, 29 Sep 2026 08:19:29 +0530 Subject: [PATCH 1/4] Added support for placement simulation --- plugins/cluster.py | 61 +++++++++++++++++++++- plugins/tasks.py | 17 +++---- tests/mock_server.py | 54 ++++++++++++++++++++ tests/test_cluster.py | 33 +++++++++++- tests/test_offline_cluster.py | 90 +++++++++++++++++++++++++++++++++ tests/test_offline_full_help.py | 4 +- 6 files changed, 246 insertions(+), 13 deletions(-) diff --git a/plugins/cluster.py b/plugins/cluster.py index 36e5132..9b4fdbe 100644 --- a/plugins/cluster.py +++ b/plugins/cluster.py @@ -41,6 +41,14 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg sub_parser.add_argument("--textfmt", "-s", help="Use the format string to print message", type=str, default="{type: <25} | {id: <36} | {time: <20} | {metadata}") sub_parser.set_defaults(func=self.handle_events) + simulation_parser = commands.add_parser("simulate-placement", help="Simulate placement for a deployable (app, task or local service spec)") + simulation_parser.add_argument("spec_file", help="Deployment spec file (application, task or local service JSON)") + simulation_parser.add_argument("--num-instances", "-n", help="Number of instances to simulate placement for. Default is 1", default=1, type=int) + simulation_parser.add_argument("--placement-policy", help="JSON file with a placement policy to override the spec policy") + simulation_parser.add_argument("--detail", help="Show per-instance placement details", default=False, action="store_true") + simulation_parser.add_argument("--json", help="Print raw JSON response", default=False, action="store_true") + simulation_parser.set_defaults(func=self.simulate_placement) + maintenance_parser = commands.add_parser("maintenance-on", help="Set cluster to maintenance mode") maintenance_parser.set_defaults(func=self.set_maintenance) @@ -122,6 +130,57 @@ def unset_maintenance(self, options: SimpleNamespace): except Exception as e: print("Error setting drove cluster to normal mode: " + str(e)) + def simulate_placement(self, options: SimpleNamespace): + try: + with open(options.spec_file, 'r') as fp: + spec = json.load(fp) + placement_policy = None + if options.placement_policy: + with open(options.placement_policy, 'r') as fp: + placement_policy = json.load(fp) + request_body = { + "spec": spec, + "numInstances": options.num_instances + } + if placement_policy is not None: + request_body["placementPolicy"] = placement_policy + response = self.drove_client.post("/apis/v1/cluster/placement/simulate", + body=request_body, + params={"detail": "true" if options.detail else "false"}) + if options.json: + droveutils.print_json(response) + return + placements = response.get("placements", []) + errors = response.get("errors", []) + print("Placement simulation for {spec}".format(spec=options.spec_file)) + print("Requested instances: {requested}".format(requested=response.get("requested", 0))) + print("Placed instances: {placed}".format(placed=response.get("placed", 0))) + print("Errors: {errors}".format(errors=len(errors))) + for error in errors: + print(" - {error}".format(error=error)) + if options.detail and len(placements) > 0: + rows = [] + for placement in placements: + rows.append([ + placement.get("executorId", ""), + placement.get("hostname", ""), + placement.get("port", ""), + placement.get("executorState", "") + ]) + droveutils.print_table(["Executor Id", "Hostname", "Port", "State"], rows) + except (OSError, IOError) as e: + print("Error reading simulation input: " + str(e)) + except droveclient.DroveException as e: + errors = [] + if e.api_response is not None: + errors = e.api_response.get("data", []).get("errors", []) \ + if isinstance(e.api_response.get("data", []), dict) else [] + print("Placement simulation failed: {message}".format(message=str(e))) + for error in errors: + print(" - {error}".format(error=error)) + except Exception as e: + print("Error simulating placement: " + str(e)) + def convert_event(self, format: str, event: dict) -> str: data = dict() data["type"] = event["type"] @@ -129,4 +188,4 @@ def convert_event(self, format: str, event: dict) -> str: data["time"] = droveutils.to_date(event["time"]) data["metadata"] = json.dumps(event["metadata"]) - return format.format_map(data) \ No newline at end of file + return format.format_map(data) diff --git a/plugins/tasks.py b/plugins/tasks.py index 86e9f4b..bd56c9b 100644 --- a/plugins/tasks.py +++ b/plugins/tasks.py @@ -124,15 +124,14 @@ def list_task(self, options: SimpleNamespace): def show_task(self, options: SimpleNamespace): raw = self.drove_client.get("/apis/v1/tasks/{source_app}/instances/{task_id}".format(source_app = options.source_app, task_id = options.task_id)) data = OrderedDict() - data["Source App"] = raw["sourceAppName"] - data["Task ID"] = raw["taskId"] - data["Instance ID"] = raw["instanceId"] - data["State"] = raw["state"] - data["Executor Host"] = raw["hostname"] + data["Source App"] = raw.get("sourceAppName", "") + data["Task ID"] = raw.get("taskId", "") + data["Instance ID"] = raw.get("instanceId", "") + data["State"] = raw.get("state", "") + data["Executor Host"] = raw.get("hostname", "") droveutils.populate_resources(raw, data) - data["Executable"] = "{url} ({type})".format_map(raw["executable"]) - data["Volumes"] = ", ".join(["{pathOnHost} (Mounted as {pathInContainer} in {mode} mode)".format_map(v) for v in raw["volumes"]]) - data["Logging Mode"] = raw.get("logging", dict()).get("type", "") + data["Executable"] = "{url} ({type})".format_map(raw.get("executable", {})) + data["Volumes"] = ", ".join(["{pathOnHost} (Mounted as {pathInContainer} in {mode} mode)".format_map(v) for v in raw.get("volumes", [])]) data["Metadata"] = ", ".join(["%s: %s" % (key,value) for (key, value) in raw.get("metadata", dict())]) result = raw.get("taskResult", dict()) if len(result) > 0: @@ -155,4 +154,4 @@ def log_download(self, options): filename = options.tasklogfile if options.out and len(options.out) > 0: filename = options.out - droveutils.download_log(self.drove_client, "tasks", options.source_app, options.task_id, options.tasklogfile, filename) \ No newline at end of file + droveutils.download_log(self.drove_client, "tasks", options.source_app, options.task_id, options.tasklogfile, filename) diff --git a/tests/mock_server.py b/tests/mock_server.py index df48abe..7850d33 100644 --- a/tests/mock_server.py +++ b/tests/mock_server.py @@ -638,6 +638,60 @@ def executor_blacklist(): def executor_unblacklist(): return ok({"successful": [], "failed": []}) + @app.route("/apis/v1/cluster/placement/simulate", methods=["POST"]) + def placement_simulate(): + body = request.get_json(force=True, silent=True) + if not isinstance(body, dict): + return jsonify({"status": "FAILED", + "data": {"validationErrors": ["Invalid request body"]}, + "message": "JSON validation failure"}), 400 + spec = body.get("spec") or {} + if "type" not in spec: + # Matches the real Drove error for a spec without a type discriminator + return jsonify({"status": "FAILED", + "data": {"validationErrors": + ["Could not resolve subtype of [simple type, class " + "com.phonepe.drove.models.interfaces.DeploymentSpec]: " + "missing type id property 'type' (for POJO property 'spec')"]}, + "message": "JSON validation failure"}), 400 + num_instances = body.get("numInstances", 1) + if not isinstance(num_instances, int) or num_instances < 1 or num_instances > 2048: + return jsonify({"status": "FAILED", + "data": {"validationErrors": + ["numInstances must be between 1 and 2048"]}, + "message": "JSON validation failure"}), 400 + resources = spec.get("resources", []) + cores = next((r.get("count", 0) for r in resources + if r.get("type") == "CPU"), 0) + memory = next((r.get("sizeInMB", 0) for r in resources + if r.get("type") == "MEMORY"), 0) + free_executor = state.executor_list_entry + max_by_cores = free_executor["freeCores"] // cores if cores > 0 else num_instances + max_by_memory = free_executor["freeMemory"] // memory if memory > 0 else num_instances + placed = min(num_instances, max_by_cores, max_by_memory) + detail = request.args.get("detail", "false").lower() == "true" + placements = [] + if detail: + for _ in range(placed): + placements.append({ + "executorId": free_executor["executorId"], + "hostname": free_executor["hostname"], + "port": free_executor["port"], + "transportType": free_executor["transportType"], + "executorState": "ACTIVE", + "tags": free_executor.get("tags", []), + }) + errors = [] + if placed < num_instances: + errors.append( + "Cluster can place only {placed} of {requested} requested instances".format( + placed=placed, requested=num_instances)) + return ok({ + "requested": num_instances, + "placed": placed, + "errors": errors, + "placements": placements, + }) # ------------------------------------------------------------------ apps @app.route("/apis/v1/applications") def apps_list(): diff --git a/tests/test_cluster.py b/tests/test_cluster.py index d32370a..1efae11 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -1,9 +1,10 @@ """ tests/test_cluster.py — tests for `drove cluster` commands. -Covers: ping, summary, leader, endpoints, describe cluster +Covers: ping, summary, leader, endpoints, describe cluster, simulate-placement All tests are smoke/read-only (no mutation). """ +from pathlib import Path import pytest from conftest import drove_ok, drove @@ -95,3 +96,33 @@ def test_describe_cluster_json(self): def test_describe_cluster_contains_executors(self): out = drove_ok("describe", "cluster") assert "Executor" in out or "exec" in out.lower() + + +# --------------------------------------------------------------------------- +# drove cluster simulate-placement +# --------------------------------------------------------------------------- + +class TestClusterSimulatePlacement: + """Simulation is read-only — safe to run against a live cluster.""" + + TASK_SPEC = str(Path(__file__).parent.parent / "sample" / "test_task.json") + + def test_simulate_placement_succeeds(self): + out = drove_ok("cluster", "simulate-placement", self.TASK_SPEC, + "--num-instances", "2") + assert "Requested instances: 2" in out, f"Unexpected output: {out}" + assert "Placed instances: 2" in out, f"Unexpected output: {out}" + + def test_simulate_placement_json(self): + import json + out = drove_ok("cluster", "simulate-placement", self.TASK_SPEC, + "--num-instances", "1", "--json") + data = json.loads(out) + assert data["requested"] == 1 + assert "placed" in data + assert "errors" in data + + def test_simulate_placement_detail_shows_executors(self): + out = drove_ok("cluster", "simulate-placement", self.TASK_SPEC, + "--num-instances", "1", "--detail") + assert "Executor Id" in out, f"Expected placement table: {out}" diff --git a/tests/test_offline_cluster.py b/tests/test_offline_cluster.py index 3008eba..dbf7f92 100644 --- a/tests/test_offline_cluster.py +++ b/tests/test_offline_cluster.py @@ -95,3 +95,93 @@ def test_describe_cluster_contains_executors(self, offline_env): from conftest import drove_ok out = drove_ok("describe", "cluster") assert "Executor" in out or "exec" in out.lower() + + +class TestOfflineClusterSimulatePlacement: + """Offline tests for `drove cluster simulate-placement`.""" + + SIM_SPEC = { + "type": "COMPUTATION", + "sourceAppName": "TEST_APP", + "taskId": "SIM001", + "executable": { + "type": "DOCKER", + "url": "ghcr.io/appform-io/test-task", + "dockerPullTimeout": "100 seconds", + }, + "resources": [ + {"type": "CPU", "count": 1}, + {"type": "MEMORY", "sizeInMB": 512}, + ], + "placementPolicy": {"type": "ANY"}, + } + + @pytest.fixture() + def sim_spec_file(self, tmp_path): + spec_file = tmp_path / "sim_task.json" + spec_file.write_text(json.dumps(self.SIM_SPEC)) + return str(spec_file) + + def test_simulate_placement_succeeds(self, offline_env, sim_spec_file): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file, + "--num-instances", "2") + assert "Requested instances: 2" in out, f"Unexpected output: {out}" + assert "Placed instances: 2" in out, f"Unexpected output: {out}" + assert "Errors: 0" in out, f"Unexpected output: {out}" + + def test_simulate_placement_default_one_instance(self, offline_env, sim_spec_file): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file) + assert "Requested instances: 1" in out, f"Unexpected output: {out}" + + def test_simulate_placement_detail_shows_placements(self, offline_env, sim_spec_file): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file, + "--num-instances", "2", "--detail") + assert "Executor Id" in out, f"Expected placement table in output: {out}" + assert "exec-host-1" in out, f"Expected seed executor in placements: {out}" + + def test_simulate_placement_json_output(self, offline_env, sim_spec_file): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file, + "--num-instances", "2", "--json") + data = json.loads(out) + assert data["requested"] == 2 + assert data["placed"] == 2 + assert data["errors"] == [] + assert data["placements"] == [] + + def test_simulate_placement_json_detail_has_placements(self, offline_env, sim_spec_file): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file, + "--num-instances", "2", "--detail", "--json") + data = json.loads(out) + assert len(data["placements"]) == 2 + placement = data["placements"][0] + assert placement["hostname"] == "exec-host-1" + assert placement["executorState"] == "ACTIVE" + + def test_simulate_placement_over_capacity_reports_shortfall(self, offline_env, sim_spec_file): + """Seed executor: 8 free cores, 2048 MB. One instance needs 1 core, + 512 MB — so at most 4 instances can be placed.""" + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", sim_spec_file, + "--num-instances", "100") + assert "Placed instances: 4" in out, f"Unexpected output: {out}" + assert "Cluster can place only 4 of 100" in out, f"Unexpected output: {out}" + + def test_simulate_placement_spec_without_type_fails(self, offline_env): + from conftest import drove_ok + bad_spec = {k: v for k, v in self.SIM_SPEC.items() if k != "type"} + with open("/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/bad_sim_spec.json", "w") as fp: + json.dump(bad_spec, fp) + out = drove_ok("cluster", "simulate-placement", + "/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/bad_sim_spec.json") + assert "Placement simulation failed" in out, f"Unexpected output: {out}" + + def test_simulate_placement_missing_file_fails(self, offline_env): + from conftest import drove_ok + out = drove_ok("cluster", "simulate-placement", + "/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/no_such_spec.json") + assert "Error reading simulation input" in out, f"Unexpected output: {out}" diff --git a/tests/test_offline_full_help.py b/tests/test_offline_full_help.py index 3384692..975bfa5 100644 --- a/tests/test_offline_full_help.py +++ b/tests/test_offline_full_help.py @@ -25,9 +25,9 @@ # ── constants ──────────────────────────────────────────────────────────────── SEPARATOR = "=" * 72 -EXPECTED_SECTIONS = 80 # 1 root + 9 plugin groups + ~70 sub-commands +EXPECTED_SECTIONS = 81 # 1 root + 9 plugin groups + ~71 sub-commands MIN_EXPECTED_LINES = 975 # lower bound (argparse wrapping varies by Python version) -MAX_EXPECTED_LINES = 1000 # upper bound +MAX_EXPECTED_LINES = 1025 # upper bound # All top-level plugin groups that must appear in the output TOP_LEVEL_GROUPS = [ From 5c466fac6a076c562ba8d9a5821c4ea792d61497 Mon Sep 17 00:00:00 2001 From: Santanu Sinha Date: Tue, 29 Sep 2026 08:25:24 +0530 Subject: [PATCH 2/4] Fixed paths --- tests/test_offline_cluster.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/tests/test_offline_cluster.py b/tests/test_offline_cluster.py index dbf7f92..929432c 100644 --- a/tests/test_offline_cluster.py +++ b/tests/test_offline_cluster.py @@ -171,17 +171,16 @@ def test_simulate_placement_over_capacity_reports_shortfall(self, offline_env, s assert "Placed instances: 4" in out, f"Unexpected output: {out}" assert "Cluster can place only 4 of 100" in out, f"Unexpected output: {out}" - def test_simulate_placement_spec_without_type_fails(self, offline_env): + def test_simulate_placement_spec_without_type_fails(self, offline_env, tmp_path): from conftest import drove_ok bad_spec = {k: v for k, v in self.SIM_SPEC.items() if k != "type"} - with open("/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/bad_sim_spec.json", "w") as fp: - json.dump(bad_spec, fp) - out = drove_ok("cluster", "simulate-placement", - "/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/bad_sim_spec.json") + bad_spec_file = tmp_path / "bad_sim_spec.json" + bad_spec_file.write_text(json.dumps(bad_spec)) + out = drove_ok("cluster", "simulate-placement", str(bad_spec_file)) assert "Placement simulation failed" in out, f"Unexpected output: {out}" - def test_simulate_placement_missing_file_fails(self, offline_env): + def test_simulate_placement_missing_file_fails(self, offline_env, tmp_path): from conftest import drove_ok - out = drove_ok("cluster", "simulate-placement", - "/tmp/sai/23729860-4dae-4f67-b7d5-29d69883e573/scratch/no_such_spec.json") + missing_file = tmp_path / "no_such_spec.json" + out = drove_ok("cluster", "simulate-placement", str(missing_file)) assert "Error reading simulation input" in out, f"Unexpected output: {out}" From 3dcf6e22203e5090a8dcb4054c053e3b9a0e9b5d Mon Sep 17 00:00:00 2001 From: Santanu Sinha Date: Tue, 29 Sep 2026 08:36:32 +0530 Subject: [PATCH 3/4] Added full help --- tests/test_offline_full_help.py | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/tests/test_offline_full_help.py b/tests/test_offline_full_help.py index 975bfa5..a8564f1 100644 --- a/tests/test_offline_full_help.py +++ b/tests/test_offline_full_help.py @@ -9,10 +9,10 @@ * Print exactly 80 ``=``-separator sections (1 root + 79 command/sub-command parsers). * Cover every top-level plugin group and representative sub-commands. - -All tests use the mock server fixture only to inherit the offline_env -environment; ``--full-help`` itself never issues any HTTP request. - +* Print exactly 81 ``=``-separator sections (1 root + 9 plugin groups + + 71 sub-command parsers). +* Cover every top-level plugin group and representative sub-commands, + including ``cluster simulate-placement``. Run with: pytest -m offline tests/test_offline_full_help.py """ import subprocess @@ -62,12 +62,9 @@ ("apps", "summary"), ("apps", "cancelop"), ("cluster", "ping"), + ("cluster", "ping"), ("cluster", "summary"), - ("describe", "app"), - ("describe", "cluster"), - ("describe", "executor"), - ("describe", "instance"), - ("describe", "localservice"), + ("cluster", "simulate-placement"), ("describe", "lsinstance"), ("describe", "task"), ("executor", "list"), @@ -129,10 +126,7 @@ def test_output_line_count(self, offline_env): def test_works_without_drove_endpoint(self, offline_env): """--full-help must not require a live cluster endpoint.""" - env_override = { - "DROVE_ENDPOINT": "", # explicitly blank - } - # Also strip DROVE_CLUSTER so ~/.drove is not consulted + # Strip DROVE_ENDPOINT and DROVE_CLUSTER so ~/.drove is not consulted clean_env = os.environ.copy() clean_env.pop("DROVE_ENDPOINT", None) clean_env.pop("DROVE_CLUSTER", None) From 7129c373203b839ff4fe3559fc7d6050e8b1b211 Mon Sep 17 00:00:00 2001 From: Santanu Sinha Date: Tue, 29 Sep 2026 19:09:36 +0530 Subject: [PATCH 4/4] Added back missing lines and a test to ensure this doesn't recur --- plugins/tasks.py | 1 + tests/test_offline_tasks.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/plugins/tasks.py b/plugins/tasks.py index bd56c9b..e51ad80 100644 --- a/plugins/tasks.py +++ b/plugins/tasks.py @@ -132,6 +132,7 @@ def show_task(self, options: SimpleNamespace): droveutils.populate_resources(raw, data) data["Executable"] = "{url} ({type})".format_map(raw.get("executable", {})) data["Volumes"] = ", ".join(["{pathOnHost} (Mounted as {pathInContainer} in {mode} mode)".format_map(v) for v in raw.get("volumes", [])]) + data["Logging Mode"] = raw.get("logging", dict()).get("type", "") data["Metadata"] = ", ".join(["%s: %s" % (key,value) for (key, value) in raw.get("metadata", dict())]) result = raw.get("taskResult", dict()) if len(result) > 0: diff --git a/tests/test_offline_tasks.py b/tests/test_offline_tasks.py index 8fbe993..50f6066 100644 --- a/tests/test_offline_tasks.py +++ b/tests/test_offline_tasks.py @@ -105,6 +105,15 @@ def test_task_show_contains_task_id(self, app_for_offline_tasks): f"Expected {TASK_ID} in tasks show output:\n{result.stdout}" ) + def test_task_show_contains_logging_mode(self, app_for_offline_tasks): + from conftest import drove + result = drove("tasks", "show", TASK_SOURCE, TASK_ID, + check=False, timeout=10) + assert result.returncode == 0 + assert "Logging Mode" in result.stdout, ( + f"Expected 'Logging Mode' in tasks show output:\n{result.stdout}" + ) + def test_task_logs_list(self, app_for_offline_tasks): from conftest import drove result = drove("tasks", "logs", TASK_SOURCE, TASK_ID,