StackStorm faster with-items

StackStorm faster with-items

Orquesta is the workflow engine of StackStorm.
with-items is a construct used in Orquesta workflows that allows iterating over a list much like a for each loop.

Problem

In StackStorm, the service that runs these workflows is called st2workflow. Every step in the workflow st2workflow needs to read the workflow definition, the spec, and the current state of the workflow, the context. When a step completes st2workflow gets woken up and reads the current workflow spec and context from the database, MongoDB, then modifies it and writes it back to the database.
When using with-items for large lists you may notice a slowdown in your workflow processing. This is because every iteration of the workflow increases the size of the context that is written back to the database. MongoDB is slower at writing data due to the indexing that takes place. Eventually the st2workflow service will become bogged down by the time it takes to read and write these workflow contexts, you may see transitions taking up to several minutes between iterations.

Solution

The root of the problem is the growing workflow context. We solve this by preventing the context from having information added for each item processed in the with-items loop. This is accomplished by executing a Python action that calls back into the st2api and invokes each item’s action individually.
We have seen iterations that would have taken a week or more to finish instead finish in a an hour or two hours.
Below is the code that implements this solution.

First, we create a new python-script action that will handle creating invoking the individual item actions. The action’s metadata yaml file is as so:

---
name: "with_items"
runner_type: "python-script"
description: "faster with items;"
enabled: true
entry_point: "with_items.py"
parameters:
  action:
    type: string
    required: true
  parameters:
    type: array
    items:
      type: object
    description: "list of objects for parameters"
  paging_limit:
    type: integer
    default: 5
  sleep_time:
    type: integer
    default: 25
  result:
    type: string
    description: "jinja string to {_{ _.result.xxx }_} portion of result to return"
output_schema:
  results:
    type: "array"
    items:
      type: "object"

The python action file:

import copy
import jinja2
import re
import os
import six
import time
import urllib3

from st2client.models import LiveAction
from st2client.client import Client
from st2client.commands import action as st2action
from st2common.runners.base_action import Action

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

JINJA_REGEXP = '({{(.*?)}})'

class WithItemsAction(Action):

    def __init__(self, config):
        super(WithItemsAction, self).__init__(config)
        self.jinja_env = jinja2.Environment()
        self.jinja_pattern = re.compile(JINJA_REGEXP)
        self.api_key = self.config.get("st2api_key", None)

        if self.api_key is not None:
            if os.environ.get("ST2_AUTH_TOKEN", None) is not None:
                del os.environ["ST2_AUTH_TOKEN"]
        self.client = Client(base_url=config["st2baseurl"],
                auth_url=config["st2authurl"],
                api_key=self.api_key,
                api_url=config["st2apiurl"]
                )

    def unescape_jinja(self, expr):
        if isinstance(expr, six.string_types):
            return self.unescape_jinja_str(expr)
        elif isinstance(expr, list):
            return self.unescape_jinja_list(expr)
        elif isinstance(expr, dict):
            return self.unescape_jinja_dict(expr)
        else:
            raise TypeError("Unable to escape jinja expression for type: {var}".format(var=str(type(expr))))

    def unescape_jinja_str(self, expr_str):
        expr_str = expr_str.replace("{_{", "{{")
        expr_str = expr_str.replace("}_}", "}}")
        return expr_str

    def unescape_jinja_list(self, expr_list):
        return [self.unescape_jinja(expr) for expr in expr_list ]

    def unescape_jinja_dict(self, expr_dict):
        for k, v in six.iteritems(expr_dict):
            expr_dict[k] = self.unescape_jinja(v)
        return expr_dict

    def render_jinja(self, context, expr):
        if isinstance(expr, six.string_types):
            return self.render_jinja_str(context, expr)
        elif isinstance(expr, list):
            return self.render_jinja_list(context, expr)
        elif isinstance(expr, dict):
            return self.render_jinja_dict(context, expr)
        else:
            raise TypeError("Unable to render Jinja expression for type: {}".format(type(expr)))

    def render_jinja_str(self, context, expr_str):
        # find all of the jinja patterns in expr_str
        patterns = self.jinja_pattern.findall(expr_str)

        # if the matched pattern matches the full expression, then pull out
        # the first group [0][1] which is the content between the {{ }}
        # then use this special rendering method
        if patterns[0][0] == expr_str:
            # we only have a single pattern, render it so that a native type
            # will be returned
            func = self.jinja_env.compile_expression(patterns[0][1], expr_str)
            return func(**context)
        else:
            # we have multiple patterns in one string so rendering it
            # "normallY" and this will return a string
            template = self.jinja_env.from_string(expr_str)
            return template.render(context)

    def render_jinja_list(self, context, expr_list):
        return [self.render_jinja(context, expr) for expr in expr_list]

    def render_jinja_dict(self, context, expr_dict):
        rendered = {}
        for k, expr in six.iteritems(expr_dict):
            rendered[k] = self.render_jinja(context, expr)
        return rendered

    def run(self, action, parameters, paging_limit, sleep_time, result=None, **kwargs):
        # unescape jinja
        result_expr = result
        if result_expr:
            result_expr = self.unescape_jinja(result_expr)

        running_ids = []
        finished = []
        param_index = 0
        success = True
        while len(finished) < len(parameters):
            # fill paging pool
            while len(running_ids) < paging_limit and \
                    len(parameters) != len(running_ids) + len(finished) : # end of list no more to page
                self.logger.debug("creating action index: " + str(param_index))
                execution = self.client.executions.create(
                    LiveAction(action=action,
                               parameters=parameters[param_index]))
                running_ids.append(execution.id)
                param_index+=1

            # sleep and wait for some work to be done
            time.sleep(sleep_time)

            # check for completed or failures
            execution_list = self.client.liveactions.query(id=",".join(running_ids))
            for e_updated in execution_list:
                if e_updated.status in st2action.LIVEACTION_COMPLETED_STATES:
                    if e_updated.status != st2action.LIVEACTION_STATUS_SUCCEEDED:
                        success = False
                    self.logger.debug("finished execution: " + str(e_updated.id))
                    running_ids.remove(e_updated.id)
                    finished.append(e_updated)

        # extract the information out of the output
        outputs = []
        for exe in finished:
            outputs.append({"status": copy.deepcopy(exe.status),
                            "result": copy.deepcopy(exe.__dict__["result"])})
        results = []
        if result_expr:
            for output in outputs:
                result = output["result"]
                result_context = {"_": {"result": result}}
                result = self.render_jinja(result_context, result_expr)
                results.append(result)
        else:
            results = outputs

        return (success, results)

Installation

st2 pack install [path]
st2 pack config with_items

Example usage

The input parameters must be fed via a list of dicts. One dict for each iteration. The YAQL select function for lists is very handy to generate the parameters.

version: 1.0 

description: A with items example

input:
  - list_items

tasks:
  task1:
    action: withitems.with_items
    input:
      action: core.echo
      parameters: "<% ctx().list_items.select({\"message\"=>concat(\"message \", $)}) %>"
    next:
      - when: <% succeeded() %>
        do: task2
  task2:
    action: core.noop

Gotchas

Do not use if your sub-action has inquiries. The inquires that pause the workflow will cause this action’s token to not be valid after your token timeout period. Also this workflow will appear as running when it is in fact paused.
Set the api key to prevent token timeouts.