SSM Automation document graph visualization to help with branching in complex workflows

Documents/Automation/DetachEBSVolumes/Makefile

@@ -19,6 +19,10 @@ TARGET_DIR = "./Output"
 documents: targetdir createdocuments
 	@echo "Done making documents"
 
+graph: targetdir createdocuments
+	python ./Setup/create_document_graph.py > ./Output/aws-DetachEBSVolume.dot
+	@echo "Done making document graph"
+
 targetdir:
 	@echo "Making $(TARGET_DIR)"
 	mkdir -p ./Output

Documents/Automation/DetachEBSVolumes/Setup/create_document_graph.py

@@ -0,0 +1,37 @@
+#
+# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of this
+# software and associated documentation files (the "Software"), to deal in the Software
+# without restriction, including without limitation the rights to use, copy, modify,
+# merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+#
+import os
+import sys
+
+DOC_DIR = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+REPO_ROOT = os.path.dirname(DOC_DIR)
+
+# Import shared testing code
+sys.path.append(os.path.join(REPO_ROOT, 'Testing'))
+import ssm_testing  # noqa pylint: disable=import-error,wrong-import-position
+
+
+def process():
+    ssm_doc_name = "aws-DetachEBSVolume"
+    print(ssm_testing.SSMTester.convert_document_to_dot_graph(doc_filename=os.path.join(DOC_DIR,
+                                                                   'Output',
+                                                                   ('{}.json'.format(ssm_doc_name)))))
+
+
+if __name__ == '__main__':
+    process()
+

Documents/Automation/Testing/ssm_testing.py

@@ -17,6 +17,8 @@
 #!/usr/bin/env python
 """Testing support module for SSM documents."""
 
+from collections import OrderedDict
+import json
 import logging
 import time
 
@@ -146,6 +148,106 @@ def destroy(self):
         """Delete SSM document."""
         self.ssm_client.delete_document(Name=self.doc_name)
 
+    @staticmethod
+    def convert_document_to_dot_graph(doc_filename):
+        """Create a graph representation of the SSM document
+        in dot language to visualize when and how branching occurs."""
+        # Loading the document as json
+        with open(doc_filename, 'r') as jsonfile:
+            json_doc = json.load(jsonfile, object_pairs_hook=OrderedDict)
+
+        # Initializating the graph variable with the document description and the default Start and End nodes
+        graph = []
+        graph.append("// {}".format(json_doc["description"]))
+        graph.append("digraph {")
+        graph.append("    Start [label=Start]")
+        graph.append("    End [label=End]")
+
+        # If the document step does not explicitly define the next step on failure and on success,
+        # then the next step from the document will use the following variables to create the edge
+        add_edge_from_previous_step = False
+        label = ""
+        previous_step_name = ""
+
+        for index, step in enumerate(json_doc["mainSteps"]):
+            if add_edge_from_previous_step:
+                graph.append("    {} -> {} [label={}]".format(
+                    previous_step_name, step["name"], label))
+                add_edge_from_previous_step = False
+
+            # Create the edge from the Start node if this is the first node of the document
+            if index == 0:
+                graph.append("    {} -> {}".format("Start", step["name"]))
+            # Create the two edges to the End node if this is the last node of the document, then exit the loop
+            elif index == (len(json_doc["mainSteps"]) - 1):
+                graph.append("    {} -> {} [label={}]".format(step["name"], "End", "onSuccess"))
+                graph.append("    {} -> {} [label={}]".format(step["name"], "End", "onFailure"))
+                break
+
+            # If action is aws:branch, checking all choices to visualize each branch
+            if step["action"] == "aws:branch":
+                for choice in step["inputs"]["Choices"]:
+                    next_step = choice["NextStep"]
+                    del choice["NextStep"]
+                    # Removing first and last character from the choice (that removes the curly brackets),
+                    # escaping and adding a new line for each comma)
+                    label = "\"{}\"".format(json.dumps(choice)[1:-1].replace('", "','"\\l"').replace('"','\\"'))
+                    graph.append("    {} -> {} [label={}]".format(
+                        step["name"], next_step, label))
+
+                if "Default" in step["inputs"]:
+                    graph.append("    {} -> {} [label={}]".format(
+                        step["name"], step["inputs"]["Default"], "Default"))
+            else:
+                # If nextStep is used in the step, using it to create the edge,
+                # else we save the current step information to be able to create the edge when inspecting the next available step
+                if "nextStep" in step:
+                    graph.append("    {} -> {} [label={}]".format(
+                        step["name"], step["nextStep"], "onSuccess"))
+                # When isEnd is true, create an edge to the End node
+                elif "isEnd" in step:
+                    if step["isEnd"] == "true":
+                        graph.append("    {} -> {} [label={}]".format(step["name"], "End", "onSuccess"))
+                else:
+                    add_edge_from_previous_step = True
+                    label = "onSuccess"
+                    previous_step_name = step["name"]
+
+            # If onFailure is Abort or not specified, create an edge to the End node.
+            if "onFailure" in step:
+                if step["onFailure"] == "Abort":
+                    graph.append("    {} -> {} [label={} color=\"red\"]".format(
+                        step["name"], "End", "onFailure"))
+                # If onFailure is Continue, we look for nextStep,
+                # or save the current step information to be able to create the edge when inspecting the next available step
+                elif step["onFailure"] == "Continue":
+                    if "nextStep" in step:
+                        label="onFailure color=\"red\""
+                        if "isCritical" in step:
+                            if step["isCritical"] == "false":
+                                label="onFailure"
+                        graph.append("    {} -> {} [label={}]".format(
+                            step["name"], step["nextStep"], label))
+                    else:
+                        add_edge_from_previous_step = True
+                        label="onFailure color=\"red\""
+                        previous_step_name = step["name"]
+                # Lastly, retrieve the next step from onFailure directly
+                else:
+                    label="onFailure color=\"red\""
+                    if "isCritical" in step:
+                        if step["isCritical"] == "false":
+                            label="onFailure"
+                    graph.append("    {} -> {} [label={}]".format(
+                        step["name"], step["onFailure"].replace("step:", ""), label))
+            else:
+                graph.append("    {} -> {} [label={}]".format(
+                        step["name"], "End", "onFailure color=\"red\""))
+
+        graph.append("}")
+
+        return "\n".join(graph)
+
     @staticmethod
     def automation_execution_status(ssm_client, execution_id,
                                     block_on_waiting=True, status_callback=None, poll_interval=10):