Rollback field changes on workflow and task execution collection

contrib/examples/actions/orquesta-mock-create-vm.yaml

@@ -14,3 +14,9 @@ parameters:
   vm_name:
     required: true
     type: string
+  ip:
+    default: "10.1.23.99" 
+    required: true
+    type: string
+  meta:
+    type: object

contrib/examples/actions/python-mock-create-vm.py

@@ -0,0 +1,21 @@
+import eventlet
+import random
+
+from st2common.runners.base_action import Action
+
+
+class MockCreateVMAction(Action):
+
+    def run(self, cpu_cores, memory_mb, vm_name, ip):
+        eventlet.sleep(5)
+
+        data = {
+            'vm_id': 'vm' + str(random.randint(0, 10000)),
+            ip: {
+                'cpu_cores': cpu_cores,
+                'memory_mb': memory_mb,
+                'vm_name': vm_name
+            }
+        }
+
+        return data

contrib/examples/actions/python-mock-create-vm.yaml

@@ -0,0 +1,19 @@
+---
+name: python-mock-create-vm
+runner_type: python-script
+description: Action which mocks creation of a VM.
+enabled: true
+entry_point: python-mock-create-vm.py
+parameters:
+  cpu_cores:
+    default: 1
+    type: integer
+  memory_mb:
+    default: 1024
+    type: integer
+  vm_name:
+    required: true
+    type: string
+  ip:
+    required: true
+    type: string

contrib/examples/actions/workflows/orquesta-mock-create-vm.yaml

@@ -6,10 +6,16 @@ input:
   - vm_name
   - cpu_cores
   - memory_mb
+  - ip
+  - meta
+
+vars:
+  - extra: <% ctx().meta %>
 
 output:
   - vm_id: <% ctx().vm_id %>
   - ip: <% ctx().ip %>
+  - extra: <% ctx().extra %>
 
 tasks:
   register_dns:
@@ -19,22 +25,24 @@ tasks:
     next:
       - when: <% succeeded() %>
         publish:
-          - ip: "10.1.23.99"
+          - ip: <% ctx().ip %>
           - status_message: "DNS for <% ctx().vm_name %> is registered."
         do:
           - configure_vm
           - notify
 
 
   create_vm:
-    action: core.local
+    action: examples.python-mock-create-vm
     input:
-      cmd: "printf 'vm1234'; sleep 6"
+      vm_name: <% ctx().vm_name %>
+      ip: <% ctx().ip %>
     next:
       - when: <% succeeded() %>
         publish:
-          - vm_id: <% result().stdout %>
+          - vm_id: <% result().result.vm_id %>
           - status_message: "VM <% ctx().vm_name %> is created."
+          - extra: <% result().result %>
         do:
           - configure_vm
           - notify

scripts/travis/prepare-integration.sh

@@ -28,6 +28,9 @@ echo " === START: Catting screen process log files. ==="
 cat logs/screen-*.log
 echo " === END: Catting screen process log files. ==="
 
+# Setup the virtualenv for the examples pack which is required for orquesta integration tests.
+st2 run packs.setup_virtualenv packs=examples
+
 # This script runs as root on Travis which means other processes which don't run
 # as root can't write to logs/ directory and tests fail
 chmod 777 logs/

st2common/st2common/models/db/workflow.py

@@ -38,12 +38,12 @@ class WorkflowExecutionDB(stormbase.StormFoundationDB, stormbase.ChangeRevisionF
     action_execution = me.StringField(required=True)
     spec = me.DictField()
     graph = me.DictField()
-    input = me.DictField()
+    input = stormbase.EscapedDictField()
     notify = me.DictField()
     context = me.DictField()
-    state = me.DictField()
+    state = stormbase.EscapedDictField()
     status = me.StringField(required=True)
-    output = me.DictField()
+    output = stormbase.EscapedDictField()
     errors = me.DynamicField()
     start_timestamp = db_field_types.ComplexDateTimeField(default=date_utils.get_datetime_utc_now)
     end_timestamp = db_field_types.ComplexDateTimeField()
@@ -67,9 +67,9 @@ class TaskExecutionDB(stormbase.StormFoundationDB, stormbase.ChangeRevisionField
     itemized = me.BooleanField(default=False)
     items_count = me.IntField(min_value=0)
     items_concurrency = me.IntField(min_value=1)
-    context = me.DictField()
+    context = stormbase.EscapedDictField()
     status = me.StringField(required=True)
-    result = me.DictField()
+    result = stormbase.EscapedDictField()
     start_timestamp = db_field_types.ComplexDateTimeField(default=date_utils.get_datetime_utc_now)
     end_timestamp = db_field_types.ComplexDateTimeField()
 

st2tests/integration/orquesta/test_performance.py

@@ -17,6 +17,7 @@
 from __future__ import absolute_import
 
 import eventlet
+import json
 
 from integration.orquesta import base
 from six.moves import range
@@ -31,14 +32,14 @@ def test_concurrent_load(self):
         delay_poll = load_count * 5
 
         wf_name = 'examples.orquesta-mock-create-vm'
-        wf_input = {'vm_name': 'demo1'}
+        wf_input = {'vm_name': 'demo1', 'meta': {'demo1.itests.org': '10.3.41.99'}}
         exs = [self._execute_workflow(wf_name, wf_input) for i in range(load_count)]
 
         eventlet.sleep(delay_poll)
 
         for ex in exs:
             e = self._wait_for_completion(ex)
-            self.assertEqual(e.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+            self.assertEqual(e.status, ac_const.LIVEACTION_STATUS_SUCCEEDED, json.dumps(e.result))
             self.assertIn('output', e.result)
             self.assertIn('vm_id', e.result['output'])