zstandard compression for parameters and results

CHANGELOG.rst

@@ -3,6 +3,11 @@ Changelog
 
 in development
 --------------
+* implemented zstandard compression for parameters and results. #5995
+  contributed by @guzzijones12
+
+* removed embedded liveaction in action execution database table #5995
+  contributed by @guzzijones12
 
 Python 3.6 is no longer supported; Stackstorm requires at least Python 3.8.
 

conf/st2.conf.sample

@@ -140,6 +140,9 @@ connection_timeout = 3000
 db_name = st2
 # host of db server
 host = 127.0.0.1
+# compression for parameter and result storage in liveaction and execution models
+# Valid values: zstandard, none
+parameter_result_compression = zstandard
 # password for db login
 password = None
 # port of db server

conf/st2.dev.conf

@@ -1,6 +1,7 @@
 # Config used by local development environment (tools/launch.dev.sh)
 [database]
 host = 127.0.0.1
+parameter_result_compression = zstandard
 
 [api]
 # Host and port to bind the API server.

contrib/runners/action_chain_runner/action_chain_runner/action_chain_runner.py

@@ -333,7 +333,7 @@ def cancel(self):
                 and child_exec.status in action_constants.LIVEACTION_CANCELABLE_STATES
             ):
                 action_service.request_cancellation(
-                    LiveAction.get(id=child_exec.liveaction["id"]),
+                    LiveAction.get(id=child_exec.liveaction_id),
                     self.context.get("user", None),
                 )
 
@@ -353,7 +353,7 @@ def pause(self):
                 and child_exec.status == action_constants.LIVEACTION_STATUS_RUNNING
             ):
                 action_service.request_pause(
-                    LiveAction.get(id=child_exec.liveaction["id"]),
+                    LiveAction.get(id=child_exec.liveaction_id),
                     self.context.get("user", None),
                 )
 
@@ -966,7 +966,7 @@ def _format_action_exec_result(
 
         execution_db = None
         if liveaction_db:
-            execution_db = ActionExecution.get(liveaction__id=str(liveaction_db.id))
+            execution_db = ActionExecution.get(liveaction_id=str(liveaction_db.id))
 
         result["id"] = action_node.name
         result["name"] = action_node.name

contrib/runners/action_chain_runner/tests/unit/test_actionchain_cancel.py

@@ -171,7 +171,7 @@ def test_chain_cancel_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is running.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_RUNNING
         )
@@ -189,7 +189,7 @@ def test_chain_cancel_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is canceling.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_CANCELING
         )
@@ -206,7 +206,7 @@ def test_chain_cancel_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is canceled.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_CANCELED
         )
@@ -248,7 +248,7 @@ def test_chain_cancel_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is running.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_RUNNING
         )
@@ -260,7 +260,7 @@ def test_chain_cancel_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is canceling.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_CANCELING
         )
@@ -271,7 +271,7 @@ def test_chain_cancel_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is canceled.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_CANCELED
         )

contrib/runners/action_chain_runner/tests/unit/test_actionchain_notifications.py

@@ -151,7 +151,7 @@ def test_skip_notify_for_task_with_notify(self):
 
         # Assert task1 notify is skipped
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_SUCCEEDED
         )
@@ -162,7 +162,7 @@ def test_skip_notify_for_task_with_notify(self):
 
         # Assert task2 notify is not skipped
         task2_exec = ActionExecution.get_by_id(execution.children[1])
-        task2_live = LiveAction.get_by_id(task2_exec.liveaction["id"])
+        task2_live = LiveAction.get_by_id(task2_exec.liveaction_id)
         notify = notify_api_models.NotificationsHelper.from_model(
             notify_model=task2_live.notify
         )
@@ -186,7 +186,7 @@ def test_skip_notify_default_for_task_with_notify(self):
 
         # Assert task1 notify is set.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_on_status(
             task1_live, action_constants.LIVEACTION_STATUS_SUCCEEDED
         )
@@ -200,7 +200,7 @@ def test_skip_notify_default_for_task_with_notify(self):
 
         # Assert task2 notify is not skipped by default.
         task2_exec = ActionExecution.get_by_id(execution.children[1])
-        task2_live = LiveAction.get_by_id(task2_exec.liveaction["id"])
+        task2_live = LiveAction.get_by_id(task2_exec.liveaction_id)
         self.assertIsNone(task2_live.notify)
         MockLiveActionPublisherNonBlocking.wait_all()
 

contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py

@@ -431,7 +431,7 @@ def test_chain_pause_resume_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is running.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_RUNNING
         )
@@ -452,7 +452,7 @@ def test_chain_pause_resume_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is pausing.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_PAUSING
         )
@@ -477,7 +477,7 @@ def test_chain_pause_resume_cascade_to_subworkflow(self):
 
         # Wait until the subworkflow is paused.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_PAUSED
         )
@@ -548,7 +548,7 @@ def test_chain_pause_resume_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is running.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_RUNNING
         )
@@ -559,7 +559,7 @@ def test_chain_pause_resume_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is pausing.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_PAUSING
         )
@@ -574,7 +574,7 @@ def test_chain_pause_resume_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is paused.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_PAUSED
         )
@@ -611,7 +611,7 @@ def test_chain_pause_resume_cascade_to_parent_workflow(self):
 
         # Wait until the subworkflow is paused.
         task1_exec = ActionExecution.get_by_id(execution.children[0])
-        task1_live = LiveAction.get_by_id(task1_exec.liveaction["id"])
+        task1_live = LiveAction.get_by_id(task1_exec.liveaction_id)
         task1_live = self._wait_for_status(
             task1_live, action_constants.LIVEACTION_STATUS_SUCCEEDED
         )

contrib/runners/inquirer_runner/inquirer_runner/inquirer_runner.py

@@ -74,7 +74,7 @@ def pre_run(self):
 
     def run(self, action_parameters):
         liveaction_db = action_utils.get_liveaction_by_id(self.liveaction_id)
-        exc = ex_db_access.ActionExecution.get(liveaction__id=str(liveaction_db.id))
+        exc = ex_db_access.ActionExecution.get(liveaction_id=str(liveaction_db.id))
 
         # Assemble and dispatch trigger
         trigger_ref = sys_db_models.ResourceReference.to_string_reference(

contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py

@@ -136,12 +136,22 @@ def start_workflow(self, action_parameters):
                 wf_def, self.execution, st2_ctx, notify_cfg=notify_cfg
             )
         except wf_exc.WorkflowInspectionError as e:
+            _, ex, tb = sys.exc_info()
             status = ac_const.LIVEACTION_STATUS_FAILED
-            result = {"errors": e.args[1], "output": None}
+            result = {
+                "errors": e.args[1],
+                "output": None,
+                "traceback": "".join(traceback.format_tb(tb, 20)),
+            }
             return (status, result, self.context)
         except Exception as e:
+            _, ex, tb = sys.exc_info()
             status = ac_const.LIVEACTION_STATUS_FAILED
-            result = {"errors": [{"message": six.text_type(e)}], "output": None}
+            result = {
+                "errors": [{"message": six.text_type(e)}],
+                "output": None,
+                "traceback": "".join(traceback.format_tb(tb, 20)),
+            }
             return (status, result, self.context)
 
         return self._handle_workflow_return_value(wf_ex_db)
@@ -178,7 +188,7 @@ def pause(self):
             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
             if self.task_pauseable(child_ex):
                 ac_svc.request_pause(
-                    lv_db_access.LiveAction.get(id=child_ex.liveaction["id"]),
+                    lv_db_access.LiveAction.get(id=child_ex.liveaction_id),
                     self.context.get("user", None),
                 )
 
@@ -209,7 +219,7 @@ def resume(self):
             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
             if self.task_resumeable(child_ex):
                 ac_svc.request_resume(
-                    lv_db_access.LiveAction.get(id=child_ex.liveaction["id"]),
+                    lv_db_access.LiveAction.get(id=child_ex.liveaction_id),
                     self.context.get("user", None),
                 )
 
@@ -270,7 +280,7 @@ def cancel(self):
             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
             if self.task_cancelable(child_ex):
                 ac_svc.request_cancellation(
-                    lv_db_access.LiveAction.get(id=child_ex.liveaction["id"]),
+                    lv_db_access.LiveAction.get(id=child_ex.liveaction_id),
                     self.context.get("user", None),
                 )
 

contrib/runners/orquesta_runner/tests/unit/test_basic.py

@@ -185,7 +185,7 @@ def test_run_workflow(self):
         tk1_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk1_ex_db.id)
         )[0]
-        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction["id"])
+        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction_id)
         self.assertEqual(tk1_lv_ac_db.context.get("user"), username)
         self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk1_ac_ex_db))
@@ -205,7 +205,7 @@ def test_run_workflow(self):
         tk2_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk2_ex_db.id)
         )[0]
-        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction["id"])
+        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction_id)
         self.assertEqual(tk2_lv_ac_db.context.get("user"), username)
         self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk2_ac_ex_db))
@@ -225,7 +225,7 @@ def test_run_workflow(self):
         tk3_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk3_ex_db.id)
         )[0]
-        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction["id"])
+        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction_id)
         self.assertEqual(tk3_lv_ac_db.context.get("user"), username)
         self.assertEqual(tk3_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk3_ac_ex_db))
@@ -275,7 +275,7 @@ def test_run_workflow_with_unicode_input(self):
         tk1_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk1_ex_db.id)
         )[0]
-        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction["id"])
+        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction_id)
         self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
         tk1_ex_db = wf_db_access.TaskExecution.get_by_id(tk1_ex_db.id)
@@ -287,7 +287,7 @@ def test_run_workflow_with_unicode_input(self):
         tk2_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk2_ex_db.id)
         )[0]
-        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction["id"])
+        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction_id)
         self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         wf_svc.handle_action_execution_completion(tk2_ac_ex_db)
         tk2_ex_db = wf_db_access.TaskExecution.get_by_id(tk2_ex_db.id)
@@ -299,7 +299,7 @@ def test_run_workflow_with_unicode_input(self):
         tk3_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk3_ex_db.id)
         )[0]
-        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction["id"])
+        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction_id)
         self.assertEqual(tk3_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         wf_svc.handle_action_execution_completion(tk3_ac_ex_db)
         tk3_ex_db = wf_db_access.TaskExecution.get_by_id(tk3_ex_db.id)
@@ -348,7 +348,7 @@ def test_run_workflow_action_config_context(self):
         tk1_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk1_ex_db.id)
         )[0]
-        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction["id"])
+        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction_id)
         self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
         self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk1_ac_ex_db))
 
@@ -401,7 +401,7 @@ def test_run_workflow_with_action_less_tasks(self):
         tk2_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk2_ex_db.id)
         )[0]
-        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction["id"])
+        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction_id)
         self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
 
         # Manually handle action execution completion.
@@ -413,7 +413,7 @@ def test_run_workflow_with_action_less_tasks(self):
         tk3_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk3_ex_db.id)
         )[0]
-        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction["id"])
+        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction_id)
         self.assertEqual(tk3_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
 
         # Manually handle action execution completion.
@@ -434,7 +434,7 @@ def test_run_workflow_with_action_less_tasks(self):
         tk5_ac_ex_db = ex_db_access.ActionExecution.query(
             task_execution=str(tk5_ex_db.id)
         )[0]
-        tk5_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk5_ac_ex_db.liveaction["id"])
+        tk5_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk5_ac_ex_db.liveaction_id)
         self.assertEqual(tk5_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
 
         # Manually handle action execution completion.