Draft: operations schema update: send op metadata

README.md

@@ -110,6 +110,22 @@ drove -c prod apps info <app-name>
 
 Use `drove -h` or `drove <command> -h` for detailed help.
 
+### Operation Metadata
+
+Supply an optional JSON object with `--metadata` on `apps create`, `destroy`,
+`deploy`, `scale`, `suspend`, `restart`, and `appinstances replace` or `kill`:
+
+```bash
+drove -c prod apps scale TEST_APP-1 3 \
+    --metadata '{"initiator":"ops","reason":"capacity","ticket":123}'
+```
+
+The object is sent as the top-level `operationMetadata` field, separate from
+the application spec and `opSpec`. Values can be nested JSON objects, arrays,
+strings, numbers, booleans, or null. The server must support operation metadata
+to use this option. Omitting it leaves the request unchanged for older servers.
+It is not supported for task or local-service commands.
+
 ### Task Logs
 
 ```bash

droveutils.py

@@ -1,9 +1,20 @@
+import argparse
 import datetime
 import droveclient
 import json
 import tabulate
 import time
 
+def parse_operation_metadata(value: str) -> dict[str, object]:
+    try:
+        metadata = json.loads(value)
+        json.dumps(metadata, allow_nan=False)
+    except ValueError as error:
+        raise argparse.ArgumentTypeError("metadata must be a JSON object with valid JSON values") from error
+    if not isinstance(metadata, dict):
+        raise argparse.ArgumentTypeError("metadata must be a JSON object")
+    return metadata
+
 def print_dict(data: dict, level: int = 0):
     for key, value in data.items():
         print(level * 4 * " ", end='')

plugins/appinstances.py

@@ -58,6 +58,7 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure all instances are replaced", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.replace)
 
         sub_parser = commands.add_parser("kill", help="Kill specific app instances")
@@ -66,6 +67,7 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure all instances are killed", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.kill)
 
         # sub_parser = commands.add_parser("create", help="Create application")
@@ -143,6 +145,8 @@ def replace(self, options):
                 "failureStrategy": "STOP"
             }
         }
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             self.ensure_replaced(options.app_id, set(options.instance_ids))
@@ -162,6 +166,8 @@ def kill(self, options):
                 "failureStrategy": "STOP"
             }
         }
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             self.ensure_replaced(options.app_id, set(options.instance_ids))

plugins/applications.py

@@ -36,10 +36,12 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg
 
         sub_parser = commands.add_parser("create", help="Create application on cluster")
         sub_parser.add_argument("spec_file", metavar="spec-file", help="JSON spec file for the application")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.create_app)
 
         sub_parser = commands.add_parser("destroy", help="Destroy an app with zero instances")
         sub_parser.add_argument("app_id", metavar="app-id", help="Application ID")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.destroy_app)
 
         sub_parser = commands.add_parser("deploy", help="Deploy new app instances.")
@@ -48,6 +50,7 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure instance count is reached", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.deploy_app)
 
 
@@ -57,20 +60,23 @@ def populate_options(self, drove_client: droveclient.DroveClient, subparser: arg
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure instance count is reached", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.scale_app)
 
         sub_parser = commands.add_parser("suspend", help="Suspend the app")
         sub_parser.add_argument("app_id", metavar="app-id", help="Application ID")
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure all instances are suspended", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.suspend_app)
 
         sub_parser = commands.add_parser("restart", help="Restart am existing app instances.")
         sub_parser.add_argument("app_id", metavar="app-id", help="Application ID")
         sub_parser.add_argument("--parallelism", "-p", help="Number of parallel threads to be used to execute operation", type=int, default = 1)
         sub_parser.add_argument("--timeout", "-t", help="Timeout for the operation on the cluster", type=str, default = "5m")
         sub_parser.add_argument("--wait", "-w", help="Wait to ensure all instances are replaced", default=False, action="store_true")
+        sub_parser.add_argument("--metadata", metavar="JSON", type=droveutils.parse_operation_metadata, help="Optional operation metadata as a JSON object")
         sub_parser.set_defaults(func=self.restart_app)
 
         sub_parser = commands.add_parser("cancelop", help="Cancel current operation")
@@ -123,6 +129,8 @@ def create_app(self, options: SimpleNamespace):
                     "failureStrategy": "STOP"
                 }
             }
+            if getattr(options, "metadata", None) is not None:
+                operation["operationMetadata"] = options.metadata
             data = self.drove_client.post("/apis/v1/applications/operations", operation)
             print("Application created with app id: {appid}".format(appid=data["appId"]))
         except (OSError, IOError) as e:
@@ -138,6 +146,8 @@ def destroy_app(self, options: SimpleNamespace):
                 "failureStrategy": "STOP"
             }
         }
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         print("Application destroyed")
 
@@ -152,6 +162,8 @@ def scale_app(self, options: SimpleNamespace):
                 "failureStrategy": "STOP"
             }
         }
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             print("Waiting till required scale is reached")
@@ -170,6 +182,8 @@ def suspend_app(self, options: SimpleNamespace):
                 "failureStrategy": "STOP"
             }
         }
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             print("Waiting till all instances shut down")
@@ -190,6 +204,8 @@ def deploy_app(self, options: SimpleNamespace):
             }
         }
         existing_count = 0 if options.wait == False else len(self.drove_client.app_instances(options.app_id))
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             print("Waiting till required scale is reached")
@@ -209,6 +225,8 @@ def restart_app(self, options: SimpleNamespace):
             }
         }
         existing = [] if options.wait == False else self.drove_client.app_instances(options.app_id)
+        if getattr(options, "metadata", None) is not None:
+            operation["operationMetadata"] = options.metadata
         data = self.drove_client.post("/apis/v1/applications/operations", operation)
         if options.wait:
             self.ensure_replaced(options.app_id, set(existing))

tests/test_offline_apps.py

@@ -5,7 +5,11 @@
 All tests use the mock Drove server; no live cluster is required.
 Run with:  pytest -m offline tests/test_offline_apps.py
 """
+import argparse
 import json
+from unittest.mock import Mock
+
+import plugins
 import pytest
 
 pytestmark = pytest.mark.offline
@@ -176,3 +180,73 @@ def test_destroyed_app_not_listed(self, offline_env):
         from conftest import drove_ok
         out = drove_ok("apps", "list")
         assert CLI_APP_ID not in out, f"{CLI_APP_ID} still appears after destroy"
+
+
+OPERATION_COMMANDS = [
+    (("apps", "create"), {"type": "CREATE"}),
+    (("apps", "destroy", CLI_APP_ID), {"type": "DESTROY", "appId": CLI_APP_ID}),
+    (("apps", "deploy", CLI_APP_ID, "2"), {"type": "START_INSTANCES", "appId": CLI_APP_ID, "instances": 2}),
+    (("apps", "scale", CLI_APP_ID, "2"), {"type": "SCALE", "appId": CLI_APP_ID, "requiredInstances": 2}),
+    (("apps", "suspend", CLI_APP_ID), {"type": "SUSPEND", "appId": CLI_APP_ID}),
+    (("apps", "restart", CLI_APP_ID), {"type": "REPLACE_INSTANCES", "appId": CLI_APP_ID}),
+    (("appinstances", "replace", CLI_APP_ID, "AI-test-1", "AI-test-2"),
+     {"type": "REPLACE_INSTANCES", "appId": CLI_APP_ID, "instanceIds": ["AI-test-1", "AI-test-2"]}),
+    (("appinstances", "kill", CLI_APP_ID, "AI-test-1", "AI-test-2"),
+     {"type": "STOP_INSTANCES", "appId": CLI_APP_ID, "instanceIds": ["AI-test-1", "AI-test-2"], "skipRespawn": True}),
+]
+
+
+class TestOfflineOperationMetadata:
+    @pytest.mark.parametrize("command, expected", OPERATION_COMMANDS)
+    @pytest.mark.parametrize("metadata", [None, {}, {
+        "initiator": "ops", "reason": "capacity", "ticket": 123,
+        "details": {"approved": True, "tags": ["manual"], "previous": None},
+    }])
+    def test_operation_payload(self, command, expected, metadata):
+        from conftest import APP_SPEC
+
+        client = Mock()
+        client.post.return_value = {"appId": CLI_APP_ID}
+        plugin = next(plugin_type() for plugin_type in plugins.DrovePlugin.plugins
+                      if plugin_type().name() == command[0])
+        parser = argparse.ArgumentParser()
+        plugin.populate_options(client, parser.add_subparsers())
+        args = list(command)
+        payload = dict(expected)
+        payload["opSpec"] = {"timeout": "5m", "parallelism": 1, "failureStrategy": "STOP"}
+        if command[1] == "create":
+            args.append(APP_SPEC)
+            with open(APP_SPEC) as spec_file:
+                payload["spec"] = json.load(spec_file)
+        if metadata is not None:
+            args += ["--metadata", json.dumps(metadata)]
+            payload["operationMetadata"] = metadata
+
+        options = parser.parse_args(args)
+        options.func(options)
+
+        client.post.assert_called_once_with("/apis/v1/applications/operations", payload)
+
+    @pytest.mark.parametrize("command, expected", OPERATION_COMMANDS)
+    @pytest.mark.parametrize("metadata", [
+        "not-json", "[]", "null", '"text"', "42", "true",
+        '{"value": NaN}', '{"value": Infinity}', '{"value": -Infinity}', '{"value": 1e400}',
+    ])
+    def test_invalid_metadata_rejected(self, command, expected, metadata, capsys):
+        from conftest import APP_SPEC
+
+        client = Mock()
+        plugin = next(plugin_type() for plugin_type in plugins.DrovePlugin.plugins
+                      if plugin_type().name() == command[0])
+        parser = argparse.ArgumentParser()
+        plugin.populate_options(client, parser.add_subparsers())
+        args = list(command)
+        if command[1] == "create":
+            args.append(APP_SPEC)
+
+        with pytest.raises(SystemExit) as error:
+            parser.parse_args(args + ["--metadata", metadata])
+
+        assert error.value.code == 2
+        assert "metadata must be a JSON object" in capsys.readouterr().err
+        client.post.assert_not_called()