Selinon concept overview¶

A system consist of flows, storages or databases and tasks. Each flow defines a directed graph (that can be even cyclic, so no DAG limitations) of well defined dependencies on nodes that compute results. A node can be either a task or another flow, so you can make flows as nested as desired. You can make decisions on when to run which nodes based on conditions that are made of predicates.

YAML configuration overview¶

Selinon is configured by easy to learn, easy to read and easy to maintain declarative configuration files written in YAML markup language.

In order to use Selinon, you have to implement tasks and define your flows in a YAML configuration file (or split it across multiple YAML configuration files).

$ pip3 install selinon

$ pip3 install selinon[celery,mongodb,postgresql,redis,s3,sentry]

from selinon import Config
from celery import Celery
from myapp.celery_settings import CelerySettings

app = Celery('tasks')
app.config_from_object(CelerySettings)

Config.set_celery_app(app)
Config.set_config_yaml('path/to/nodes.yaml', ['path/to/flow1.yaml', 'path/to/flow2.yaml'])

Naming convention¶

Imagine you defined two flows (flow1 and flow2) that consist of five tasks named Task1, Task2, Task3, Task4 and Task5. Such flows are illustrated on images bellow.

In the flow flow2 (shown above) we start node Task4 on condition that is always true (we start if Selinon was requested to start flow2). After Task4 finishes, we start (always) node Task5 which ends the flow flow2. Results of tasks are stored in the database named Storage2.

The second flow is slightly more complex. We (always) start with Task1. Task1 will transparently store results in Storage1. After Task1 finishes, Selinon (to be more precise dispatcher task) checks results of Task1 in Storage1 and if condition result['foo'] == 'bar' is evaluated as True, dispatcher starts nodes Task2 and flow2. After both Task2 and flow2 finish, dispatcher starts Task3. If the condition result['foo'] == bar (now result of Task3) is met, Task1 is started and the whole process is iteratively done again. Results of all tasks are stored in database named Storage1 except for results computed in sub-flow flow2, where Storage2 is used (see previous flow graph above).

Note that Task2 and the whole flow2 could be executed in parallel as there are no data nor time dependencies between these two nodes. Selinon runs as many nodes as possible in parallel. This makes it really easy to scale your system - the only bottleneck you can get is number of computational nodes in your cluster or limitations on storage/database side.

Flow definitions¶

Conditions¶

Conditions are made of predicates that can be nested as desired using logical operators - and, or and not. There are predefined predicates available in selinon.predicates. However you can define your own predicates based on your requirements.

These conditions are evaluated by dispatcher and if a condition is met, desired node or nodes are scheduled. If the condition is evaluated as false, destination nodes on the given edge are not run. Note that conditions are run only once only if all source nodes successfully finish.

If you do not state condition in edge definition, edge condition will be evaluated always as true.

Since there could run multiple nodes of a type (name) due to cyclic dependencies, an edge condition is evaluated for each possible combination (and only once for the given combination). If you want to avoid such behaviour, check Useful flow patterns section for possible solutions.

Starting nodes¶

You can have a single or multiple starting nodes in your flow. If you define a single starting node, the result of starting node can be propagated to other nodes as arguments if node_args_from_first is set. If you define more than one starting node, the result cannot be propagated (due to time-dependent evaluation), however you can still explicitly define arguments that are passed to the flow (or make part of your flow a sub-flow).

Flows¶

Flows can be nested as desired. The only limitation is that you cannot now inspect results of sub-flow using edge conditions in a parent flow. There is a plan to remove such limitation in next Selinon releases. Nevertheless you can still reorganize your flow (in most cases) so you are not limited with such restriction.

Running a flow¶

Once you set up Selinon and Selinon does not report any errors in your configuration files, you can run flow simply by calling the run_flow function (see documentation of run_flow()):

from selinon import run_flow

dispatcher_id = run_flow('flow1', {'foo': 'bar'})

If you wish to do selective task runs, please refer to Selective task run documentation.

Node failures¶

You can define fallback tasks and fallback flows that are run if a node fails. These fallback tasks and flows (fallback nodes) are not prone to time-dependent evaluation (to be more precise - there is no such thing in the whole Selinon design, so you can be sure that such thing does not occur on Selinon level). These fallback nodes are scheduled on task or flow failures and their aim is to recover from a failure.

Failures are propagated from sub-flows to parent flows. You can find analogy to exceptions as known in many programming languages (like in Python). If a node fails and there is no fallback node that would handle node failure, the whole flow is marked as failed. You can than capture this failure in the parent flow, but this failure will be marked as failure of the whole flow. Note that even in this case, there is no time-dependent evaluation - so if a node in your flow fails, dispatcher can still continue scheduling nodes that are not affected by the failure and once there is nothing to do more, dispatcher marks the flow as failed.

Now let’s assume that you defined two fallbacks. One waits for Task2 failure (Fallback1) and another one waits for a failure of Task1 as well as Task2 failure (Fallback2).

Let’s say that Task1 failed. In that case the decision which fallback would be run depends on Task2 failure (not on time-dependent evaluation). Fallback evaluation is greedy, so if Task2 fails, there is run Fallback2. If Task2 succeeds, Fallback1 is run.

Results of tasks¶

Results of tasks are stored in databases transparently based on your definition in YAML configuration files. The only thing you need to provide is a database adapter that handles database connection and data storing/retrieval. See storage section for more info.

YAML configuration example¶

In this section you can find YAML configuration files that were used for generating images in the previous sections. You can separate flows into multiple files, just provide flow-definitions key to find all flows defined in the YAML file.

---
  flow-definitions:
    - name: 'flow1'
      edges:
          - from:
            to:
              - 'Task1'
          - from:
              - 'Task1'
            to:
              - 'Task2'
              - 'flow2'
            condition:
                name: 'fieldEqual'
                node: 'Task1'
                args:
                    key: 'foo'
                    value: 'bar'
          - from:
              - 'Task2'
              - 'flow2'
            to:
              - 'Task3'
          - from:
              - 'Task3'
            to:
              - 'Task1'
            condition:
                name: 'argsFieldEqual'
                node: 'Task3'
                args:
                    key: 'foo'
                    value: 'bar'

---
  flow-definitions:
    - name: 'flow2'
      edges:
          - from:
            to:
              - 'Task4'
          - from:
              - 'Task4'
            to:
              - 'Task5'

Configuration for failures and failure handling fallbacks that were introduced in Node failures section can be found bellow (no storages in the example).

---
  flow-definitions:
    - name: 'exampleFallback'
      edges:
        - from:
          to: 'Task1'
        - from:
          to: 'Task2'
      failures:
        - nodes:
            - 'Task1'
            - 'Task1'
          fallback:
            - 'Fallback1'
        - nodes:
            - 'Task1'
          fallback:
            - 'Fallback2'

---
  tasks:
    - name: 'Task1'
      output_schema: 'path/to/schema1.json'
      # `classname` is omitted, it defaults to `name`
      # from worker.task1 import Task1
      import: 'worker.task1'
      storage: 'Storage1'
      # queue name to which messages will be sent
      queue: 'queue_Task1_v0'

    - name: 'Task2'
      import: 'worker.task2'
      storage: 'Storage1'
      output_schema: 'path/to/schema2.json'
      # task names are not bound to class names (you can create aliases)
      # from worker.task2 import MyTask2 as Task2
      classname: 'MyTask2'
      queue: 'queue_Task2_v1'

    - name: 'Task3'
      import: 'worker.task3'
      storage: 'Storage1'
      output_schema: 'path/to/schema3.json'
      classname: 'Task1'
      max_retry: 1
      # If queue is omitted, Celery's default queue (celery) will be used
      #queue: 'celery'

    - name: 'Task4'
      import: 'worker.task4'
      storage: 'Storage2'
      output_schema: 'path/to/schema4.json'
      classname: 'Task4'
      max_retry: 1

    - name: 'Task5'
      import: 'worker.task1'
      storage: 'Storage2'
      output_schema: 'path/to/schema1.json'
      classname: 'Task4'
      # in case of failure retry once after 10 seconds before marking node as failed
      max_retry: 1
      retry_countdown: 10


  flows:
    # state all flows you have in your system, otherwise Selinon will complain
    - 'flow1'
    - 'flow2'


  storages:
    - name: 'Storage1'
      # from storage.storage1 import MyStorage as Storage1
      # This way you can have multiple storages of a same type with different
      # configuration (different reference name)
      classname: 'MyStorage'
      import: 'storage.storage1'
      configuration: 'put your configuration for Storage1 here'

    - name: 'Storage2'
      # classname is omitted, it defaults to `name`
      # from storage.storage2 import Storage2
      import: 'storage.storage2'
      configuration: 'put your configuration for Storage2 here'

Task failures¶

First, make sure you are familiar with retry options that can be passed in the YAML configuration.

If your task should not be rescheduled due to a fatal error, raise FatalTaskError. This will cause fatal task error and task will not be rescheduled. Keep in mind that max_retry from YAML configuration file will be ignored! If you want to retry, just raise any appropriate exception that you want to track in trace logs.

In case you want to reschedule your task without affecting max_retry, just call self.retry(). Optional argument countdown specifies countdown in seconds for rescheduling. Note that this method is not fully compatible with Celery’s retry mechanism.

Check SelinonTask code documentation.

Some implementation details¶

Here are some implementation details that are not necessary helpful for you:

• SelinonTask is not Celery task
• the constructor of the task is transparently called by SelinonTaskEnvelope, which handles flow details propagation and also Selinon tracepoints
• SelinonTaskEnvelope is of type Celery task

SqlStorage - SQLAlchemy adapter for SQL databases¶

pip3 install selinon[postgresql]

A configuration example:

storages:
  - name: 'MySqlStorage'
    classname: 'PostgreSQL'
    import: 'selinon.storages.postgresql'
    configuration:
      connection_string: 'postgres://postgres:postgres@postgres:5432/postgres'
      encoding: 'utf-8'
    echo: false

The connection string can be parametrized using environment variables. The implementation is available in selinon.storages.postgresql.

Redis - Redis database adapter¶

pip3 install selinon[redis]

A configuration example:

storages:
  - name: 'MyRedisStorage'
    classname: 'Redis'
    import: 'selinon.storages.redis'
    configuration:
      host: 'redishost'
      port: 6379
      db: 0
      charset: 'utf-8'
    port: 27017

Configuration entries host, port, password and db can be parametrized using environment variables. The implementation is available in selinon.storages.redis.

MongoDB - MongoDB database adapter¶

pip3 install selinon[mongodb]

A configuration example:

storages:
  - name: 'MyMongoStorage'
    classname: 'MongoDB'
    import: 'selinon.storages.mongodb'
    configuration:
      db_name: 'database_name'
      collection_name: 'collection_name'
      host: 'mongohost'
    port: 27017

Configuration entries db_name, collection_name, host and port can be parametrized using environment variables. The implementation is available in selinon.storages.mongodb.

S3 - AWS S3 database adapter¶

`pip3 install selinon[s3]`

A configuration example:

storages:
  - name: 'MyS3Storage'
    classname: 'S3Storage'
    import: 'selinon.storages.s3'
    configuration:
      bucket: 'my-bucket-name'
      aws_access_key_id: 'AAAAAAAAAAAAAAAAAAAA'
      aws_secret_access_key: 'BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB'
      region_name: 'us-east-1'

Configuration entries bucket, aws_access_key_id, aws_secret_access_key, region_name, location, use_ssl and endpoint_url can be parametrized using environment variables. The implementation is available in selinon.storages.s3.

In memory storage¶

A configuration example:

storages:
  - name: 'Memory'
    classname: 'InMemoryStorage'
    import: 'selinon.storages.memory'
    configuration:
      echo: false

No additional requirements are necessary to be installed. This storage adapter stores results in memory. It is suitable for use with Selinon CLI and executor where you just want to run a flow and check results. As results are stored in memory, it is not possible to scale number of workers in many cases as results are stored in memory of a node.

The implementation is available in selinon.storages.memory.

Few notes on using adapters¶

If you want to you multiple adapters, you can specify multiple adapters in extras when installing:

pip3 install selinon[mongodb,postgresql,s3]

Note that spaces are not allowed in extras (also escape brackets when using zsh).

from selinon import DataStorage

class MyStorage(DataStorage):
    def __init__(self, host, port):
        # arguments from YAML file are pasased to constructor as key-value arguments
        pass

    def is_connected():
        # predicate used to check connection
        return False

    def connect():
        # define how to connect based on your configuration
        pass

    def disconnect():
        # define how to disconnect from storage
        pass

    def retrieve(self, flow_name, task_name, task_id):
        # define how to retrieve results
        pass

    def store(self, flow_name, task_name, task_id, result):
        # define how to store results
        pass

    def store_error(self, node_args, flow_name, task_name, task_id, exc_info):
        # optionally define how to track errors/task failures if you need to
        pass

    def delete(self, flow_name, task_name, task_id):
        # define how to delete results
        pass

storages:
  # from myapp.storages import MyStorage
  - name: 'MyStorage'
    import: 'myapp.storages'
    configuration:
      host: 'localhost'
      port: '5432'

from selinon import StoragePool

# Name of storage was set to MyMongoStorage in nodes.yaml configuration file (section storages).
mongo = StoragePool.get_connected_storage('MyMongoStorage')

Sentry integration¶

If you would like to use Sentry for monitoring, you can use already existing support. Selinon reports all exceptions to the Sentry instance if you provide the following configuration:

global:
  trace:
    - sentry:
        # You can use environment variables (DSN picked from SENTRY_DSN in this case) using the following notation:
        # dsn: '{SENTRY_DSN}'
        dsn: 'http://5305e373726b40ca894d8cfd121dea34:78c848fac46040d1a3218cc0bf8ef6a7@sentry:9000/2'

You need to adjust the Sentry DSN configuration so it points to correctly set up Sentry instance.

Also don’t forget to install extras providing Sentry integration:

$ pip3 install selinon[sentry]

Reuse results in selective flows¶

If you would like to avoid running some task in selective runs (let’s say Task1 from the previous example) and reuse results from previous runs, you can simply register a selective run function in your YAML configuration file in task definition:

tasks:
  - name: 'Task1'
    import: 'myproject.tasks'
    selective_run_function:
      # from myproject.selective import selective_run_function
      name: 'selective_run_function'
      import: 'myproject.selective'

This function gets called by dispatcher in order to decide whether the desired task should be run based on result that this function computes.

The selective run function signature is:

selective_run_function(flow_name, node_name, node_args, task_names, storage_pool)

Arguments passed to this function:

• flow_name - name of flow in which this task was requested to run
• node_name - name of the node in the flow for which the selective_run_function was called
• node_args - arguments supplied to the flow
• task_names - tasks that were requested to start in the selective flow run (same as passed to run_flow_selective)
• storage_pool - instantiated StoragePool with appropriate mapping (so you can use this storage pool for querying parent tasks results)

The result of selective_run_function should be None if the node should be run (to be more precise scheduled) or id of task which result should be reused.

The selective run function is called only for tasks on the path, they are never run for tasks that are not on the direct path to desired task. It is also not called on tasks that you requested to run in selective flow as they are always run.

Sub-flows and subsequent tasks in selective task runs¶

Selinon by default computes only paths for one flow - the flow that you stated in the selective run. If you wish to run desired task also in sub-flows, configure run_subsequent as true. In this case Selinon will check all sub-flows for desired task occurrence and run also sub-flows, if necessary. Note that desired task in this case needs to be present in any of sub-flows (not necessary in the top-level one).

If you wish to run all subsequent tasks that depend on tasks that you stated in your selective task run, pass run_subsequent as true in your selective configuration. In this case the selective run function will not get called, rather all subsequent tasks get scheduled based on condition as in basic flow run.

Using selective task runs from YAML configuration¶

Now let’s consider that you defined a flow in our YAML configuration file and you want to reuse this definition in order to run this flow from another flow. Moreover, we want to run only some certain tasks. Selinon easily offers you a solution to this:

---
  flow-definitions:
    - name: 'selective_flow2'
      edges:
        - from:
          to: 'Task4'
        - from: 'Task4'
          to: 'selective_flow1'
          selective:
            tasks:
              - 'Task2'
            follow_subflows: false
            run_subsequent: false

The configuration stated above will define selective sub-flow, that basically runs only Task2 from our previous flow selective_flow1. Semantics of keys in the YAML configuration conform to arguments that are passed to the selective run function.

For better understanding, here is your selective_flow2 visualization:

Note that in this particular scenario you can also do:

from selinon import run_flow_selective

# requesting to run Task2 (stated in selective_flow1), but selective_flow1 is a sub-flow of selective_flow2
# note follow_subflows!
run_flow_selective('selective_flow2', ['Task2'], node_args={'foo': 'bar'}, follow_subflows=True)

Without the selective part in your selective_flow2 configuration. Using selective in your YAML configuration is highly dependent on your use-case (and the selective run function implementation).

YAML configuration used in examples¶

---

  tasks:
    - name: 'Task1'
      import: 'myproject.tasks'
    - name: 'Task2'
      import: 'myproject.tasks'
    - name: 'Task3'
      import: 'myproject.tasks'
    - name: 'Task4'
      import: 'myproject.tasks'

  flows:
    - 'selective_flow1'
    - 'selective_flow2'

  flow-definitions:
    - name: 'selective_flow1'
      edges:
        - from:
          to: 'Task1'
        - from: 'Task1'
          to:
            - 'Task2'
            - 'Task3'
    - name: 'selective_flow2'
      edges:
        - from:
          to: 'Task4'
        - from: 'Task4'
          to: 'selective_flow1'
          selective:
            tasks:
              - 'Task2'
            follow_subflows: false
            run_subsequent: false

Tasks¶

Configuration of a task in the YAML configuration can look like the following example (all possible configuration keys stated):

tasks:
   - name: 'MyTask1'
     classname: 'MyTask1Class'
     import: 'myapp.tasks'
     max_retry: 5
     retry_countdown: 120
     output_schema: 'myapp/schemas/my-task1.json'
     storage: 'Storage1'
     storage_read_only: false
     storage_task_name: 'MyTaskOne'
     selective_run_function:
        function: 'my_selective_function'
        import: 'myapp.selective'
     queue: 'my_task1_queue'
     throttling:
        seconds: 10

A task definition has to be placed into tasks section, which consists of list of task definitions.

Storages¶

Here is an example of a storage configuration with all the configuration options:

storages:
  - name: 'Storage1'
    import: 'myapp.storages'
    classname: 'SqlStorage'
    cache:
      name: 'Cache1'
      import: 'myapp.caches'
      configuration:
        size: 10
    configuration:
      connection_string: 'postgresql://postgres:postgres@localhost:5432/mydatabase'
      echo: false

A storage definition has to be placed into storages section, which is a list of storage definitions.

Flow definition¶

A flow definition is placed into a list of flow definitions in the YAML configuration file.

flow-definitions:
  - name: 'flow1'
    propagate_parent:
      - 'flow2'
    node_args_from_first: true
    #propagate_compound_finished:
    max_retry: 2
    retry_countdown: 10
    propagate_finished:
      - 'flow2'
    propagate_node_args:
      - 'flow2'
    nowait:
     - 'Task1'
    eager_failures:
     - 'Task2'
    cache:
      name: 'Cache1'
      import: 'myapp.caches'
      configuration:
        size: 10
    sampling:
      name: 'constant'
      args:
        # check for flow state each 10 seconds
        retry: 10
    throttling:
       seconds: 10
    edges:
      - from:
          - 'Task1'
        to:
          - 'Task2'
          - 'flow2'
        condition:
          name: 'fieldEqual'
          args:
            key:
              - 'foo'
              - 'bar'
            value: 'baz'
      - from: 'flow2'
        to: 'Task3'
      - from: 'Task3'
        to: 'flow3'
        condition:
          name: 'fieldEqual'
          args:
            key:
              - 'foo'
            value: 'bar'
        foreach:
          import: 'myapp.foreach'
          function: 'iter_foreach'
          # result of the function would be used as sub-flow arguments to flow3
          propagate_result: false
    failures:
      nodes:
        - 'Task1'
      fallback:
        - 'Fallback1'

Flow edge definition¶

A flow consist of time or data dependencies between nodes that are used in the flow. These dependencies are modeled using edges which are conditional and can have multiple source and multiple destination nodes (tasks or flows).

condition:
  #or:
  and:
    - name: 'fieldEqual'
      node: 'Task1'
      args:
        key: 'foo'
        value: 'bar'
    - not:
        name: 'fieldExist'
        node: 'Task2'
        args:
          key: 'baz'
          value: 42

Flow failures¶

A list of failures that can occur in the system and their fallback nodes.

• Possible values:
  • a list of failures each item defining:
    • nodes - a node name or a list of node names that can trigger fallback scheduling in case of failure
    • fallback - a node name or a list of node names (a task name or flow names) that should be scheduled in case of failure
    • condition - condition that would be evaluated, if true the fallback is triggered; see condition definition on task flow edges for more info and examples
• Required: false
• Default: an empty list of failures - all failures will be propagated to parent flows

An example of a failure definition:

failures:
   - nodes:
       - 'Task1'
       - 'Task2'
     fallback: 'Fallback1'

   - nodes: 'Task1'
     fallback:
       - 'Fallback1'
       - 'Fallback2'

Failures are greedy, if multiple fallbacks can be run, there is used failure that covers as mush as possible of the failed nodes.

global¶

Global configuration section for Selinon.

cache¶

Define cache for result caching or for task state caching - see distributed caches in Optimization section. Each cache has to be of type Cache.

Do not introduce a new flow unless it is necessary¶

Each flow adds some overhead. It is reasonable to introduce a new flow when:

• you want to reuse certain implementation.
• you want to encapsulate flow so you can run fallback nodes for the whole group of tasks
• you want to throttle or optimize certain tasks (you want to let specific workers do specific flows or tasks)
• you want to recursively run the whole flow or you have dependencies on flows so they are not resolvable except copying YAML configuration logic
• you want to refer to flow as a group of tasks - a separate indivisible logic
• you need to run a flow with different arguments - each flow is defined by a group of tasks, their time and data dependencies and arguments that are propagated in the whole flow to each task (possibly except the very first one)
• re-usability in general
• based on semantics of your flows
• any other reasonable argument…

Do not add flags unless you really need them¶

Selinon implementation was desingned to be lazy. That means it was designed not to add overhead for you unless you really need it. If you don’t need flags such as propagate_parent or propagate_finished you don’t need to state them in the YAML configuration file. If you state them, it will add additional computational overhead that you don’t necessarily need.

Make flow edges as minimal as possible¶

Even though Selinon supports multiple source and destination nodes in the edge configuration it is generally a good idea to make these edges as minimal as possible. This way you can mark unused edges with alwaysFalse condition without need to purge queues when you do redeployment with different configuration.

Aliases for flows, tasks and storages¶

You probably saw (based on examples) that you can easily define a task alias - tasks share Python class/implementation, but have different name:

tasks:
  # from myapp.tasks import Task1
  - name: 'Task1'
    import: 'myapp.tasks'

  # from myapp.tasks import Task1 as Task2
  - name: 'Task2'
    classname: 'Task1'
    import: 'myapp.tasks'

This is useful when you want to run same code multiple times in a flow (since nodes are referenced by names). Also check storage_task_name configuration option that enables you to do store with different task name compared to one stated in the nodes.yaml file.

You can also define storage alias - useful when you want to use same database/storage adapter but with different configuration:

storages:
  # from myapp.storages import PostgreSQL as UserPostgres
  - name: 'UserPostgres'
    classname: 'PostgreSQL'
    import: 'myapp.storages'
    configuration:
      host: postgres-user
      port: 5432

  # from myapp.storages import PostgreSQL as AdminPostgres
  - name: 'AdminPostgres'
    classname: 'PostgreSQL'
    import: 'myapp.storages'
    configuration:
      host: postgres-admin
      port: 5432

You can also create flow aliases. This is especially helpful if you want to re-use flow configuration such as edges, but you want to separate one flow to different queue (for example due to SLA requirements, so more workers can serve serve messages on SLA-specific queue). You can do so easily since YAML supports references:

flow-definitions:
    - &flow1_def
      name: 'flow1'
      queue: 'flow1_v1'
      propagate_node_args: true
      edges:
          - from:
            to: 'Task4'
          - from: 'Task4'
            to: 'Task5'

    - <<: *flow1_def
      name: 'flow1_sla'
      queue: 'flow1_sla_v1'
      # node_args_from_first and edges configuration will be taken from flow1

Make your queue names configurable via environment variables¶

You can easily make queue names (for tasks and for flows/dispatcher) dependent on environment variables:

tasks:
  - name: 'Task1'
    import: 'myapp.tasks'
    queue: '{DEPLOYMENT_PREFIX}_task1_v1'

Selinon will expand queue name for you based on DEPLOYMENT_PREFIX. Let’s say you set DEPLOYEMENT_PREFIX to testing. The expanded queue name will be testing_task1_v1.

Now you can deploy two instances of your system on the same cluster without affecting each other. This might be helpful for testing purposes when you have a testing cluster where you want to run integration tests that do not affect each other. Similarly you can introduce such option for storages.

Variable number of child tasks or sub-flows¶

If you would like to schedule multiple tasks (or sub-flows) of a same type, this configuration will perfectly work for you (assuming you want to schedule 3 tasks Task2):

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to:
          - 'Task2'
          - 'Task2'
          - 'Task2'

OK, now let’s assume that you don’t know how many tasks should be executed and you want to determine (or compute) the exact task count on runtime. Selinon offers you to do that by plugging your foreach function:

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'Task2'
        foreach:
          # from myapp.foreach import my_foreach_function
          function: 'my_foreach_function'
          import: 'myapp.foreach'

Here is the corresponding visualization:

An example of such foreach function could be:

def my_foreach_function(storage_pool, node_args):
    return range(node_args.get('task2_schedule_count', 0))

The return value for foreach function should be an iterable (see bellow why) which length determines number tasks that should be scheduled.

Arguments for the foreach function are:

• storage_pool - instantiated StoragePool that can be used to query parent task results
• node_args - arguments of the current flow

If you would like to schedule sub-flows dynamically, you can do so by providing a sub-flow in the destination of edge definition:

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'flow2'
        foreach:
          # from myapp.foreach import my_foreach_function
          function: 'my_foreach_function'
          import: 'myapp.foreach'
          # uncomment if results of the foreach function should be used as sub-flow arguments
          #propagate_result: true

In case you also provide propagate_result as true, each entry in the resulting iterable would be propagated to sub-flow flow2 as sub-flow arguments.

Permutations of all parent tasks a.k.a Cyclic diamond with combinations¶

Consider the following flow definition:

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to:
          - 'Task2'
          - 'Task3'
      - from:
          - 'Task2'
          - 'Task3'
        to: 'Task4'
      - from: 'Task4'
        to: 'Task1'

A visualization of such flow using Selinonlib CLI would be:

If the visualization is not straightforward for you at first, basically there is a “diamond” that is called in cycle (without conditions):

As one would expect Task4 is called after Task2 and Task3 are finished. Now let’s simulate how Selinon would evaluate such cyclic dependencies.

We start with task Task1. After task Task1 finishes, Selinon schedules Task2 and Task3. Let’s assign identifiers (id) to these tasks:

• Task2 with id <task2_1>
• Task3 with id <task3_1>

So now Selinon can proceed to task Task4. This task will have parent tasks with identifiers <task2_1> and <task3_1> (for Task2 and Task3 respectively).

Once task Task4 finishes, Selinon will schedule Task1 again. After the second Task1 finishes, there are started Task2 and Task3, now with the following identifiers:

• Task2 with id <task2_2>
• Task3 with id <task3_2>

And we can proceed to task Task4. Now the question could be: What will be parent tasks of task Task4 in the second iteration?

The answer is that there will be scheduled multiple tasks Task4 with the following parent task identifiers:

1. Task4 with parent identifiers:

• <task2_1> for task Task2 from the first iteration
• <task3_2> for task Task3 from the second iteration

2. Task4 with parent identifiers:

• <task2_2> for task Task2 from the second iteration
• <task3_1> for task Task3 from the first iteration

3. Task4 with parent identifiers:

• <task2_2> for task Task2 from the second iteration
• <task3_2> for task Task3 from the second iteration

If we would stop flow after the second iteration (before task Task1 is run for the third time), we would see that there were executed four tasks Task4 to respect all possible combinations.

Cyclic Diamond without combinations¶

If you would like to avoid running Task4 in the previous example for each combination and rather run it once per iteration, you need to create a sub-flow:

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'flow2'
      - from: 'flow2'
        to: 'Task4'
      - from: 'Task4'
        to: 'Task1'

   - name: 'flow2'
     edges:
      - from:
        to: 'Task2'
      - from:
        to: 'Task3'

The the visualization of flow flow1 would be:

And the visualization of flow flow2 would be:

Now flow flow2 would be executed once per each iteration in flow flow1 and there will be exactly one Task4 invocation per each flow flow1 iteration.

Aggregating tasks from sub-flows¶

Results of flows are serialized and stored as JSONs in the result backend (configured by Celery configuration). Each dispatcher reports only tasks and flows that were started inside flow that was handled by dispatcher task (except nowait nodes). Tasks from sub-flows are captured in dispatchers that handle started sub-flows.

If you request to propagate finished nodes from sub-flows to tasks as parents you need to explicitly state propagate_finished as true in your YAML configuration option. Information is gathered and you can transparently query parent task results (or failures).

By default you need respect how sub-flows are organized - so if your flow flow1 starts sub-flow flow2 and that starts flow3, you need to respect this organization if you would like ask for results of task Task1 started in flow flow3 from flow flow1:

from selinon import SelinonTask

# this task is run in flow1
class MyTask(SelinonTask):
    def run(self, node_args):
        print('Task1' in self.parent['flow2']['flow3'].keys())  # should be True iff Task1 was successful in flow3
        task1_result = self.parent_flow_result(['flow2', 'flow3'], 'Task1')  # optionally pass index if there were multiple instances of Task1
        print('Result of Task1 run in parent sub-flow flow3 is {}'.format(task1_result))

If you would like to compound (or flatten) these parent sub-flow organization details, just set propagate_compound_finished instead of propagate_finished. In that case you can directly use:

from selinon import SelinonTask

# this task is run in flow1
class MyTask(SelinonTask):
    def run(self, node_args):
        print('Task1' in self.parent['flow2']['flow3'].keys())  # should be True iff Task1 was successful in flow3
        task1_result = self.parent_flow_result('flow2', 'Task1')  # optionally pass index if there were multiple instances of Task1
        print('Result of Task1 run in parent sub-flow flow3 (ignoring flow2 organization) is {}'.format(task1_result))

I see only one contributor, should I trust you?¶

There is currently one contributor, but every project started somehow. Selinon was designed for fabric8-analytics project at Red Hat that gathers project information for openshift.io (introduced at the Red Hat 2017 summit keynote), where it is still used and it already served millions flows and even more tasks. If you find Selinon interesting for your use-case, feel free to use it (and buy me some beer or at least let me know that you like it or share any experiences you have).

If you find a bug, place for enhancement or anything where I can be helpful, feel free to let me know. And not to forget - even you can be Selinon developer.

Dispatcher does not work properly or hangs in an infinite loop.¶

Check your result backend configuration for Celery. Currently, there is supported and well tested only Redis and PostgreSQL (feel free to extend this list based on your experiences). There were found serious issues when rpc was used.

Can I see Selinon in action?¶

See Selinon demo or fabric8-analytics project, especially it’s fabric8-analytics-worker.

Can I simulate Selinon run without deploying huge infrastructure?¶

Yes, you can. Just use shipped executor:

selinon-cli execute --nodes-definition nodes.yml --flow-definitions flow1.yml flow2.yml --help

This way you can also use Selinon to run your flows from a CLI. You can also explore prepared containerized demo.

I’m getting Python related errors!¶

Check your Python version. Selinon currently support only Pyhon3+ as Celery project is about to drop Python2 support.

Should I replace Celery with Selinon?¶

Well, hard to say. Celery is a great project and offers a lot of features. Selinon should be suitable for you when you have time and data dependencies between tasks and you can group these tasks into flows that are more sophisticated than Celery’s primitives such as chain or chord. If this is true for you, just give Selinon a try. If you are already using Celery, check prepared guide on how to migrate from raw Celery to Selinon.

How should I name tasks and flows?¶

You should use names that can became part of function name (or Python3 identifier). Keep in mind that there is no strict difference between tasks, flows and sub-flows, so they share name space.

How can I access nested keys in a dict in default predicates?¶

Assuming you are using predicates from selinon.predicates. What you want is (in Python3):

message['foo']['bar'] == "baz"

Predicates were designed to deal with this - just provide list of keys, where position in a list describes key position:

condition:
  name: 'fieldEqual'
  args:
      key:
          - 'foo'
          - 'bar'
      value: 'baz'

I need a custom predicate, how to write it?¶

If selinon.predicates predicates are not suitable for you or you miss a specific predicate, you can define your own module in the global configuration. See YAML configuration section for details.

What exceptions can predicates raise?¶

Predicates were designed to return always true or false. If a condition cannot be satisfied, there is returned false. So it is safe for example to access possibly non-existing keys - predicates will return false. This idea has to be kept even in your predicates as predicates are executed by dispatcher. If you rise an exception inside predicate the behaviour is undefined.

Do I need result backend?¶

Or more precisely: Do I need a result backend even when I am using my custom database/storage for task results?

Yes, you do. The result backend is used by Celery to store information about tasks (their status, errors). Without result backend, Selinon is not capable to get information about tasks as it uses Celery. Do not use rpc backend as there were noted issues.

Why there is used generated code by Selinon?¶

Since YAML config files cover some logic (such as conditions), this needs to be evaluated somehow. We could simply interpret YAML file each time, but it was easier to generate directly Python code from YAML configuration files and let Python interpreter interpret it for us. Other parts from YAML file could be directly used, but mostly because of consistency and debugging the whole YAML file is used for code generation.

You can easily check how YAML files is transformed to Python code simply by running:

selinon-cli inspect --nodes-definition nodes.yml --flow-definitions flow1.yml flow2.yml --dump outputfile.py

How to write conditions for sub-flows?¶

This is currently a limitation of Selinon. You can try to reorganize your flows so you don’t need to inspect parent subflows, for most use cases it will work. Adding support for this is for future releases planned.

Is it possible to do changes in the configuration and do continuous redeployment?¶

Yes, you can do so. BUT make sure you do migrations - see the migration section to get insights on how to do it properly.

What happens if I forgot to do migrations?¶

If you do changes in the YAML configuration files and you do not perform migrations, unpredictable things may happen if your queues have still old messages. It’s always a good idea to check whether migration files need to be generated. See Migrations - Redeployment with changes for more details.

Is my YAML config file correct? How to improve or correct it?¶

See Best practices section for tips.

Can I rely on checks of YAML files?¶

You can a bit, but think before you write configuration. There are captured some errors, but checks are not bullet-proof. If you make logical mistakes or your flow is simply wrong, Selinon is not AI to check your configuration. There are not done checks on transitive dependencies, if given conditions could evaluate or so.

Is there a way how to limit task execution time?¶

Currently there is no such mechanism. Celery has time limit configuration option, but note that Selinon tasks are not Celery tasks.

Why there is no support for older Celery versions?¶

One of the requirements of Selinon is, that it defines tasks (Dispatcher and SelinonTaskEnvelope) before the Celery’s application gets instantiated. Older versions of Celery requested tasks to be registered after the Celery’s application was created. This makes it chicken-egg problem.

What broker type do I need?¶

Selinon uses Celery for queue handling and running, so you have to use broker implementation that is supported by Celery - such as SQS or RabbitMQ.

Selinon requires that you messages are delivered - it’s okay if messages are delivered more than once (see for example SQS details regarding deliver at least one). You will just end up with multiple tasks executed at the same time. You can tackle that in your application logic.

Why does a flow finishes too early when using AWS SQS?¶

Most likely you are using AWS SQS standard queues that can deliver a single message multiple times. If your application logic processes one message but a task fails when the second message is processed (e.g. integrity errors if task ids are unique in PostgreSQL), Celery overwrites task state stored in the result backend. This causes that even if task succeeds (first run) it’s state can be tracked as failed.

A solution to this problem is to patch Celery’s result backend to restrict only one task, something like (in case of PosgreSQL as a result backend):

diff --git a/celery/backends/database/__init__.py b/celery/backends/database/__init__.py
index 506a4cc69..57d29a6ca 100644
--- a/celery/backends/database/__init__.py
+++ b/celery/backends/database/__init__.py
@@ -110,6 +110,9 @@ class DatabaseBackend(BaseBackend):
                 task = Task(task_id)
                 session.add(task)
                 session.flush()
+            elif task.status in states.READY_STATES:
+                # Do not overwrite on multiple message delivery (e.g. SQS).
+                return task.result
             task.result = result
             task.status = state
             task.traceback = traceback

Or simply switch to AWS SQS FIFO queues that guarantee exactly once delivery of a message.

What does Selinon mean?¶

Selinon means Celery in Greek language. The main reason for using Greek language was the fact that there are already successful project out there that do distributed systems and have Greek names (see Kubernetes as an example). But Greek language is cool anyway :-).

Sub-flows¶

As dispatcher is a task as any other, sub-flows are handled by scheduling dispatcher task that handles given sub-flow. So if you have two flows flow1 and flow2, and you run flow2 as a sub-flow in flow1 flow, dispatcher in flow1 will schedule dispatcher task handling flow2 as a task that is a node in the task dependency graph you provided in the YAML configuration file.

Each dispatcher reports information about finished and failed nodes as a JSON that is stored in the result backend. There are tracked only nodes that are run in the given flow. Any sub-flow related information needs to be computed when requested (such as propagate_finished, see YAML configuration).

Failure handling¶

Failure of a task means that the task raised an exception. Currently, there is done quiet straightforward failure handling. If a task fails, dispatcher will keep the failure in dispatcher arguments for the next run. Once there is nothing to proceed with and there are no active nodes, dispatcher will try to recover from flow failure by scheduling fallback tasks or fallback flows (fallback flow nodes). If the fallback was successfully run and it means that dispatcher recovered from flow failure, dispatcher can continue with scheduling new tasks or sub-flows as there would be no failure.

If dispatcher cannot recover from failures in the flow, there is propagated flow failure to parent flows (if any) and the current flow is marked as failed.

Inspecting configuration¶

Selinon CLI offers you the inspect sub-command. This sub-command parses all the config files and outputs requested results or just checks your configuration consistency.

You can use the inspect sub-command for example to list all queues stated in your YAML files or do other stuff for querying parsed and checked configuration files.

One of the interesting options worth to state is the --dump command. As Selinon uses YAML config files that carry some logic, there is needed interpretation of some parts. Thus there is generated Python code from YAML configuration files that is afterwards supplied to Selinon itself to orchestrate flows. You can request the generated Python code with the --dump command if you are interesting in it or you just want to explore Selinon internals.

Plotting flow graphs¶

As flows and dependencies between tasks might get pretty complex, Selinon offers you a way to visualize flows. For this purpose the plot sub-command is available in Selinon CLI. It can plot flow graphs for you in any format that is supported by graphviz.

All the flow graphs available in this documentation were plotted using the plot sub-command.

You can also adjust style of the resulting image by supplying a YAML-based configuration that states style information. As Selinon uses graphviz under the hood to plot flow graphs, all options that are supported by graphviz are applicable to nodes, edges, storages and arrows. In the example bellow, you can see how to adjust style configuration of different components that occur in the resulting flow plot (this is default configuration).

# A task node.
task:
    style: filled
    color: black
    fillcolor: '#66cfa7'
    shape: ellipse
# A flow node (a sub-flow in a flow).
flow:
    style: filled
    color: black
    fillcolor: '#197a9f'
    shape: box3d
# A condition on a simple edge.
condition:
    style: filled
    color: gray
    fillcolor: '#e8e3c8'
    shape: octagon
# A condition on foreach edge.
condition_foreach:
    style: filled
    color: gray
    fillcolor: '#e8e3c8'
    shape: doubleoctagon
# A storage node.
storage:
    style: filled
    color: black
    fillcolor: '#894830'
    shape: cylinder
# A simple edge.
edge:
    arrowtype: open
    color: black
# An edge from a task to storage that was assigned to the task.
store_edge:
    arrowtype: open
    color: '#894830'
    style: dashed
# An edge that leads to a fallback node.
fallback_edge:
    arrowtype: open
    color': '#cc1010'
# A special mark signalizing to always recover from a failure (fallback set to true).
fallback_true:
    style: filled
    color: black
    fillcolor: '#5af47b'
    shape: plain

You can find more configuration options in the graphviz library documentation.

Simulating flow execution¶

To debug, explore, play or interact with task flow execution anyhow, Selinon CLI offers you a built in executor. This executor tries to simulate message queueing and message consuming so no broker (and Celery’s result backend) is involved.

Executor currently supports only single-process, single threaded executor - one worker serving tasks. Worker accepts messages in a round-robin fashion based on message availability in queues.

In order to see what is happening during executor run, you can run executor in a verbose mode. Executor in that case prints all the execution actions. It can help you when you want to experiment with your flow configuration or you would like to debug strange flow behaviour.

Generating migrations of configuration files¶

As Selinon offers you a mechanism to do changes in your configuration files and do re-deployment of workers, there needs to be a mechanism that ensures changes done in your configuration files are reflected to already present messages on queue. This lead to migrations design.

You can generate migration files using the migrate sub-command. Please take a look to the section that explains migrations in more detail.

Using environment variables to supply options¶

If you run Selinon CLI in various scripts or you would like to interact with Selinon CLI in different environments, you can explicitly state your options in environment variables:

export SELINON_NODES_DEFINITION=/path/to/nodes.yaml
export SELINON_FLOW_DEFINITIONS=/path/to/flows/
export SELINON_EXECUTE_NODE_ARGS_JSON=1
# No need to explicitly state YAML configuration files and option to do JSON parsing.
$ selinon-cli execute --flow-name flow1 --node-args '{"foo": "bar"}

export SELINON_NODES_DEFINITION=/path/to/nodes.yaml
export SELINON_FLOW_DEFINITIONS=/path/to/flows/
$ selinon-cli inspect --list-task-queues  # No need to supply --nodes-definition and --flow-definitions explicitly

The schema for constructing environment variables is SELINON_<SUBCOMMAND>_<OPTION> where <SUBCOMMAND> is Selinon’s CLI sub-command in uppercase and <OPTION> is requested option (converted to uppercase, dashes converted to underscores). The only exception are --nodes-definition and --flow-definitions where <SUBCOMMAND> is omitted.

Why are migrations needed¶

If you inspect the message payload that is sent to message broker by dispatcher task (see Deep diving into Selinon internals section for more info), you can find that there is captured the current flow status (e.g. active edges in the flow graph, finished nodes, active nodes, etc.). Actually this needs to be captured somewhere as dispatcher needs to know which tasks should be and which shouldn’t be scheduled during the dispatcher run (at least during flow execution). The way how dispatcher is designed (sampling by retrying) makes ideal place for storing such information the dispatcher message as this message is adjusted each time dispatcher is rescheduled (according to the current flow state). After the flow is finished, this information is no longer needed and thus it’s safe to drop it (message is acked and automatically removed from queue).

Imagine you have a flow. You configured workers, did deployment to production, but after some time you realized that some tasks should be removed or added to your flow or you would like to change how the flow is structured. As you already run in production, you don’t want to purge queues and drop all the messages that were already published on the message broker. As messages that are available on message broker carry information (state) of the flow, they refer to the configuration that was used when the flow started. If you would like to change flow behaviour, there are 2 main options how to do that.

Tainted flows¶

If your change in the configuration file affects flow behaviour, dispatcher marks the affected flow as tainted. A tainted flow is flow that would behave differently if it would be executed with the new configuration, but it has already started and the current state resolves in different flow execution compared to the new configuration file.

Whether a flow is tainted highly depends on state of the flow in which the migration is done and the migration itself.

There are two types of events that lead to tainted flows:

• tainted edges - flows tainted based on edges that were fired but they do not exist in the new configuration
• tainting nodes - flows tainted based on nodes that weren’t executed as the state of the flow outrun added edges

Tainted edges¶

Tainted edges are edges that were present in the old configuration but are not present in the new one, or destination nodes of edge were modified. If the state of flow in the deployed system already fired such edge (destination nodes were run as edge condition was evaluated as true), the flow is marked as tainted and requested tainted flow strategy will be triggered based on the migration configuration (see bellow).

Let’s say you have a flow as shown on the figure above. In your configuration file you remove the edge that leads from Task1 to Task3 so the flow structure will change as visualized on the figure bellow. You generate a migration and do the redeployment, but you have already some flows in progress. Let’s say that worker picks dispatcher task, the flow already run Task1 and fired both edges leading to Task2 and Task3. In this case the flow progress basically waits for these tasks to finish based on the old configuration. However with migration that you did, there is no edge to Task3 - as this edge was already fired, it tainted the flow. In this case dispatcher will report tainted flow and will behave according to your tainted flow strategy (see Tainted flow strategy bellow).

In another case, a worker would pick message respecting old configuration in which Task1 was scheduled, but dispatcher in the previous run did not find that Task1 finished (edges to Task2 and Task3 were not fired). In this case the flow is not marked as tainted and dispatcher continues checking state of nodes respecting the new configuration with migration.

Tainting nodes¶

Tainting nodes are nodes that were run based on the current state of flow, but there was added an edge in migration that would be evaluated (as source/tainting nodes already run) if all of the tainting nodes would get finished.

Let’s illustrate this scenario as well. In this case you have a flow that runs two tasks - Task1 and Task2 after the Task1 finishes. You deployed this version of configuration, but after a while you decided to add a new task called Task3 that is scheduled after Task1 successfully finishes. This scenario is visualized on the figure bellow. You create a migration file do the redeployment with new configuration file and generated migration. Some flows were already in progress - let’s say that worker picked dispatcher task and the state of the flow indicates that Task1 was already run, the edge to Task2 was already triggered (dispatcher waits for Task2 to finish), but with new configuration and migration, there would be evaluated the edge to Task3 as well (this was not done with the old configuration). Dispatcher does not schedule Task3 as this edge would be fired after Task1 finishes (the event noticing that Task1 finished was already tracked), but rather marks flow as tainted. Once the flow is marked as tainted, the tainted flow strategy (see Tainted flow strategy bellow) is evaluated.

Note

Conditions are not evaluated in case of tainting nodes. Tainted flow indicates that the edge would be evaluated and possibly fired based on result of condition result.

Tainted flow strategy¶

When a flow is marked as failed, you can configure how dispatcher should behave in such situations. There are three types of tainted flow strategies:

1. IGNORE (default) - migrations marked as IGNORE do not perform any actions on tainted flows, a tainted flow is reported using the tracing mechanism, but the execution continues
2. RETRY - this strategy causes that the whole flow is retried - all results computed until migration will not be reused in the retried flow
3. FAIL - flow will be immediately marked as failed - previously scheduled tasks will be run (as they are already scheduled) ignoring state

See Trace flow actions section for information about captured details. If there is performed migration chaining (see bellow), state is distinguished based on tainted flow strategy with the highest impact:

• if at least one flow migration has strategy FAIL - the tainted flow will fail
• if at least one flow migration has strategy RETRY but no FAIL, the tainted flow will be marked as RETRY
• if all flow migrations have strategy IGNORE and there is not present RETRY nor FAIL strategy, tainted flows will be ignored but reported in the tracing mechanism

Migration chaining¶

Migrations can be chained - they are incrementally numbered starting with 1. A special migration version 0 means that no migrations are present - Selinon was not set up to use migrations and no migrations are present. If you would like to do migrations, add anytime (even when you deployed worker with no migrations present) migration_dir to the global configuration section in the nodes.yaml file (see YAML configuration section for more details). These migrations will be run afterwards once worker detects migrations to be present.

global:
  # The migration dir can be an absolute or relative path to the process working directory.
  # Preferably state absolute path.
  migration_dir: 'path/to/dir/containing/migrations'

Note that migration version is not per flow specific, but rather deployment specific. This means that you can see even in messages incrementally changing migration version for flows that do not have any affecting migrations.

If you would like to perform more changes that should trigger different migration strategies, it’s perfectly okay to generate multiple migrations and apply them with different tainted flow strategy based on the flow state in your deployment. Migrations get applied based on migration versions (incrementally) as you would expect respecting tainted flow strategy.

Generating migrations¶

Migrations can be generated using the Selinon CLI. It operates either on explicitly supplied old and new configuration files on which the migration is computed or you can use the built in support for Git VCS. Selinon in that case checks Git history for older versions of configuration files supplied.

$ # Add --verbose parameters for noisy output, use git to retrieve old configuration files. Path to migration_dir will be taken from nodes.yaml.
$ selinon-cli -vvv migrate --flow-definitions myapp/flows/*.yaml --nodes-definition myapp/nodes.yaml --git

You can also explicitly adjust migration directory. Implicitly Selinon will take into account migration_dir stated in the global section of nodes.yaml file. Also make sure you are pointing to the right directory as migrations files are numbered incrementally so Selinon needs to know the previous migration version that is distinguished based on presence of migration files in the migration_dir directory.

$ # Explicitly point to old configuration files and the migration directory, set tainted flow strategy as FAILED.
$ selinon-cli -vvv migrate --flow-definitions myapp/flows/*.yaml --nodes-definition myapp/nodes.yaml --old-flow-definitions myapp/old_1/flows/*.yaml --nodes-definition myapp/old_1/nodes.yaml --migration-dir myapp/migration_dir/ --tainted-flow-strategy FAIL

Generated migrations are JSON files with computed tainting nodes, tainted edges and edge translation (renumbering and edge dropping if needed). These JSON files also carry some metadata such as where, when and by who the given migration was created. You can suppress capturing metadata by generating migrations with the --no-meta option.

Migrations and fallbacks¶

Migrations do not apply to fallback nodes and their dependencies. Fallback edges represent exceptional path that is triggered if there was a failure. You can safely do redeployment without a need of generating migrations for changes in fallbacks. The behaviour of fallback evaluation will respect fallback configuration based on the current configuration available in deployment.

Using and tracing migrations¶

Once you create migrations, make sure your migrations are present in the deployed worker so Selinon can transparently apply generated migrations. Without migration files present, Selinon is unable to perform migrations - that could lead to undefined behaviour when old messages would be processed with newer and modified configuration.

Take a look at the Trace flow actions section that will point you to tracing events that are emitted on migrations. They also carry some information that could help you with debugging and understanding what’s going on in your system.

Migration skew¶

There is a special error that is reported in case of inconsistency in migration files and the migration version carried within the message for dispatcher task - MigrationSkew. Dispatcher issues retry (without affecting the flow retry configuration option) that leads to republishing the message. This error indicates that the worker was unable to process the given message as migration version in the received message is newer than available migrations on the worker. In other words, worker is unable to correctly process the message.

This error can indicate that you have workers available with different migration versions available. If you do rolling updates where your replica count of old workers decrements while new workers are spawned, the MigrationSkew error can be seen. New workers start to publish messages that are not suitable for old workers, but some of the old workers that are still alive tried to process newly published messages. As you would expect, by republishing messages, old workers give up with processing new message types and leave the processing for a new worker with the new migration present.

Flow configuration for examples¶

The tainted flow figures were created using the following configuration.

tasks:
  - name: 'Task1'
    import: 'myapp.tasks'
  - name: 'Task2'
    import: 'myapp.tasks'
  - name: 'Task3'
    import: 'myapp.tasks'

flows:
  - 'flow1_tainted'
  - 'flow2_tainted'

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'Task2'
      - from: 'Task1'
        to: 'Task3'

  - name: 'flow2'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'Task2

Flow optimization & Dispatcher scheduling¶

In order to optimize your flow execution, you need to deeply understand core concept behind Selinon. As you already read, the key idea behind Selinon is dispatcher (see Dispacher). Dispatcher is periodically scheduled and checks state of tasks in the flow and schedules new if necessary.

As there is non-zero overhead for each dispatcher run - queue message, receive message, check task status (querying result backend) and queue message for dispatcher again, it is generally a good idea to optimize dispatcher scheduling.

Selinon offers you to optimize dispatcher scheduling by configuring sampling strategies.

A sampling strategy basically gives information on when the dispatcher should be scheduled (or to be more precise retried) in order to check the current flow status.

By default dispatcher is rescheduled every 2 seconds. Note that it means “give dispatcher at least 2 seconds to retry”. If you have full queues (and busy workers) your dispatcher can be rescheduled even after days.

Selinon offers you couple of predefined sampling strategies (refer to selinon.strategies for more info):

flow-definitions:
  - name: 'flow1'
    # define sampling strategy
    sampling:
      name: 'constant'
      args:
        # check for flow state at least each 10 seconds
        retry: 10
    edges:
      # a list of edges follows

You can also provide your own sampling strategy. Basically sampling strategy is given by a number which tells Selinon when to reschedule dispatcher in the flow. This number is computed in the sampling strategy function which accepts one positional argument status and keywords arguments that are passed to sampling strategy function based on the YAML configuration you provided.

Now let’s assume that we want to optimize scheduling dispatcher. We have the following flow definition:

flow-definitions:
  - name: 'flow1'
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'Task2'

Here is corresponding flow visualization:

Based on statistics we have we know that the execution time of task Task1 is 10 seconds in average and the execution time for task Task2 is 15 seconds in average. You can easily write your sampling strategy function:

import random

def my_sampling_strategy(status, randomness):
    if not status['active_nodes']:
        return None

    if 'Task1' in status['new_started_nodes']:
        return 10 * random.uniform(-randomness, randomness)

    if 'Task2' in status['new_started_nodes']:
        return 15 * random.uniform(-randomness, randomness)

    return max(status['previous_retry'] / 2, 2)

Now you can plug and use your sampling strategy function in your YAML configuration file:

flow-definitions:
  - name: 'flow1'
    sampling:
       # from myapp.sampling import my_sampling_strategy
       name: 'my_sampling_strategy'
       import: 'myapp.sampling'
       args:
         randomness: 3
    edges:
      - from:
        to: 'Task1'
      - from: 'Task1'
        to: 'Task2'

Now your sampling strategy function will be called each time dispatcher will want to reschedule. If None is returned, dispatcher should end flow immediately. Otherwise a positive integer has to be returned that represents number of seconds for retry.

Storage optimization & Distributed caches¶

By using Selinon you can reach to two main issues with your cluster on heavy load:

1. Your cluster is not powerful enough to serve requested number of tasks.
2. Your storage/database cannot process requested numbers of requests or your network is not capable to transmit such number of queries.

In the first case the solution is simple: buy/use more hardware.

In the latter one there are two main approaches how to tackle such bottleneck. You can always use more storage replicas or split data accross multiple storages and transparently configure Selinon to use different storages for different purposes (see storages aliasing in Best practices).

If the above solution is not suitable for you or you want to optimize even more, Selinon offers you an optimization that introduces distributed caches. These caches are distributed across nodes (workers) in your cluster and act like a caching mechanism to reduce number of requests to storages/databases and keep data more close to execution nodes.

Selinon by default uses cache of size 0 (no items are added to the cache). There are prepared in-memory caches like FIFO (First-In-First-Out cache), LIFO (Last-In-First-Out cache), LRU (Least-Recently-Used cache), MRU (Most-Recently-Used cache), RR (Random-Replacement cache). See selinon.caches for more info.

tasks:
  - name: 'Task1'
    import: 'myapp.tasks'
    cache:
      # from myapp.cache import RedisCache
      name: 'RedisCache'
      import: 'myapp.cache'
      configuration:
        host: 'redis'
        port: 6379
        db: 0
        password: 'secretpassword'
        charset: 'utf-8'
        host: 'mongo'
        port: 27017

flow-definitions:
  - name: 'flow1'
    cache:
      # from myapp.cache import RedisCache
      name: 'RedisCache'
      import: 'myapp.cache'
      configuration:
        host: 'redis-cache'
        port: 6379
        db: 0
        password: 'secretpassword'
        charset: 'utf-8'
        host: 'mongohost'
        port: 27017
    edges:
      - from:
        to: 'Task1'

Prioritization of tasks and flows¶

You can easily prioritize tasks or flows by placing them on separate queues and dedicate more workers to process these tasks or flows (see queue option in the YAML configuration section). Note that queues that are reserved for flows are used for dispatcher task that does scheduling based on state (not for tasks in general that are stated in the flow).

To run Celery worker only the given queues queue1 and queue2 specify Celery’s -Q argument:

celery worker -A myapp.start -Q queue1,queue2 # additional parameters follow

Throttling of tasks and flows¶

Throttling can be performed by reducing number of workers that process given task or flow as opposite to described prioritization (see above).

If you would like to perform even more robust throttling, you can apply throttling option on task or flow level (see throttling option in the YAML configuration section). Note that for setting up throttling properly you need to dedicate one cluster node that would accept dispatcher task (flow on a separate queue) in the given flow with throttled tasks or flows. This guarantees that scheduling of throttled tasks or flows is done on one place so the given worker node can perform time-based throttling properly based on internally kept state.

Flow throttling is done on flow level (not on task level) similarly as flow prioritization described above.

An example of configuration:

---
tasks:
  - name: Task1
    import: myapp.tasks
    queue: task1_v0
    throttling:  # task level throttling
      seconds: 10

# additional entries follow

flow-definitions:
  - name: flow1
    queue: flow1_v0
    throttling:  # flow level throttling
      minutes: 30
      seconds: 10
    edges:

# additional entries follow

Other optimizations¶

If you would like to optimize performance take a look at Celery’s user guide for optimizing Celery execution.

For example configuring prefetch multiplier (number of messages that are prefetched in a bulk mode from broker to worker) can result in significant speed up (more than 10 times based on my experience). In order to find the right and balanced solution for your cluster, you will need to experiment a bit.

Preparing development environment¶

To prepare your environment make sure that you have all development requirements installed. Check shipped Makefile for prepared commands that can be directly issued.

If you would like to create a virtual environment not to install (possibly) unwanted requirements on your system run:

make venv

To enter the prepared virtual environment run:

source venv/bin/activate

Now make sure that you have installed all development requirements:

make devenv

Now you are ready to hack! B-)

Tests¶

Selinon come with test suites. If you make changes, make sure that the test suite passes:

make check

The command above will run test suite and report any unexpected behaviour. It will also run linters and some code-quality related tools to ensure your changes look good.

If you make any changes, make sure that the test suite passes before your pull request. If you would like to test changes in some specific situations, Selinon demo could be a great starter to point to some specific use cases.

And not to forget… If you make any improvements in the source code, feel free to add your name to CONTRIBUTORS.txt file.

Package documentation¶

Automatically generated package documentation is available here