Forem: orkes

Netflix Conductor : Handling retries and timeout

orkes — Tue, 25 Jan 2022 18:39:54 +0000

What is Conductor

Conductor is a Microservices orchestration platform from Netflix, released under Apache 2.0 Open Source License.

Design for failures

Failures and service degradation are the fact of any system, this is especially true with large interconnected systems running in cloud. Conductor is designed with principles that systems can and will go down, degrade in performance and any dependencies should be able to handle such failures.

Tasks in Conductor

Conductor workflows are orchestration of many activities known as task. Each task represents a (ideally) stateless worker that given a specific input does the work and produces output. The tasks are typically running outside of Conductor server and there are many factors that could effect their availability.

Designing for failures

Conductor allows you to define your stateful applications that can handle failures and temporary degradation of services and without having to write code for that.

Configuring tasks to handle failures

Each task in Conductor can be configured how it responds to availability events such as: 1) Failures 2) Timeouts and 3) Rate limits
Here is a sample task defintion:

{
  "createdBy": "user",
  "name": "sample_task_name_1",
  "description": "This is a sample task for demo",
  "responseTimeoutSeconds": 10,
  "timeoutSeconds": 30,
  "timeoutPolicy": "TIME_OUT_WF",
  "retryCount": 3,
  "retryLogic": "FIXED",
  "retryDelaySeconds": 5,
  "rateLimitPerFrequency": 0,
  "rateLimitFrequencyInSeconds": 1
}

retry* parameters specify how to handle cases where the task execution fails and retries can be configured to be with fixed delay or exponential backoff. Similarly timeout* parameters specify how time time to give for task to complete execution and if the task should be marked as 'Timed Out' if it runs longer than that.

More Details

https://orkes.io/content/docs/how-tos/task-configurations

TIL: You can use JQ in Conductor

orkes — Thu, 09 Dec 2021 20:59:05 +0000

What is Conductor

Conductor is a Microservices orchestration platform from Netflix, released under Apache 2.0 Open Source License.

What is jq

jq (https://stedolan.github.io/jq/) is like sed for JSON data - you can use it to slice and filter and map and transform structured data with the same ease that sed, awk, grep and friends let you play with text.

Conductor + JQ

Conductor allows you to create tasks that can process JSON payloads in your workflows without writing custom code using jq.

Use cases

JSON is a popular format of choice for data-interchange. It is widely used in web and server applications, document storage, API I/O etc. It’s also used within Conductor to define workflow and task definitions and passing data and state between tasks and workflows. This makes a tool like JQ a natural fit for processing task related data. Some common usages within Conductor includes, working with HTTP task, JOIN tasks or standalone tasks that try to transform data from the output of one task to the input of another.

Configuration

Here is an example of a JSON_JQ_TRANSFORM task. The inputParameters attribute is expected to have a value object
that has the following

A list of key value pair objects denoted key1/value1, key2/value2 in the example below. Note the key1/value1 are
arbitrary names used in this example.
A key with the name queryExpression, whose value is a JQ expression. The expression will operate on the value of
the inputParameters attribute. In the example below, the inputParameters has 2 inner objects named by attributes
key1 and key2, each of which has an object that is named value1 and value2. They have an associated array of
strings as values, "a", "b" and "c", "d". The expression key3: (.key1.value1 + .key2.value2) concat's the 2
string arrays into a single array against an attribute named key3

{
  "name": "jq_example_task",
  "taskReferenceName": "my_jq_example_task",
  "type": "JSON_JQ_TRANSFORM",
  "inputParameters": {
    "key1": {
      "value1": [
        "a",
        "b"
      ]
    },
    "key2": {
      "value2": [
        "c",
        "d"
      ]
    },
    "queryExpression": "{ key3: (.key1.value1 + .key2.value2) }"
  }
}

The execution of this example task above will provide the following output. The resultList attribute stores the full
list of the queryExpression result. The result attribute stores the first element of the resultList. An
optional error
attribute along with a string message will be returned if there was an error processing the query expression.

{
  "result": {
    "key3": [
      "a",
      "b",
      "c",
      "d"
    ]
  },
  "resultList": [
    {
      "key3": [
        "a",
        "b",
        "c",
        "d"
      ]
    }
  ]
}

Input Configuration

Attribute	Description
name	Task Name. A unique name that is descriptive of the task function
taskReferenceName	Task Reference Name. A unique reference to this task. There can be multiple references of a task within the same workflow definition
type	Task Type. In this case, JSON_JQ_TRANSFORM
inputParameters	The input parameters that will be supplied to this task. The parameters will be a JSON object of atleast 2 attributes, one of which will be called queryExpression. The others are user named attributes. These attributes will be accessible by the JQ query processor
inputParameters/user-defined-key(s)	User defined key(s) along with values.
inputParameters/queryExpression	A JQ query expression

Output Configuration

Attribute	Description
result	The first results returned by the JQ expression
resultList	A List of results returned by the JQ expression
error	An optional error message, indicating that the JQ query failed processing

Want to learn more about JQ?

See tutorials at https://stedolan.github.io/jq/tutorial/

Go: Netflix Conductor Worker

orkes — Sun, 05 Dec 2021 16:43:23 +0000

What is Conductor

Conductor is a Microservices orchestration platform from Netflix, released under Apache 2.0 Open Source License.

Download and start Conductor server in 5 minutes

Quickstart

Install Golang package

go get github.com/netflix/conductor/client/go

Implementing a Task a Worker

task package provides the types used to implement the worker. Here is a reference worker implementation:

package task

import (
    "fmt"
)

// Implementation for "task_1"
func Task_1_Execution_Function(t *task.Task) (taskResult *task.TaskResult, err error) {
    log.Println("Executing Task_1_Execution_Function for", t.TaskType)

    //Do some logic
    taskResult = task.NewTaskResult(t)

    output := map[string]interface{}{"task":"task_1", "key2":"value2", "key3":3, "key4":false}
    taskResult.OutputData = output
    taskResult.Status = "COMPLETED"
    err = nil

    return taskResult, err
}

Worker Polling

Here is an example that shows how to start polling for tasks after defining the tasks.

package main

import (
    "github.com/netflix/conductor/client/go"
    "github.com/netflix/conductor/client/go/task/sample"
)

func main() {
    c := conductor.NewConductorWorker("http://localhost:8080", 1, 10000)

    c.Start("task_1", "", sample.Task_1_Execution_Function, false)
    c.Start("task_2", "mydomain", sample.Task_2_Execution_Function, true)
}

NewConductorWoker parameters

baseUrl: Server address.
threadCount: No. of threads. Number of threads should be at-least same as the number of workers
pollingInterval: Time in millisecond between subsequent polls

See
https://github.com/Netflix/conductor/tree/main/polyglot-clients/go
for the source code and follow us on GitHub for updates.

Netflix Conductor Workers in Python

orkes — Thu, 02 Dec 2021 23:47:17 +0000

What is Conductor

Conductor is a Microservices orchestration platform from Netflix, released under Apache 2.0 Open Source License.

Install the python client

virtualenv conductorclient
source conductorclient/bin/activate
cd ../conductor/client/python
python setup.py install

Implement a Task Worker

ConductorWorker class is used to implement task workers.
The following script shows how to bring up two task workers named book_flight and book_car:

from __future__ import print_function
from conductor.ConductorWorker import ConductorWorker

def book_flight_task(task):
    return {'status': 'COMPLETED', 'output': {'booking_ref': 2341111, 'airline': 'delta'}, 'logs': ['trying delta', 'skipping aa']}

def book_car_task(task):
    return {'status': 'COMPLETED', 'output': {'booking_ref': "84545fdfd", 'agency': 'hertz'}, 'logs': ['trying hertz']}

def main():
    print('Starting Travel Booking workflows')
    cc = ConductorWorker('http://localhost:8080/api', 1, 0.1)
    cc.start('book_flight', book_flight_task, False)
    cc.start('book_car', book_car_task, True)

if __name__ == '__main__':
    main()

`ConductorWorker` parameters

server_url: str
    The url to the server hosting the conductor api.
    Ex: 'http://localhost:8080/api'

thread_count: int
    The number of threads that will be polling for and
    executing tasks in case of using the start method.

polling_interval: float
    The number of seconds that each worker thread will wait
    between polls to the conductor server.

worker_id: str, optional
    The worker_id of the worker that is going to execute the
    task. For further details, refer to the documentation
    By default, it is set to hostname of the machine

`start` method parameters

taskType: str
    The name of the task that the worker is looking to execute

exec_function: function
    The function that the worker will execute. The function
    must return a dict with the `status`, `output` and `logs`
    keys present. If this is not present, an Exception will be
    raised

wait: bool
    Whether the worker will block execution of further code.
    Since the workers are being run in daemon threads, when the
    program completes execution, all the threads are destroyed.
    Setting wait to True prevents the program from ending.
    If multiple workers are being called from the same program,
    all but the last start call but have wait set to False.
    The last start call must always set wait to True. If a
    single worker is being called, set wait to True.

domain: str, optional
    The domain of the task under which the worker will run. For
    further details refer to the conductor server documentation
    By default, it is set to None

See
https://github.com/Netflix/conductor/tree/main/polyglot-clients/python
for the source code and follow us on GitHub for updates.

Forem: orkes

Netflix Conductor : Handling retries and timeout

What is Conductor

Design for failures

Tasks in Conductor

Designing for failures

Configuring tasks to handle failures

More Details

TIL: You can use JQ in Conductor

What is Conductor

What is jq

Conductor + JQ

Use cases

Configuration

Input Configuration

Output Configuration

Want to learn more about JQ?

Go: Netflix Conductor Worker

What is Conductor

Download and start Conductor server in 5 minutes

Install Golang package

Implementing a Task a Worker

Worker Polling

NewConductorWoker parameters

Netflix Conductor Workers in Python

What is Conductor

Install the python client

Implement a Task Worker

ConductorWorker parameters

start method parameters

`ConductorWorker` parameters

`start` method parameters