We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Custom webhook fails action depending on payload.
Hi there, I've faced a weird issue.

Say it we have a webhook action like this: https://gist.github.com/dennybaa/00d1936af2f441cfff15
If I issue the following commands:
1. `curl -X POST --data "{\"message\": \"6fd781b849bb\"}" http://172.17.0.8:9101/v1/webhooks/sample`
2. `curl -X POST --data "{\"hits.rate_15m\": 0.4765927732670934, \"hits.count\": 30, \"tags\": [\"metric\"], \"@timestamp\": \"2015-04-23T13:58:28.319Z\", \"hits.rate_5m\": 0.31123263029226506, \"@version\": \"1\", \"message\": \"6fd781b849bb\", \"hits.rate_1m\": 0.08765952620175455}"`

In both cases trigger is dispatched. But the action takes place only in the #1 case.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `st2actions/st2actions/resultstracker.py`
Content:
```
1 # Licensed to the StackStorm, Inc ('StackStorm') under one or more
2 # contributor license agreements.  See the NOTICE file distributed with
3 # this work for additional information regarding copyright ownership.
4 # The ASF licenses this file to You under the Apache License, Version 2.0
5 # (the "License"); you may not use this file except in compliance with
6 # the License.  You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import eventlet
17 import importlib
18 import six
19 
20 from collections import defaultdict
21 from kombu import Connection
22 from oslo.config import cfg
23 
24 from st2actions.query.base import QueryContext
25 from st2common import log as logging
26 from st2common.models.db import action as action_models
27 from st2common.persistence.action import ActionExecutionState
28 from st2common.transport import actionexecutionstate, consumers, publishers
29 
30 
31 LOG = logging.getLogger(__name__)
32 
33 ACTIONSTATE_WORK_Q = actionexecutionstate.get_queue('st2.resultstracker.work',
34                                                     routing_key=publishers.CREATE_RK)
35 
36 
37 class ResultsTracker(consumers.MessageHandler):
38     message_type = action_models.ActionExecutionStateDB
39 
40     def __init__(self, connection, queues):
41         super(ResultsTracker, self).__init__(connection, queues)
42         self._queriers = {}
43         self._query_threads = []
44         self._failed_imports = set()
45 
46     def start(self, wait=False):
47         self._bootstrap()
48         super(ResultsTracker, self).start(wait=wait)
49 
50     def wait(self):
51         super(ResultsTracker, self).wait()
52         for thread in self._query_threads():
53             thread.wait()
54 
55     def shutdown(self):
56         super(ResultsTracker, self).shutdown()
57         LOG.info('Stats from queriers:')
58         self._print_stats()
59 
60     def _print_stats(self):
61         for name, querier in six.iteritems(self._queriers):
62             if querier:
63                 querier.print_stats()
64 
65     def _bootstrap(self):
66         all_states = ActionExecutionState.get_all()
67         LOG.info('Found %d pending states in db.' % len(all_states))
68 
69         query_contexts_dict = defaultdict(list)
70         for state_db in all_states:
71             try:
72                 context = QueryContext.from_model(state_db)
73             except:
74                 LOG.exception('Invalid state object: %s', state_db)
75                 continue
76             query_module_name = state_db.query_module
77             querier = self.get_querier(query_module_name)
78 
79             if querier is not None:
80                 query_contexts_dict[querier].append(context)
81 
82         for querier, contexts in six.iteritems(query_contexts_dict):
83             LOG.info('Found %d pending actions for query module %s', len(contexts), querier)
84             querier.add_queries(query_contexts=contexts)
85 
86     def process(self, query_context):
87         querier = self.get_querier(query_context.query_module)
88         context = QueryContext.from_model(query_context)
89         querier.add_queries(query_contexts=[context])
90         return
91 
92     def get_querier(self, query_module_name):
93         if (query_module_name not in self._queriers and
94                 query_module_name not in self._failed_imports):
95             try:
96                 query_module = self._import_query_module(query_module_name)
97             except:
98                 LOG.exception('Failed importing query module: %s', query_module_name)
99                 self._failed_imports.add(query_module_name)
100                 self._queriers[query_module_name] = None
101             else:
102                 querier = query_module.get_instance()
103                 self._queriers[query_module_name] = querier
104                 self._query_threads.append(eventlet.spawn(querier.start))
105 
106         return self._queriers[query_module_name]
107 
108     def _import_query_module(self, module_name):
109         return importlib.import_module(module_name, package=None)
110 
111 
112 def get_tracker():
113     with Connection(cfg.CONF.messaging.url) as conn:
114         return ResultsTracker(conn, [ACTIONSTATE_WORK_Q])
115
```

Path: `st2common/st2common/models/db/reactor.py`
Content:
```
1 # Licensed to the StackStorm, Inc ('StackStorm') under one or more
2 # contributor license agreements.  See the NOTICE file distributed with
3 # this work for additional information regarding copyright ownership.
4 # The ASF licenses this file to You under the Apache License, Version 2.0
5 # (the "License"); you may not use this file except in compliance with
6 # the License.  You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import mongoengine as me
17 from st2common.models.db import MongoDBAccess
18 from st2common.models.db import stormbase
19 
20 __all__ = [
21     'SensorTypeDB',
22     'TriggerTypeDB',
23     'TriggerDB',
24     'TriggerInstanceDB',
25     'ActionExecutionSpecDB',
26     'RuleDB'
27 ]
28 
29 
30 class SensorTypeDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):
31     """
32     Description of a specific type of a sensor (think of it as a sensor
33     template).
34 
35     Attribute:
36         pack - Name of the content pack this sensor belongs to.
37         artifact_uri - URI to the artifact file.
38         entry_point - Full path to the sensor entry point (e.g. module.foo.ClassSensor).
39         trigger_type - A list of references to the TriggerTypeDB objects exposed by this sensor.
40         poll_interval - Poll interval for this sensor.
41     """
42     name = me.StringField(required=True)
43     pack = me.StringField(required=True, unique_with='name')
44     artifact_uri = me.StringField()
45     entry_point = me.StringField()
46     trigger_types = me.ListField(field=me.StringField())
47     poll_interval = me.IntField()
48     enabled = me.BooleanField(default=True,
49                               help_text=u'Flag indicating whether the sensor is enabled.')
50 
51 
52 class TriggerTypeDB(stormbase.StormBaseDB,
53                     stormbase.ContentPackResourceMixin,
54                     stormbase.TagsMixin):
55     """Description of a specific kind/type of a trigger. The
56        (pack, name) tuple is expected uniquely identify a trigger in
57        the namespace of all triggers provided by a specific trigger_source.
58     Attribute:
59         pack - Name of the content pack this trigger belongs to.
60         trigger_source: Source that owns this trigger type.
61         payload_info: Meta information of the expected payload.
62     """
63     name = me.StringField(required=True)
64     pack = me.StringField(required=True, unique_with='name')
65     payload_schema = me.DictField()
66     parameters_schema = me.DictField(default={})
67 
68     meta = {
69         'indexes': stormbase.TagsMixin.get_indices()
70     }
71 
72 
73 class TriggerDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):
74     """
75     Attribute:
76         pack - Name of the content pack this trigger belongs to.
77         type - Reference to the TriggerType object.
78         parameters - Trigger parameters.
79     """
80     name = me.StringField(required=True)
81     pack = me.StringField(required=True, unique_with='name')
82     type = me.StringField()
83     parameters = me.DictField()
84 
85 
86 class TriggerInstanceDB(stormbase.StormFoundationDB):
87     """An instance or occurrence of a type of Trigger.
88     Attribute:
89         trigger: Reference to the Trigger object.
90         payload (dict): payload specific to the occurrence.
91         occurrence_time (datetime): time of occurrence of the trigger.
92     """
93     trigger = me.StringField()
94     payload = me.DictField()
95     occurrence_time = me.DateTimeField()
96 
97 
98 class ActionExecutionSpecDB(me.EmbeddedDocument):
99     ref = me.StringField(required=True, unique=False)
100     parameters = me.DictField()
101 
102     def __str__(self):
103         result = []
104         result.append('ActionExecutionSpecDB@')
105         result.append(str(id(self)))
106         result.append('(ref="%s", ' % self.ref)
107         result.append('parameters="%s")' % self.parameters)
108         return ''.join(result)
109 
110 
111 class RuleDB(stormbase.StormBaseDB, stormbase.TagsMixin):
112     """Specifies the action to invoke on the occurrence of a Trigger. It
113     also includes the transformation to perform to match the impedance
114     between the payload of a TriggerInstance and input of a action.
115     Attribute:
116         trigger: Trigger that trips this rule.
117         criteria:
118         action: Action to execute when the rule is tripped.
119         status: enabled or disabled. If disabled occurrence of the trigger
120         does not lead to execution of a action and vice-versa.
121     """
122     trigger = me.StringField()
123     criteria = stormbase.EscapedDictField()
124     action = me.EmbeddedDocumentField(ActionExecutionSpecDB)
125     enabled = me.BooleanField(required=True, default=True,
126                               help_text=u'Flag indicating whether the rule is enabled.')
127 
128     meta = {
129         'indexes': stormbase.TagsMixin.get_indices()
130     }
131 
132 # specialized access objects
133 sensor_type_access = MongoDBAccess(SensorTypeDB)
134 triggertype_access = MongoDBAccess(TriggerTypeDB)
135 trigger_access = MongoDBAccess(TriggerDB)
136 triggerinstance_access = MongoDBAccess(TriggerInstanceDB)
137 rule_access = MongoDBAccess(RuleDB)
138 
139 MODELS = [SensorTypeDB, TriggerTypeDB, TriggerDB, TriggerInstanceDB, RuleDB]
140
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/st2actions/st2actions/resultstracker.py b/st2actions/st2actions/resultstracker.py
--- a/st2actions/st2actions/resultstracker.py
+++ b/st2actions/st2actions/resultstracker.py
@@ -49,7 +49,7 @@
 
     def wait(self):
         super(ResultsTracker, self).wait()
-        for thread in self._query_threads():
+        for thread in self._query_threads:
             thread.wait()
 
     def shutdown(self):
diff --git a/st2common/st2common/models/db/reactor.py b/st2common/st2common/models/db/reactor.py
--- a/st2common/st2common/models/db/reactor.py
+++ b/st2common/st2common/models/db/reactor.py
@@ -91,7 +91,7 @@
         occurrence_time (datetime): time of occurrence of the trigger.
     """
     trigger = me.StringField()
-    payload = me.DictField()
+    payload = stormbase.EscapedDictField()
     occurrence_time = me.DateTimeField()

{"golden_diff": "diff --git a/st2actions/st2actions/resultstracker.py b/st2actions/st2actions/resultstracker.py\n--- a/st2actions/st2actions/resultstracker.py\n+++ b/st2actions/st2actions/resultstracker.py\n@@ -49,7 +49,7 @@\n \n     def wait(self):\n         super(ResultsTracker, self).wait()\n-        for thread in self._query_threads():\n+        for thread in self._query_threads:\n             thread.wait()\n \n     def shutdown(self):\ndiff --git a/st2common/st2common/models/db/reactor.py b/st2common/st2common/models/db/reactor.py\n--- a/st2common/st2common/models/db/reactor.py\n+++ b/st2common/st2common/models/db/reactor.py\n@@ -91,7 +91,7 @@\n         occurrence_time (datetime): time of occurrence of the trigger.\n     \"\"\"\n     trigger = me.StringField()\n-    payload = me.DictField()\n+    payload = stormbase.EscapedDictField()\n     occurrence_time = me.DateTimeField()\n", "issue": "Custom webhook fails action depending on payload.\nHi there, I've faced a weird issue.\n\nSay it we have a webhook action like this: https://gist.github.com/dennybaa/00d1936af2f441cfff15\nIf I issue the following commands:\n1. `curl -X POST --data \"{\\\"message\\\": \\\"6fd781b849bb\\\"}\" http://172.17.0.8:9101/v1/webhooks/sample`\n2. `curl -X POST --data \"{\\\"hits.rate_15m\\\": 0.4765927732670934, \\\"hits.count\\\": 30, \\\"tags\\\": [\\\"metric\\\"], \\\"@timestamp\\\": \\\"2015-04-23T13:58:28.319Z\\\", \\\"hits.rate_5m\\\": 0.31123263029226506, \\\"@version\\\": \\\"1\\\", \\\"message\\\": \\\"6fd781b849bb\\\", \\\"hits.rate_1m\\\": 0.08765952620175455}\"`\n\nIn both cases trigger is dispatched. But the action takes place only in the #1 case.\n\n", "before_files": [{"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport eventlet\nimport importlib\nimport six\n\nfrom collections import defaultdict\nfrom kombu import Connection\nfrom oslo.config import cfg\n\nfrom st2actions.query.base import QueryContext\nfrom st2common import log as logging\nfrom st2common.models.db import action as action_models\nfrom st2common.persistence.action import ActionExecutionState\nfrom st2common.transport import actionexecutionstate, consumers, publishers\n\n\nLOG = logging.getLogger(__name__)\n\nACTIONSTATE_WORK_Q = actionexecutionstate.get_queue('st2.resultstracker.work',\n                                                    routing_key=publishers.CREATE_RK)\n\n\nclass ResultsTracker(consumers.MessageHandler):\n    message_type = action_models.ActionExecutionStateDB\n\n    def __init__(self, connection, queues):\n        super(ResultsTracker, self).__init__(connection, queues)\n        self._queriers = {}\n        self._query_threads = []\n        self._failed_imports = set()\n\n    def start(self, wait=False):\n        self._bootstrap()\n        super(ResultsTracker, self).start(wait=wait)\n\n    def wait(self):\n        super(ResultsTracker, self).wait()\n        for thread in self._query_threads():\n            thread.wait()\n\n    def shutdown(self):\n        super(ResultsTracker, self).shutdown()\n        LOG.info('Stats from queriers:')\n        self._print_stats()\n\n    def _print_stats(self):\n        for name, querier in six.iteritems(self._queriers):\n            if querier:\n                querier.print_stats()\n\n    def _bootstrap(self):\n        all_states = ActionExecutionState.get_all()\n        LOG.info('Found %d pending states in db.' % len(all_states))\n\n        query_contexts_dict = defaultdict(list)\n        for state_db in all_states:\n            try:\n                context = QueryContext.from_model(state_db)\n            except:\n                LOG.exception('Invalid state object: %s', state_db)\n                continue\n            query_module_name = state_db.query_module\n            querier = self.get_querier(query_module_name)\n\n            if querier is not None:\n                query_contexts_dict[querier].append(context)\n\n        for querier, contexts in six.iteritems(query_contexts_dict):\n            LOG.info('Found %d pending actions for query module %s', len(contexts), querier)\n            querier.add_queries(query_contexts=contexts)\n\n    def process(self, query_context):\n        querier = self.get_querier(query_context.query_module)\n        context = QueryContext.from_model(query_context)\n        querier.add_queries(query_contexts=[context])\n        return\n\n    def get_querier(self, query_module_name):\n        if (query_module_name not in self._queriers and\n                query_module_name not in self._failed_imports):\n            try:\n                query_module = self._import_query_module(query_module_name)\n            except:\n                LOG.exception('Failed importing query module: %s', query_module_name)\n                self._failed_imports.add(query_module_name)\n                self._queriers[query_module_name] = None\n            else:\n                querier = query_module.get_instance()\n                self._queriers[query_module_name] = querier\n                self._query_threads.append(eventlet.spawn(querier.start))\n\n        return self._queriers[query_module_name]\n\n    def _import_query_module(self, module_name):\n        return importlib.import_module(module_name, package=None)\n\n\ndef get_tracker():\n    with Connection(cfg.CONF.messaging.url) as conn:\n        return ResultsTracker(conn, [ACTIONSTATE_WORK_Q])\n", "path": "st2actions/st2actions/resultstracker.py"}, {"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport mongoengine as me\nfrom st2common.models.db import MongoDBAccess\nfrom st2common.models.db import stormbase\n\n__all__ = [\n    'SensorTypeDB',\n    'TriggerTypeDB',\n    'TriggerDB',\n    'TriggerInstanceDB',\n    'ActionExecutionSpecDB',\n    'RuleDB'\n]\n\n\nclass SensorTypeDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):\n    \"\"\"\n    Description of a specific type of a sensor (think of it as a sensor\n    template).\n\n    Attribute:\n        pack - Name of the content pack this sensor belongs to.\n        artifact_uri - URI to the artifact file.\n        entry_point - Full path to the sensor entry point (e.g. module.foo.ClassSensor).\n        trigger_type - A list of references to the TriggerTypeDB objects exposed by this sensor.\n        poll_interval - Poll interval for this sensor.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    artifact_uri = me.StringField()\n    entry_point = me.StringField()\n    trigger_types = me.ListField(field=me.StringField())\n    poll_interval = me.IntField()\n    enabled = me.BooleanField(default=True,\n                              help_text=u'Flag indicating whether the sensor is enabled.')\n\n\nclass TriggerTypeDB(stormbase.StormBaseDB,\n                    stormbase.ContentPackResourceMixin,\n                    stormbase.TagsMixin):\n    \"\"\"Description of a specific kind/type of a trigger. The\n       (pack, name) tuple is expected uniquely identify a trigger in\n       the namespace of all triggers provided by a specific trigger_source.\n    Attribute:\n        pack - Name of the content pack this trigger belongs to.\n        trigger_source: Source that owns this trigger type.\n        payload_info: Meta information of the expected payload.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    payload_schema = me.DictField()\n    parameters_schema = me.DictField(default={})\n\n    meta = {\n        'indexes': stormbase.TagsMixin.get_indices()\n    }\n\n\nclass TriggerDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):\n    \"\"\"\n    Attribute:\n        pack - Name of the content pack this trigger belongs to.\n        type - Reference to the TriggerType object.\n        parameters - Trigger parameters.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    type = me.StringField()\n    parameters = me.DictField()\n\n\nclass TriggerInstanceDB(stormbase.StormFoundationDB):\n    \"\"\"An instance or occurrence of a type of Trigger.\n    Attribute:\n        trigger: Reference to the Trigger object.\n        payload (dict): payload specific to the occurrence.\n        occurrence_time (datetime): time of occurrence of the trigger.\n    \"\"\"\n    trigger = me.StringField()\n    payload = me.DictField()\n    occurrence_time = me.DateTimeField()\n\n\nclass ActionExecutionSpecDB(me.EmbeddedDocument):\n    ref = me.StringField(required=True, unique=False)\n    parameters = me.DictField()\n\n    def __str__(self):\n        result = []\n        result.append('ActionExecutionSpecDB@')\n        result.append(str(id(self)))\n        result.append('(ref=\"%s\", ' % self.ref)\n        result.append('parameters=\"%s\")' % self.parameters)\n        return ''.join(result)\n\n\nclass RuleDB(stormbase.StormBaseDB, stormbase.TagsMixin):\n    \"\"\"Specifies the action to invoke on the occurrence of a Trigger. It\n    also includes the transformation to perform to match the impedance\n    between the payload of a TriggerInstance and input of a action.\n    Attribute:\n        trigger: Trigger that trips this rule.\n        criteria:\n        action: Action to execute when the rule is tripped.\n        status: enabled or disabled. If disabled occurrence of the trigger\n        does not lead to execution of a action and vice-versa.\n    \"\"\"\n    trigger = me.StringField()\n    criteria = stormbase.EscapedDictField()\n    action = me.EmbeddedDocumentField(ActionExecutionSpecDB)\n    enabled = me.BooleanField(required=True, default=True,\n                              help_text=u'Flag indicating whether the rule is enabled.')\n\n    meta = {\n        'indexes': stormbase.TagsMixin.get_indices()\n    }\n\n# specialized access objects\nsensor_type_access = MongoDBAccess(SensorTypeDB)\ntriggertype_access = MongoDBAccess(TriggerTypeDB)\ntrigger_access = MongoDBAccess(TriggerDB)\ntriggerinstance_access = MongoDBAccess(TriggerInstanceDB)\nrule_access = MongoDBAccess(RuleDB)\n\nMODELS = [SensorTypeDB, TriggerTypeDB, TriggerDB, TriggerInstanceDB, RuleDB]\n", "path": "st2common/st2common/models/db/reactor.py"}], "after_files": [{"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport eventlet\nimport importlib\nimport six\n\nfrom collections import defaultdict\nfrom kombu import Connection\nfrom oslo.config import cfg\n\nfrom st2actions.query.base import QueryContext\nfrom st2common import log as logging\nfrom st2common.models.db import action as action_models\nfrom st2common.persistence.action import ActionExecutionState\nfrom st2common.transport import actionexecutionstate, consumers, publishers\n\n\nLOG = logging.getLogger(__name__)\n\nACTIONSTATE_WORK_Q = actionexecutionstate.get_queue('st2.resultstracker.work',\n                                                    routing_key=publishers.CREATE_RK)\n\n\nclass ResultsTracker(consumers.MessageHandler):\n    message_type = action_models.ActionExecutionStateDB\n\n    def __init__(self, connection, queues):\n        super(ResultsTracker, self).__init__(connection, queues)\n        self._queriers = {}\n        self._query_threads = []\n        self._failed_imports = set()\n\n    def start(self, wait=False):\n        self._bootstrap()\n        super(ResultsTracker, self).start(wait=wait)\n\n    def wait(self):\n        super(ResultsTracker, self).wait()\n        for thread in self._query_threads:\n            thread.wait()\n\n    def shutdown(self):\n        super(ResultsTracker, self).shutdown()\n        LOG.info('Stats from queriers:')\n        self._print_stats()\n\n    def _print_stats(self):\n        for name, querier in six.iteritems(self._queriers):\n            if querier:\n                querier.print_stats()\n\n    def _bootstrap(self):\n        all_states = ActionExecutionState.get_all()\n        LOG.info('Found %d pending states in db.' % len(all_states))\n\n        query_contexts_dict = defaultdict(list)\n        for state_db in all_states:\n            try:\n                context = QueryContext.from_model(state_db)\n            except:\n                LOG.exception('Invalid state object: %s', state_db)\n                continue\n            query_module_name = state_db.query_module\n            querier = self.get_querier(query_module_name)\n\n            if querier is not None:\n                query_contexts_dict[querier].append(context)\n\n        for querier, contexts in six.iteritems(query_contexts_dict):\n            LOG.info('Found %d pending actions for query module %s', len(contexts), querier)\n            querier.add_queries(query_contexts=contexts)\n\n    def process(self, query_context):\n        querier = self.get_querier(query_context.query_module)\n        context = QueryContext.from_model(query_context)\n        querier.add_queries(query_contexts=[context])\n        return\n\n    def get_querier(self, query_module_name):\n        if (query_module_name not in self._queriers and\n                query_module_name not in self._failed_imports):\n            try:\n                query_module = self._import_query_module(query_module_name)\n            except:\n                LOG.exception('Failed importing query module: %s', query_module_name)\n                self._failed_imports.add(query_module_name)\n                self._queriers[query_module_name] = None\n            else:\n                querier = query_module.get_instance()\n                self._queriers[query_module_name] = querier\n                self._query_threads.append(eventlet.spawn(querier.start))\n\n        return self._queriers[query_module_name]\n\n    def _import_query_module(self, module_name):\n        return importlib.import_module(module_name, package=None)\n\n\ndef get_tracker():\n    with Connection(cfg.CONF.messaging.url) as conn:\n        return ResultsTracker(conn, [ACTIONSTATE_WORK_Q])\n", "path": "st2actions/st2actions/resultstracker.py"}, {"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport mongoengine as me\nfrom st2common.models.db import MongoDBAccess\nfrom st2common.models.db import stormbase\n\n__all__ = [\n    'SensorTypeDB',\n    'TriggerTypeDB',\n    'TriggerDB',\n    'TriggerInstanceDB',\n    'ActionExecutionSpecDB',\n    'RuleDB'\n]\n\n\nclass SensorTypeDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):\n    \"\"\"\n    Description of a specific type of a sensor (think of it as a sensor\n    template).\n\n    Attribute:\n        pack - Name of the content pack this sensor belongs to.\n        artifact_uri - URI to the artifact file.\n        entry_point - Full path to the sensor entry point (e.g. module.foo.ClassSensor).\n        trigger_type - A list of references to the TriggerTypeDB objects exposed by this sensor.\n        poll_interval - Poll interval for this sensor.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    artifact_uri = me.StringField()\n    entry_point = me.StringField()\n    trigger_types = me.ListField(field=me.StringField())\n    poll_interval = me.IntField()\n    enabled = me.BooleanField(default=True,\n                              help_text=u'Flag indicating whether the sensor is enabled.')\n\n\nclass TriggerTypeDB(stormbase.StormBaseDB,\n                    stormbase.ContentPackResourceMixin,\n                    stormbase.TagsMixin):\n    \"\"\"Description of a specific kind/type of a trigger. The\n       (pack, name) tuple is expected uniquely identify a trigger in\n       the namespace of all triggers provided by a specific trigger_source.\n    Attribute:\n        pack - Name of the content pack this trigger belongs to.\n        trigger_source: Source that owns this trigger type.\n        payload_info: Meta information of the expected payload.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    payload_schema = me.DictField()\n    parameters_schema = me.DictField(default={})\n\n    meta = {\n        'indexes': stormbase.TagsMixin.get_indices()\n    }\n\n\nclass TriggerDB(stormbase.StormBaseDB, stormbase.ContentPackResourceMixin):\n    \"\"\"\n    Attribute:\n        pack - Name of the content pack this trigger belongs to.\n        type - Reference to the TriggerType object.\n        parameters - Trigger parameters.\n    \"\"\"\n    name = me.StringField(required=True)\n    pack = me.StringField(required=True, unique_with='name')\n    type = me.StringField()\n    parameters = me.DictField()\n\n\nclass TriggerInstanceDB(stormbase.StormFoundationDB):\n    \"\"\"An instance or occurrence of a type of Trigger.\n    Attribute:\n        trigger: Reference to the Trigger object.\n        payload (dict): payload specific to the occurrence.\n        occurrence_time (datetime): time of occurrence of the trigger.\n    \"\"\"\n    trigger = me.StringField()\n    payload = stormbase.EscapedDictField()\n    occurrence_time = me.DateTimeField()\n\n\nclass ActionExecutionSpecDB(me.EmbeddedDocument):\n    ref = me.StringField(required=True, unique=False)\n    parameters = me.DictField()\n\n    def __str__(self):\n        result = []\n        result.append('ActionExecutionSpecDB@')\n        result.append(str(id(self)))\n        result.append('(ref=\"%s\", ' % self.ref)\n        result.append('parameters=\"%s\")' % self.parameters)\n        return ''.join(result)\n\n\nclass RuleDB(stormbase.StormBaseDB, stormbase.TagsMixin):\n    \"\"\"Specifies the action to invoke on the occurrence of a Trigger. It\n    also includes the transformation to perform to match the impedance\n    between the payload of a TriggerInstance and input of a action.\n    Attribute:\n        trigger: Trigger that trips this rule.\n        criteria:\n        action: Action to execute when the rule is tripped.\n        status: enabled or disabled. If disabled occurrence of the trigger\n        does not lead to execution of a action and vice-versa.\n    \"\"\"\n    trigger = me.StringField()\n    criteria = stormbase.EscapedDictField()\n    action = me.EmbeddedDocumentField(ActionExecutionSpecDB)\n    enabled = me.BooleanField(required=True, default=True,\n                              help_text=u'Flag indicating whether the rule is enabled.')\n\n    meta = {\n        'indexes': stormbase.TagsMixin.get_indices()\n    }\n\n# specialized access objects\nsensor_type_access = MongoDBAccess(SensorTypeDB)\ntriggertype_access = MongoDBAccess(TriggerTypeDB)\ntrigger_access = MongoDBAccess(TriggerDB)\ntriggerinstance_access = MongoDBAccess(TriggerInstanceDB)\nrule_access = MongoDBAccess(RuleDB)\n\nMODELS = [SensorTypeDB, TriggerTypeDB, TriggerDB, TriggerInstanceDB, RuleDB]\n", "path": "st2common/st2common/models/db/reactor.py"}]}