Added support for placement simulation

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,11 +130,62 @@ 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"]
         data["id"] = event["id"]
         data["time"] = droveutils.to_date(event["time"])
         data["metadata"] = json.dumps(event["metadata"])
 
-        return format.format_map(data)
+        return format.format_map(data)

plugins/tasks.py

@@ -124,14 +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["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())
@@ -155,4 +155,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)
+        droveutils.download_log(self.drove_client, "tasks", options.source_app, options.task_id, options.tasklogfile, filename)

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():

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}"

tests/test_offline_cluster.py

@@ -95,3 +95,92 @@ 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, tmp_path):
+        from conftest import drove_ok
+        bad_spec = {k: v for k, v in self.SIM_SPEC.items() if k != "type"}
+        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, tmp_path):
+        from conftest import drove_ok
+        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}"

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
@@ -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 = [
@@ -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)

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,