conductor

package
v2.5.0+incompatible Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 2, 2019 License: Apache-2.0 Imports: 6 Imported by: 0

README

Go client for Conductor

Go client for Conductor provides two sets of functions:

  1. Workflow Management APIs (start, terminate, get workflow status etc.)
  2. Worker execution framework

Prerequisites

Go must be installed and GOPATH env variable set. Directory $GOPATH/src/conductor must not be in use.

Install

./install.sh

This will create a Go project under $GOPATH/src/conductor and download any dependencies. It can then be ran:

go run $GOPATH/src/conductor/startclient/startclient.go

Install and Run

./install_and_run.sh

This will create a Go project under $GOPATH/src/conductor and download any dependencies. In addition, it will run the go application listed under startclient/startclient.go

Uninstall

WARNING: This will simply remove the $GOPATH/src/conductor directory where it has installed so if other work is there, it will be deleted. Use with caution.

./uninstall.sh

Using Workflow Management API

Go struct ConductorHttpClient provides client API calls to the conductor server to start and manage workflows and tasks.

Example
package main

import (
    "conductor"
)

func main() {
    conductorClient := conductor.NewConductorHttpClient("http://localhost:8080")
    
    // Example API that will print out workflow definition meta
    conductorClient.GetAllWorkflowDefs()
}

Task Worker Execution

Task Worker execution APIs facilitates execution of a task worker using go. The API provides necessary tools to poll for tasks at a specified interval and executing the go worker in a separate goroutine.

Example

The following go code demonstrates workers for tasks "task_1" and "task_2".

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
}

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

    //Do some logic
    taskResult = task.NewTaskResult(t)
    
    output := map[string]interface{}{"task":"task_2", "key2":"value2", "key3":3, "key4":false}
    taskResult.OutputData = output
    taskResult.Status = "COMPLETED"
    err = nil

    return taskResult, err
}

Then main application to utilize these workers

package main

import (
    "conductor"
    "conductor/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", sample.Task_2_Execution_Function, true)
}

Note: For the example listed above the example task implementations are in conductor/task/sample package. Real task implementations can be placed in conductor/task directory or new subdirectory.

Documentation

Overview

Copyright 2017 Netflix, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2017 Netflix, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ConductorHttpClient 

type ConductorHttpClient struct {
	// contains filtered or unexported fields
}

func NewConductorHttpClient 

func NewConductorHttpClient(baseUrl string) *ConductorHttpClient

func (*ConductorHttpClient) AckTask 

func (c *ConductorHttpClient) AckTask(taskType string, workerid string) (string, error)

func (*ConductorHttpClient) CreateWorkflowDef 

func (c *ConductorHttpClient) CreateWorkflowDef(workflowDefBody string) (string, error)

func (*ConductorHttpClient) GetAllTaskDefs 

func (c *ConductorHttpClient) GetAllTaskDefs() (string, error)

func (*ConductorHttpClient) GetAllTasksInQueue 

func (c *ConductorHttpClient) GetAllTasksInQueue() (string, error)

func (*ConductorHttpClient) GetAllWorkflowDefs 

func (c *ConductorHttpClient) GetAllWorkflowDefs() (string, error)

func (*ConductorHttpClient) GetRunningWorkflows 

func (c *ConductorHttpClient) GetRunningWorkflows(workflowName string, version int, startTime float64, endTime float64) (string, error)

func (*ConductorHttpClient) GetTask 

func (c *ConductorHttpClient) GetTask(taskId string) (string, error)

func (*ConductorHttpClient) GetTaskDef 

func (c *ConductorHttpClient) GetTaskDef(taskDefName string) (string, error)

func (*ConductorHttpClient) GetTaskQueueSizes 

func (c *ConductorHttpClient) GetTaskQueueSizes(taskNames string) (string, error)

func (*ConductorHttpClient) GetWorkflow 

func (c *ConductorHttpClient) GetWorkflow(workflowId string, includeTasks bool) (string, error)

func (*ConductorHttpClient) GetWorkflowDef 

func (c *ConductorHttpClient) GetWorkflowDef(workflowName string, version int) (string, error)

func (*ConductorHttpClient) PauseWorkflow 

func (c *ConductorHttpClient) PauseWorkflow(workflowId string) (string, error)

func (*ConductorHttpClient) PollForTask 

func (c *ConductorHttpClient) PollForTask(taskType string, workerid string) (string, error)

func (*ConductorHttpClient) RegisterTaskDefs 

func (c *ConductorHttpClient) RegisterTaskDefs(taskDefsMeta string) (string, error)

func (*ConductorHttpClient) RemoveTaskFromQueue 

func (c *ConductorHttpClient) RemoveTaskFromQueue(taskType string, taskId string) (string, error)

func (*ConductorHttpClient) RerunWorkflow 

func (c *ConductorHttpClient) RerunWorkflow(workflowId string, rerunWorkflowRequest string) (string, error)

func (*ConductorHttpClient) RestartWorkflow 

func (c *ConductorHttpClient) RestartWorkflow(workflowId string) (string, error)

func (*ConductorHttpClient) ResumeWorkflow 

func (c *ConductorHttpClient) ResumeWorkflow(workflowId string) (string, error)

func (*ConductorHttpClient) SkipTaskFromWorkflow 

func (c *ConductorHttpClient) SkipTaskFromWorkflow(workflowId string, taskReferenceName string, skipTaskRequestBody string) (string, error)

func (*ConductorHttpClient) StartWorkflow 

func (c *ConductorHttpClient) StartWorkflow(workflowName string, version int, correlationId string, inputJson string) (string, error)

func (*ConductorHttpClient) TerminateWorkflow 

func (c *ConductorHttpClient) TerminateWorkflow(workflowId string, reason string) (string, error)

func (*ConductorHttpClient) UnRegisterTaskDef 

func (c *ConductorHttpClient) UnRegisterTaskDef(taskDefName string) (string, error)

func (*ConductorHttpClient) UnRegisterWorkflowDef added in v1.10.13 

func (c *ConductorHttpClient) UnRegisterWorkflowDef(workflowDefName string, version int) (string, error)

func (*ConductorHttpClient) UpdateTask 

func (c *ConductorHttpClient) UpdateTask(taskBody string) (string, error)

func (*ConductorHttpClient) UpdateTaskDef 

func (c *ConductorHttpClient) UpdateTaskDef(taskDefMeta string) (string, error)

func (*ConductorHttpClient) UpdateWorkflowDefs 

func (c *ConductorHttpClient) UpdateWorkflowDefs(workflowDefsBody string) (string, error)

type ConductorWorker 

type ConductorWorker struct {
	ConductorHttpClient *ConductorHttpClient
	ThreadCount         int
	PollingInterval     int
}

func NewConductorWorker 

func NewConductorWorker(baseUrl string, threadCount int, pollingInterval int) *ConductorWorker

func (*ConductorWorker) Execute 

func (c *ConductorWorker) Execute(taskData string, executeFunction func(t *task.Task) (*task.TaskResult, error))

func (*ConductorWorker) PollAndExecute 

func (c *ConductorWorker) PollAndExecute(taskType string, executeFunction func(t *task.Task) (*task.TaskResult, error))

func (*ConductorWorker) Start 

func (c *ConductorWorker) Start(taskType string, executeFunction func(t *task.Task) (*task.TaskResult, error), wait bool)

Source Files

View all Source files

Directories

Path Synopsis
sample

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.