Skip to content

Designing Workflows

Jump to bottom
Miro Kubicek edited this page Jan 22, 2021 · 16 revisions

Designing Workflows

Workflow Attributes

:suspendable?

A boolean - when set to true any job instance of the workflow can be suspended via a corresponding API call (a PATCH call to /systems/:system/jobs/:jobid). Same effect also has adding a :suspendable? property to job properties. The paused job will be sent to a predefined "suspended" channel (usually same as for archival) and stored into the archived. From there it then can be resumed. Suspension (pausing) of multiple parallel job threads (when using fork/join) is supported.

All corresponding use cases and race conditions should be handled well including a job suspension while the main job thread is performing join; non-main threads still running or finished etc.

Designing Steps

A step is a node in a workflow graph - it can have its own properties and it consists of a workload function that is to be executed when the step is run. The workflow function is executed either for its side effects (e.g. sending an email) or for the value(s) it is supposed to return (e.g. load customer data from CRM system).

{:name "hello-world"
 :revision 5
 :properties {:name "World"}
 :steps [{:id "clojure-hello-world"
          :type :custom
          :supertype :tasklet
          :next []
          :workload-fn #titanoboa.exp/Expression{:value "(fn [p]   {\"greeting\" (str \"Hello \"  (:name p) \"!\")})"
                                                 :type "clojure"}
          :properties {}}]}

Step Attributes

:id

A String that contains unique name of the step.

:supertype

There can be couple of different step supertypes (denoted by :supertype key in step definition):

  • :tasklet - basic workflow step executed just for its side effects or return value
  • :join - same as tasklet, but also serves as rendezvous point for branches of the workflow that were being executed in parallel
  • :map - based on a sequence returned by this step's workload function, many separate atomic jobs are created
  • :reduce - performs reduce function over results returned by jobs triggered by a map step

:type

Apart from :supertype there is also :type attribute for each step. Currently is is just being used by titanoboa GUI for step visualization (to pick a corresponding icon) and also you can use it there to pick from ready-made step templates - but during job processing it is mostly ignored. Outside of GUI its purpose is mainly just to annotate the general purpose of the step.


:workload-fn

The workload is either defined as:

  1. clojure library function that is supposed to be executed (e.g. as a package/method name that is on the classpath or in a library in specified maven repository/artifact):
:workload-fn 'titanoboa.tasklet.httpclient/request

The function has to have arity of one parameter as job properties are passed as an input onto every workload function that is executed.

  1. java class that implements io.titanoboa.java.IWorkloadFn interface. The class will be used to automatically instantiate a singleton bean (so it has to have a constructor with no argumet) and all subsequent references to it from any workflow-fn will invoke its invoke method:
:workload-fn 'io.titanoboa.java.SampleWorkloadImpl

or in GUI:

  1. clojure anonymous function code wrapped in titanoboa.exp.Expression record:
(titanoboa.exp/map->Expression {:value "(fn [p] \n  {:message (str \"Hello \" (:name p))})", :type "clojure"})

or

#titanoboa.exp.Expression{:value "(fn [p] \n  {:message (str \"Hello \" (:name p))})", :type "clojure"}
  1. java lambda function code wrapped in titanoboa.exp.Expression record:
#titanoboa.exp.Expression{:value " p -> {String greeting = (String) p.get(\"greeting\");
greeting = greeting + \" Nice to meet you!\";
java.util.HashMap propertiesToReturn = new java.util.HashMap ();
propertiesToReturn.put (\"greeting\" , greeting);
return clojure.lang.PersistentArrayMap.create(propertiesToReturn);
}", :type "java"}

Though anonymous code (options 3. & 4.) is not that usable programmatically, it is primarily meant to be used in GUI for rapid development:

Allowed workflow-fn return values

Note: please note that this applies to :tasklet steps only, :map and :reduce steps are handled differently.

Return value type Use Handling
literals return code Literal values are used as workload return value and are matched against :next step specification
clojure map clojure.lang.PersistentArrayMap properties / optional return code
  • if the map contains key :properties it is treated as returned properties that will be merged onto job's properties map;
  • if no :properties key is present the whole map is treated as returned properties;
  • if the map contains any key such as :exit :code :return-code :exit-code its value will be treated as a return code
java.util.HashMap properties / optional return code will be converted into a clojure map and treated as such

:allow-parallel?

If multiple :next steps are defined and if condition for more than one is fulfilled then setting :allow-parallel? to true will enable all these steps (and any following steps and workflow branches) to be executed in parallel. This will effectively branch the execution graph into multiple parallel workflow branches which can be then joined back by using step :supertype :join.

If :allow-parallel? is set to false and multiple match next steps are found, only the first one is used and others are ignored.


:next

Defines next steps that will be triggered upon completion of current workflow step. Next should be defined as vector of tuples in a format of ["return value of current step" "step-id of next step to be triggered"]. Return value of current step can be a literal or a wildcard ("*" or :*). String "ERROR" (irrespective of case) is reserved and such next step will be triggered in case of an exception.

[[0 "step-id"]
 ["some-string" "step-id2"]
 [false "step-id3"]
 ["*" "step-id4"]
 ["ERROR" "step-id5"]]

If "ERROR" is not defined among a step's next steps and its workload-fn throws an exception then the workflow execution will fail at that point and the workflow's end state will be error.


:properties

A map (so clojure.lang.PersistentArrayMap) of steps properties. It can contain titanoboa.exp.Expressions or clojure quotes (i.e. a list). Upon initialization of the step all expressions/quotes in its properties are evaluated and than the step's properties map is merged onto the workflow job's main properties map (so this way you can technically override values in job's properties). Resulting map is then passed onto :workload-fn as its only argument.

Clone this wiki locally