forked from StackStorm/st2
-
Notifications
You must be signed in to change notification settings - Fork 0
/
st2-analyze-links.py
190 lines (160 loc) · 5.86 KB
/
st2-analyze-links.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
#!/usr/bin/env python
# Copyright 2020 The StackStorm Authors.
# Copyright 2019 Extreme Networks, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Visualize the links created by rules.
1. requires graphviz
pip install graphviz
apt-get install graphviz
To run :
./st2-analyze-links.py --action_ref <action-ref>
The command must run on a StackStorm box.
"""
from __future__ import print_function
from __future__ import absolute_import
import os
from oslo_config import cfg
from st2common import config
from st2common.util.monkey_patch import monkey_patch
from st2common.persistence.rule import Rule
from st2common.service_setup import db_setup
try:
from graphviz import Digraph
except ImportError:
msg = (
'Missing "graphviz" dependency. You can install it using pip: \n'
"pip install graphviz"
)
raise ImportError(msg)
def do_register_cli_opts(opts, ignore_errors=False):
for opt in opts:
try:
cfg.CONF.register_cli_opt(opt)
except:
if not ignore_errors:
raise
class RuleLink(object):
def __init__(self, source_action_ref, rule_ref, dest_action_ref):
self._source_action_ref = source_action_ref
self._rule_ref = rule_ref
self._dest_action_ref = dest_action_ref
def __str__(self):
return "(%s -> %s -> %s)" % (
self._source_action_ref,
self._rule_ref,
self._dest_action_ref,
)
class LinksAnalyzer(object):
def __init__(self):
self._rule_link_by_action_ref = {}
self._rules = {}
def analyze(self, root_action_ref, link_tigger_ref):
rules = Rule.query(trigger=link_tigger_ref, enabled=True)
# pprint.pprint([rule.ref for rule in rules])
for rule in rules:
source_action_ref = self._get_source_action_ref(rule)
if not source_action_ref:
print("No source_action_ref for rule %s" % rule.ref)
continue
rule_links = self._rules.get(source_action_ref, None)
if rule_links is None:
rule_links = []
self._rules[source_action_ref] = rule_links
rule_links.append(
RuleLink(
source_action_ref=source_action_ref,
rule_ref=rule.ref,
dest_action_ref=rule.action.ref,
)
)
analyzed = self._do_analyze(action_ref=root_action_ref)
for (depth, rule_link) in analyzed:
print("%s%s" % (" " * depth, rule_link))
return analyzed
def _get_source_action_ref(self, rule):
criteria = rule.criteria
source_action_ref = criteria.get("trigger.action_name", None)
if not source_action_ref:
source_action_ref = criteria.get("trigger.action_ref", None)
return source_action_ref["pattern"] if source_action_ref else None
def _do_analyze(self, action_ref, rule_links=None, processed=None, depth=0):
if processed is None:
processed = set()
if rule_links is None:
rule_links = []
processed.add(action_ref)
for rule_link in self._rules.get(action_ref, []):
rule_links.append((depth, rule_link))
if rule_link._dest_action_ref in processed:
continue
self._do_analyze(
rule_link._dest_action_ref,
rule_links=rule_links,
processed=processed,
depth=depth + 1,
)
return rule_links
class Grapher(object):
def generate_graph(self, rule_links, out_file):
graph_label = "Rule based visualizer"
graph_attr = {
"rankdir": "TD",
"labelloc": "t",
"fontsize": "15",
"label": graph_label,
}
node_attr = {}
dot = Digraph(
comment="Rule based links visualization",
node_attr=node_attr,
graph_attr=graph_attr,
format="png",
)
nodes = set()
for _, rule_link in rule_links:
print(rule_link._source_action_ref)
if rule_link._source_action_ref not in nodes:
nodes.add(rule_link._source_action_ref)
dot.add_node(rule_link._source_action_ref)
if rule_link._dest_action_ref not in nodes:
nodes.add(rule_link._dest_action_ref)
dot.add_node(rule_link._dest_action_ref)
dot.add_edge(
rule_link._source_action_ref,
rule_link._dest_action_ref,
constraint="true",
label=rule_link._rule_ref,
)
output_path = os.path.join(os.getcwd(), out_file)
dot.format = "png"
dot.render(output_path)
def main():
monkey_patch()
cli_opts = [
cfg.StrOpt("action_ref", default=None, help="Root action to begin analysis."),
cfg.StrOpt(
"link_trigger_ref",
default="core.st2.generic.actiontrigger",
help="Root action to begin analysis.",
),
cfg.StrOpt("out_file", default="pipeline"),
]
do_register_cli_opts(cli_opts)
config.parse_args()
db_setup()
rule_links = LinksAnalyzer().analyze(cfg.CONF.action_ref, cfg.CONF.link_trigger_ref)
Grapher().generate_graph(rule_links, cfg.CONF.out_file)
if __name__ == "__main__":
main()