A simple orchestration library for running complex processes or workflows in Ruby. Processes are defined using a simple DSL, where the sequences and tasks are defined. Processes can then be queued for execution. Sequences can be synchronous or asynchronous, and the overall process can be monitored for completion or failure.
Processes and tasks are executed by background workers and you can use any one of the following gems:
The configuration and state of each process and their respective tasks is stored using Redis key/values.
The latest MRI 2.x or 3.x version. Other versions/VMs are untested, but might work fine. MRI 1.x is not supported.
Redis 2.4 or greater is required.
One of the following background worker queue gems: resque, sidekiq or delayed_job.
NOTE:resque or sidekiq is recommended since they use Redis as a backing store as well.
Add this line to your application's Gemfile:
gem 'taskinator'
And then execute:
$ bundle install
Or install it yourself as:
$ gem install taskinator
If you are using Taskinator within a Rails application, then add an initializer, such as
config/initializers/taskinator.rb, with the following configuration content:
# config/initializers/taskinator.rbTaskinator.configuredo |config|
# configure the queue adapter to use# can be :active_job, :delayed_job, :resque or :sidekiqconfig.queue_adapter=:resque# configure redisconfig.redis={:url=>'redis://redis.example.com:7372/12',:namespace=>'mynamespace'}endSee the configuration section below for more configuration details.
Start by creating a "process" module and extending Taskinator::Definition.
require'taskinator'moduleMyProcessextendTaskinator::DefinitionendDefine the process using the define_process method.
moduleMyProcessextendTaskinator::Definition# defines a processdefine_processdoendendThe define_process method optionally takes the list of expected arguments which are used
to validate the arguments supplied when creating a new process.
These should be specified with symbols.
moduleMyProcessextendTaskinator::Definition# defines a processdefine_process:date,:optionsdo# ...endend# when creating a process, 2 arguments are expectedprocess=MyProcess.create_processDate.today,:option_1=>trueNOTE: The current implementation performs a naive check on the count of arguments.
Next, specify the tasks with their corresponding implementation methods, that make up the
process, using the task method and providing the method to execute for the task.
moduleMyProcessextendTaskinator::Definitiondefine_processdotask:first_work_steptask:second_work_stependdeffirst_work_step# TODO: supply implementationenddefsecond_work_step# TODO: supply implementationendendMore complex processes may define sequential or concurrent steps, using the sequential
and concurrent methods respectively.
moduleMyProcessextendTaskinator::Definitiondefine_processdoconcurrentdo# these tasks will be executed concurrentlytask:work_step_1task:work_step_2endsequentialdo# thes tasks will be executed sequentiallytask:work_step_3task:work_step_4endenddefwork_step_1# TODO: supply implementationend
...
defwork_step_N# TODO: supply implementationendendYou can also define data driven tasks using the for_each method, which takes an iterator method
name as an argument.
The iterator method yields the parameters necessary for the task or job. Notice that the task method takes a parameter in this case, which will be the return values provided by the iterator.
moduleMyProcessextendTaskinator::Definitiondefine_processdofor_each:yield_data_elementsdotask:work_stependenddefyield_data_elements# TODO: supply implementation to yield elementsyield1enddefwork_step(data_element)# TODO: supply implementationendendIt is possible to branch the process logic based on the options hash passed in when creating
a process. The options? method takes the options key as an argument and calls the supplied
block if the option is present and it's value is truthy.
moduleMyProcessextendTaskinator::Definitiondefine_processdooption?(:some_setting)dotask:prerequisite_stependtask:work_stependdefprerequisite_step# ...enddefwork_step# ...endend# now when creating the process, the `:some_setting` option can be used to branch the logicprocess1=MyProcess.create_process:some_setting=>trueprocess1.tasks.count#=> 2process2=MyProcess.create_processprocess2.tasks.count#=> 1In addition, it is possible to transform the arguments used by a task or job, by including
a transform step in the definition.
Similarly for the for_each method, transform takes a method name as an argument.
The transformer method must yield the new arguments as required.
moduleMyProcessextendTaskinator::Definition# this process is created with a hash argumentdefine_processdotransform:convert_argsdotask:work_stependenddefconvert_args(options)yield *[options[:date_from],options[:date_to]]enddefwork_step(date_from,date_to)# TODO: supply implementationendendProcesses can be composed of other processes too:
moduleMySubProcessA
...
endmoduleMySubProcessB
...
endmoduleMyProcessextendTaskinator::Definitiondefine_processdosub_processMySubProcessAsub_processMySubProcessBendendAny combination or nesting of task, sequential, concurrent and for_each steps are
possible. E.g.
moduleMyProcessextendTaskinator::Definitiondefine_processdofor_each:data_elementsdotask:work_step_beginconcurrentdofor_each:sub_data_elementsdotask:work_step_all_at_onceendendsub_processMySubProcesssequentialdofor_each:sub_data_elementsdotask:work_step_one_by_oneendendtask:work_step_endendend# "task" and "iterator" methods omitted for brevityendIn this example, the work_step_begin is executed, followed by the work_step_all_at_once
steps which are executed concurrently, then the sub process MySubProcess is created and
executed, followed by the work_step_one_by_one tasks which are executed sequentially and
finally the work_step_end is executed.
It is also possible to embed conditional logic within the process definition stages in order to produce steps based on the required logic.
All builder methods are available within the scope of the define_process block. These
methods include args and options which are passed into the create_process method
of the definition.
E.g.
moduleMyProcessextendTaskinator::Definitiondefine_processdotask:task_1task:task_2task:task_3ifargs[3] == 1task:send_notificationifoptions[:send_notification]end# "task" methods are omitted for brevityend# when creating this proces, you supply to option when calling `create_process`# in this example, 'args' will be an array [1,2,3]# and options will be a Hash {:send_notification => true}MyProcess.create_process(1,2,3,:send_notification=>true)It is likely that you already have one or more jobs and want to reuse them within the process definition.
Define a job step, providing the class of the Active Job to run and then taskinator will
invoke that job as part of the process.
The job step will be queued and executed on same queue as
configured by the job.
# E.g. A resque workerclassDoSomeWorkqueue:high_prioritydefself.perform(arg1,arg2)# code to do the workendendmoduleMyProcessextendTaskinator::Definition# when creating the process, supply the same arguments# that the DoSomeWork worker expectsdefine_processdojobDoSomeWorkendendA process is created by calling the generated create_process method on your "process" module.
process=MyProcess.create_processAnd then enqueued for execution by calling the enqueue! method of the process.
process.enqueue!Or, started immediately by calling the start! method of the process.
process=MyProcess.create_processprocess.start!Argument handling for defining and executing process definitions is where things can get trickey. This may be something that gets refactored down the line.
To best understand how arguments are handled, you need to break it down into 3 phases. Namely:
- Definition,
- Creation and
- Execution
Firstly, a process definition is declarative in that the define_process and a mix of
sequential, concurrent, for_each, task and job directives provide the way to
specify the sequencing of the steps for the process.
Taskinator will interprete this definition and execute each step in the desired sequence or concurrency.
Consider the following process definition:
moduleMySimpleProcessextendTaskinator::Definition# definitiondefine_processdotask:work_step_1task:work_step_2for_each:additional_stepdotask:work_step_3endend# creationdefadditional_step(options)options.steps.eachdo |k,v|
yieldk,vendend# executiondefwork_step_1(options)# ...enddefwork_step_2(options)# ...enddefwork_step_3(k,v)# ...endendThere are three tasks; namely :work_step_1, :work_step_2 and :work_step_3.
The third task, :work_step_3, is built up using the for_each iterator, which means that
the number of :work_step_3 tasks will depend on how many times the additional_step
iterator method yields to the definition.
This brings us to the creation part. When create_process is called on the given module,
you provide arguments to it, which will get passed onto the respective task and
for_each iterator methods.
So, considering the MySimpleProcess module shown above, work_step_1, work_step_2
and work_step_3 methods each expect arguments.
These will ultimately come from the arguments passed into the create_process method.
E.g.
# Given an options hashoptions={:opt1=>true,:opt2=>false,:steps=>{:a=>1,:b=>2,:c=>3,}}# You create the process, passing in the options hashprocess=MySimpleProcess.create_process(options)To best understand how the process is created, consider the following "procedural" code for how it could work.
# A process, which maps the target and a list of stepsclassProcessattr_reader:targetattr_reader:tasksdefinitialize(target)@target=target@tasks=[]endend# A task, which maps the method to call and it's argumentsclassTaskattr_reader:methodattr_reader:argsdefinitialize(method,args)@method,@args=method,argsendend# Your module, with the methods which do the actual workmoduleMySimpleProcessdefself.work_step_1(options) ...
defself.work_step_2(options) ...
defself.work_step_3(k,v) ...
end# Now, the creation phase of the definition# create a process, providing the moduleprocess=Process.new(MySimpleProcess)# create the first and second tasks, providing the method# for the task and it's arguments, which are the options defined aboveprocess.tasks << Task.new(:work_step_1,options)process.tasks << Task.new(:work_step_2,options)# iterate over the steps hash in the options, and add the third step# this time specify the key and value as the# arguments for the work_step_3 methodoptions.steps.eachdo |k,v|
process.tasks << Task.new(:work_step_3,[k,v])end# we now have a process with the tasks definedprocess.tasks#=> [<Task :method=>work_step_1, :args=>options, ...> ,# <Task :method=>work_step_2, :args=>options, ...>,# <Task :method=>work_step_3, :args=>[:a, 1], ...>,# <Task :method=>work_step_3, :args=>[:b, 2], ...>,# <Task :method=>work_step_3, :args=>[:c, 3], ...>]Finally, for the execution phase, the process and tasks will act on the supplied module.
# building out the "Process" classclassProcess#...defexecutetasks.each{|task| task.execute(target))endend# and the "Task" classclassTask#...defexecute(target)puts"Calling '#{method}' on '#{target.name}' with #{args.inspect}..."target.send(method, *args)endend# executing the process iterates over each task and# the target modules method is called with the argumentsprocess.execute# Calling 'work_step_1' on 'MySimpleProcess' with {:opt1 => true, :opt2 => false, ...}# Calling 'work_step_2' on 'MySimpleProcess' with {:opt1 => true, :opt2 => false, ...}# Calling 'work_step_3' on 'MySimpleProcess' with [:a, 1]# Calling 'work_step_3' on 'MySimpleProcess' with [:b, 2]# Calling 'work_step_3' on 'MySimpleProcess' with [:c, 3]In reality, each task is executed by a worker process, possibly on another host, so the execution process isn't as simple, but this example should help you to understand conceptually how the process is executed, and how the arguments are propagated through.
NOTE: This aspect of the library is still a work in progress.
To monitor the state of the processes, use the Taskinator::Api::Processes class.
processes=Taskinator::Api::Processes.newprocesses.eachdo |process|
# => output the unique process identifier and current stateputs[:process,process.uuid,process.current_state]endYou can also install a web interface for your Rails application. Check https://github.com/bguban/taskinator_ui for details.
To aid debugging specific processes and tasks, where the process or task identifier is
known, it is possible to retrieve the specific task or process using Taskinator::Api.
To retrieve a specific process, given the process identifier:
process_id="SUPPLY-PROCESS-IDENTIFIER"process=Taskinator::Api.find_process(process_id)putsprocess.inspectputsprocess.definitionputsprocess.current_stateputsprocess.tasks# etc...The type of process may be one of the following:
Taskinator::Process::SequentialTaskinator::Process::Concurrent
Then, to retrieve a specific task, given the task identifier:
task_id="SUPPLY-TASK-IDENTIFIER"task=Taskinator::Api.find_task(task_id)putstask.inspectputstask.classputstask.definitionputstask.args# for Step and Job typesputstask.sub_process.tasks# for SubProcess type# etc...Depending on the type of task, different attributes will be available for inspection.
The types include:
Taskinator::Task::StepTaskinator::Task::JobTaskinator::Task::SubProcess
By default Taskinator assumes Redis is located at localhost:6397. This is fine for development,
but for many production environments you will need to point to an external Redis server.
You may also what to use a namespace for the Redis keys.
NOTE: The configuration hash must have symbolized keys.
Taskinator.configuredo |config|
# redis configurationconfig.redis={:url=>'redis://redis.example.com:7372/12',:namespace=>'mynamespace'}endOr, alternatively, via an ENV variable
Set the REDIS_PROVIDER environment variable to the Redis server url.
E.g. On Heroku, with RedisGreen: set REDIS_PROVIDER=REDISGREEN_URL and Taskinator will use the
value of the REDISGREEN_URL environment variable when connecting to Redis.
You may also use the generic REDIS_URL which may be set to your own private Redis server.
The Redis configuration leverages the same setup as sidekiq. For advanced options, checkout the
Sidekiq Advanced Options
wiki page for more information.
To configure the queue adapter to use, set config.queue_adapter to one of the following values:
:active_job:delayed_job:resque:sidekiq
As follows:
Taskinator.configuredo |config|
# configure the queue adapter to use# can be :active_job, :delayed_job, :resque or :sidekiqconfig.queue_adapter=:resqueendBy default the queue names for process and task workers is default, however, you can specify
the queue names as follows:
Taskinator.configuredo |config|
# queue configurationconfig.queue_config={:process_queue=>:default,:task_queue=>:default}endIt is possible to instrument processes, tasks and jobs by providing an instrumeter such
as ActiveSupport::Notifications.
Taskinator.configuredo |config|
# configure instrumenter to useconfig.instrumenter=ActiveSupport::NotificationsendAlternatively, you can use the built-in instrumenter for logging to the console for debugging:
Taskinator.configuredo |config|
# configure instrumenter to useconfig.instrumenter=Taskinator::ConsoleInstrumenter.newendThe following instrumentation events are issued:
| Event | When |
|---|---|
taskinator.process.created | After a root process gets created |
taskinator.process.saved | After a root process has been persisted to Redis |
taskinator.process.enqueued | After a process or subprocess is enqueued for processing |
taskinator.process.processing | When a process or subprocess is processing |
taskinator.process.paused | When a process or subprocess is paused |
taskinator.process.resumed | When a process or subprocess is resumed |
taskinator.process.completed | After a process or subprocess has completed processing |
taskinator.process.cancelled | After a process or subprocess has been cancelled |
taskinator.process.failed | After a process or subprocess has failed |
taskinator.task.enqueued | After a task has been enqueued |
taskinator.task.processing | When a task is processing |
taskinator.task.completed | After a task has completed |
taskinator.task.cancelled | After a task has been cancelled |
taskinator.task.failed | After a task has failed |
For all events, the data included contains the following information:
| Key | Value |
|---|---|
:type | The type name of the component reporting the event |
:definition | The type name of the process definition |
:process_uuid | The UUID of the root process |
:process_options | Options hash of the root process |
:uuid | The UUID of the respective task, job or sub process |
:options | Options hash of the component |
:state | State of the component |
:percentage_completed | The percentage of completed tasks |
:percentage_failed | The percentage of failed tasks |
:percentage_cancelled | The percentage of cancelled tasks |
The persistence logic is decoupled from the implementation, so it is possible to implement another backing store if required.
- Fork it
- Create your feature branch (
git checkout -b my-new-feature) - Commit your changes (
git commit -am 'Add some feature') - Push to the branch (
git push origin my-new-feature) - Create new Pull Request
MIT Copyright (c) 2014 Chris Stefano
Portions of code are from the Sidekiq project, Copyright (c) Contributed Systems LLC.
Inspired by the sidekiq and workflow gems.
For other workflow solutions, checkout Stonepath, the now deprecated ruote gem and workflow.
Alternatively, for a robust enterprise ready solution checkout the AWS Flow Framework for Ruby.