Add invoke_rust_agent_expect_result
* Add invoke_rust_agent_expect_result
Create a wrapper for invoke_rust_agent that will handle errors at the
base Job level so that the handling does not have to be done in each
job.
* - Fuxups
# Use of this source code is governed by a MIT-style
# license that can be found in the LICENSE file.
-
+import json
import django.db.models
from chroma_core.services import log_register
@@ -197,6 +197,17 @@ class Step(object):
return invoke_rust_agent(host, command, args, self._cancel_event)
+ def invoke_rust_agent_expect_result(self, host, command, args={}):
+ from chroma_core.services.job_scheduler.agent_rpc import AgentException
+
+ result = json.loads(self.invoke_rust_agent(host, command, args))
+
+ if "Err" in result:
+ self.log(json.dumps(result["Err"], indent=2))
+ raise AgentException(host.fqdn, command, args, result["Err"])
+
+ return result["Ok"]
+
def invoke_agent_expect_result(self, host, command, args={}):
from chroma_core.services.job_scheduler.agent_rpc import AgentException
