faasflow

package module
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2019 License: MIT Imports: 4 Imported by: 0

README

Faas-flow - Function Composition for Openfaas

Build Status GoDoc OpenTracing Badge OpenFaaS

  • Pure              FaaS with Openfaas
  • Fast               Built with Go
  • Secured        With HMAC
  • Stateless      By design
  • Tracing         With open-tracing
  • Available      As faas-flow template

Faas-flow tower visualizes and monitors flow function

Overview

Faas-flow allows you to realize OpenFaaS function composition with ease. By defining a simple pipeline, you can orchestrate multiple functions without having to worry about internals

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
    flow.SyncNode().Apply("Func1").Apply("Func2")
    return
}

After building and deploying, it will give you a openfaas function that orchestrates calling Func2 with the output of Func1

Pipeline Definition

By supplying a number of pipeline operators, complex compostion can be achieved with little work: alt overview

The above pipelines can be achieved with little, but powerfull code:

SYNC Chain

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
        flow.SyncNode().Apply("func1").Apply("func2").
                Modify(func(data []byte) ([]byte, error) {
                        // do something 
                        return data, nil
                })
        return
}

ASYNC Chain

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
        dag := flow.Dag()
        dag.Node("n1").Apply("func1")
        dag.Node("n2").Apply("func2").
                Modify(func(data []byte) ([]byte, error) {
                        // do something
                        return data, nil
                })
        dag.Node("n3").Callback("http://gateway:8080/function/fake-storage")
        dag.Edge("n1", "n2")
        dag.Edge("n2", "n3")
        return
}

PARALLEL Branching

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
        dag := flow.Dag()
        dag.Node("n1").Modify(func(data []byte) ([]byte, error) {
                // do something
                return data, nil
        })
        dag.Node("n2").Apply("func1")
        dag.Node("n3").Apply("func2").Modify(func(data []byte) ([]byte, error) {
                // do something
                return data, nil
        })
        dag.Node("n4", faasflow.Aggregator(func(data map[string][]byte) ([]byte, error) {
                // aggregate branch result data["n2"] and data["n3"]
                return []byte(""), nil
        })).Callback("http://gateway:8080/function/fake-storage")

        dag.Edge("n1", "n2")
        dag.Edge("n1", "n3")
        dag.Edge("n2", "n4")
        dag.Edge("n3", "n4")
        return
}

DYNAMIC Branching

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
        dag := flow.Dag()
        dag.Node("n1").Modify(func(data []byte) ([]byte, error) {
                // do something
                return data, nil
        })
        conditionalDags := dag.ConditionalBranch("C",
                []string{"c1", "c2"}, // possible conditions
                func(response []byte) []string {
                        // for each returned condition the corresponding branch will execute
                        // this function executes in the runtime of condition C
                        return []string{"c1", "c2"}
                },
                faasflow.Aggregator(func(data map[string][]byte) ([]byte, error) {
                        // aggregate all dynamic branches results
                        return []byte(""), nil
                }),
        )

        conditionalDags["c2"].Node("n1").Apply("func1").Modify(func(data []byte) ([]byte, error) {
                // do something
                return data, nil
        })
        foreachDag := conditionalDags["c1"].ForEachBranch("F",
                func(data []byte) map[string][]byte {
                        // for each returned key in the hashmap a new branch will be executed
                        // this function executes in the runtime of foreach F
                        return map[string][]byte{"f1": data, "f2": data}
                },
                faasflow.Aggregator(func(data map[string][]byte) ([]byte, error) {
                        // aggregate all dynamic branches results
                        return []byte(""), nil
                }),
        )
        foreachDag.Node("n1").Modify(func(data []byte) ([]byte, error) {
                // do something
                return data, nil
        })
        dag.Node("n2").Callback("http://gateway:8080/function/fake-storage")
        dag.Edge("n1", "C")
        dag.Edge("C", "n2")
}

Full implementions of the above examples are available here

Faas-flow Design

The current design consideration are made based on the below goals

  1. Leverage the openfaas platform
  2. Not to violate the notions of function
  3. Provide flexibility, scalability and adaptibility
Just as function as any other

Faas-flow is deployed and provisioned just like any other openfaas function. It allows faas-flow to take advantage of rich functionalities available on Openfaas. faas-flow provide a openfaas template and just like any other openfaas function it can be deployed with faas-cli
alt its a function

Adapter pattern for zero intrumenttaion in code

Faas-flow function follow the adapter pattern. Here the adaptee is the functions and the adapter is faas-flow. For each node execution, faas-flow handle the calls to the functions. Once the execution is over, it forwards an event to itself. This way the arrangement logic is seperated from the functions and is implemented in the adapter. Compositions need no code instrumentations, making functions completly independent of the compositions details alt function is independent of composition

Aggregate pattern as chaining

Aggregatation of seperate function calls are done as chaining. Multiple functions can be called from a single node with order maintained as per the chain. This way one execution node can be implemented as an aggregator function that invokes multiple functions, collects the results, optionally applies business logic, and returns a consolidated response to the client or forward to next nodes. Faas-flow fuses the adapter pattern and aggregate pattern to support more complex usecases alt aggregation

Event driven iteration

Openfaas uses Nats for event delivery and faas-flow leverages openfaas platform. Node execution in faas-flow starts by a completion event of one or more previous nodes. A completion event denotes that all the previous dependent nodes have completed. The event carries the execution state and identifies the next node to execute. With events faas-flow asynchronously carry-on execution of nodes by iterating itself over and over till all nodes are executed alt iteration

3rd party KV store for coordination

When executing branches, one node is dependent on more than one predecessor nodes. In that scenario, the event for completion is generated by coordination of earlier nodes. Like any distributed system the coordination is achieved via a centralized service. Faas-flow keeps the logic of the coordination controller inside of faas-flow implementation and lets the user use any external synchronous KV store by implementing StateStore alt coordination

3rd party Storage for intermediate data

Results from function execution and intermediate data can be handled by the user manually. Faas-flow provides data-store for intermediate result storage. It automatically initializes, store, retrieve and remove data between nodes. This fits great for data processing applications. Faas-flow keeps the logic of storage controller inside of Faas-flow implementation and lets the user use any external object storage by implementing DataStore alt storage

Faas-flow design is not fixed and like any good design it is evolving. Please contribute to make it better.

Getting Started

This example implements a very simple flow to Greet

Get template

Pull faas-flow template with the faas-cli

faas template pull https://github.com/s8sg/faas-flow
Create new flow function

Create a new function using faas-flow template

faas new greet --lang faas-flow
Edit stack

Edit function stack file greet.yml

  greet:
    lang: faas-flow
    handler: ./greet
    image: greet:latest
    environment:
      read_timeout: 120 # A value larger than `max` of all execution times of Nodes
      write_timeout: 120 # A value larger than `max` of all execution times of Nodes
      write_debug: true
      combine_output: false
      workflow_name: "greet" # The name of the flow function, faasflow use this to forward completion event
    environment_file:
      - flow.yml
Add configuration

Add a seperate file flow.yml with faas-flow related configuration.

environment:
  gateway: "gateway:8080" # The address of openfaas gateway, faasflow use this to forward completion event
  # gateway: "gateway.openfaas:8080" # For K8s 
  enable_tracing: false # tracing allow to trace internal node execution with opentracing
  enable_hmac: true # hmac adds extra layer of security by validating the event source
Edit function defnition

Edit greet/handler.go and Update Define()

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
      flow.SyncNode().
	  Modify(func(data []byte) ([]byte, error) {
	  	result := "Hello " + string(data)
		return []byte(result), nil
	  })
      return nil
}
Build and Deploy

Build and deploy

faas build
faas deploy

This function will generate one Synchronous node

Modify("name") -> Hello name

All calls will be performed in one single execution of the flow function and result will be returned to the callee

Note: For flow that has more than one nodes, faas-flow doesn't return any response. External storage or callback can be used to retrive async result

Invoke
echo "Adam" | faas invoke greet

Request Tracking by ID

For each new request faas-flow generates a unique Request Id for the flow. The same Id is used when logging

2018/08/13 07:51:59 [Request `bdojh7oi7u6bl8te4r0g`] Created
2018/08/13 07:52:03 [Request `bdojh7oi7u6bl8te4r0g`] Received

The assigned requestId is set on the response header X-Faas-Flow-Reqid

Request Tracing by Open-Tracing

Request tracing can be retrived from trace_server once enabled. Tracing is the best way to monitor flows and execution status of each nodes for each requests

Edit flow.yml

Enable tracing and add trace server as:

      enable_tracing: true
      trace_server: "jaegertracing:5775"
Start The Trace Server

jaeger (opentracing-1.x) used for tracing backend
Quick start with jaegertracing: https://www.jaegertracing.io/docs/1.8/getting-started/

Use faas-flow-tower

Retrive the requestID from X-Faas-Flow-Reqid header of response

Below is an example of tracing information for example-branching-in-faas-flow in faas-flow-tower
alt monitoring

Use of context

Context can be used inside definition for differet usecases. Context provide verious information such as:
HttpQuery to retrivbe original request queries
State to get flow state
Node to get current node
along with that it wraps the DataStore to store data

Store data in context with DataStore

Context uses DataStore to store/retrive data. User can do the same by calling Get(), Set() and Del() from context:

     flow.SyncNode().
     Modify(func(data []byte) {
	  // parse data and set to be used later
          // json.Unmarshal(&req, data)
          context.Set("commitsha", req.Sha)
     }).
     Apply("myfunc").
     Modify(func(data []byte) {
          // retrived the data that was set in the context
          commitsha, _ = context.GetString("commitsha")
          // use the query
     })
Geting Http Query to Workflow:

Http Query to flow can be used from context as

    flow.SyncNode().Apply("myfunc", Query("auth-token", context.Query.Get("token"))). // pass as a function query
     	 Modify(func(data []byte) {
          	token = context.Query.Get("token") // get query inside modifier
     	 })
Other from context:

Node, requestId, State is provided by the context

   currentNode := context.GetNode()
   requestId := context.GetRequestId()
   state := context.State

for more details check `faas-flow-GoDoc

External StateStore for coordination controller

Any DAG which has a branch needs coordination for nodes completion events. Faas-flow implements coordination controller which allows user to use any external Synchoronous KV store. User can define custom state-store with StateStore interface.

type StateStore interface {
        // Configure the StateStore with flow name and request ID
        Configure(flowName string, requestId string)
        // Initialize the StateStore (called only once in a request span)
        Init() error
        // Set a value (override existing, or create one)
        Set(key string, value string) error
        // Get a value
        Get(key string) (string, error)
        // Compare and Update a value
        Update(key string, oldValue string, newValue string) error
        // Cleanup all the resorces in StateStore (called only once in a request span)
        Cleanup() error
}

A StateStore can be implemented with any KV Store that provides Synchronization. The implemented StateStore can be set with DefineStateStore() at function/handler.go:

// DefineStateStore provides the override of the default StateStore
func DefineStateStore() (faasflow.StateStore, error) {
        consulss, err := consulStateStore.GetConsulStateStore(os.Getenv("consul_url"), os.Getenv("consul_dc"))
        return consulss, err
}
Available state-stores:

External DataStore for storage controller

Faas-flow uses the DataStore to store partially completed data between nodes and request context data. Faas-flow implements storage controller to handle storage that allows user to use any external object store. User can define custom data-store with DataStore interface.

 type DataStore interface {
        // Configure the DaraStore with flow name and request ID
        Configure(flowName string, requestId string)
        // Initialize the DataStore (called only once in a request span)
        Init() error
        // Set store a value for key, in failure returns error
        Set(key string, value string) error
        // Get retrives a value by key, if failure returns error
        Get(key string) (string, error)
        // Del delets a value by a key
        Del(key string) error
        // Cleanup all the resorces in DataStore
        Cleanup() error
 }

Data Store can be implemented and set by user at the DefineDataStore() at function/handler.go:

// ProvideDataStore provides the override of the default DataStore
func DefineDataStore() (faasflow.DataStore, error) {
        // initialize minio DataStore
        miniods, err := minioDataStore.InitFromEnv()
        return miniods, err
}
Available data-stores:

Cleanup with Finally()

Finally provides an efficient way to perform post execution steps of the flow. If specified Finally() invokes in case of both failure and success of the flow. A Finally method can be set as:

func Define(flow *faasflow.Workflow, context *faasflow.Context) (err error) {
     // Define flow
     flow.SyncNode().Modify(func(data []byte) {
	  // parse data and set to be used later
          // json.Unmarshal(&req, data)
          context.Set("commitsha", req.Sha)
     }).
     Apply("myfunc").Modify(func(data []byte) {
          // retrived the data in different node from context
          commitsha, _ = context.GetString("commitsha")
     })
     flow.OnFailure(func(err error) {
          // failure handler
     }) 
     flow.Finally(func() {
          // delete the state resource
          context.Del("commitsha")
     })
}

Contribute:

Issue/Suggestion Create an issue at faas-flow-issue.
ReviewPR/Implement Create Pull Request at faas-flow-pr.

Documentation

Index

Constants

View Source 
const (
	// StateSuccess denotes success state
	StateSuccess = "success"
	// StateFailure denotes failure state
	StateFailure = "failure"
	// StateOngoing denotes onging satte
	StateOngoing = "ongoing"
)

Variables

View Source 
var (
	// Execution specify a edge doesn't forwards a data
	// but rather mention a execution direction
	Execution = InvokeEdge()
)

Functions

This section is empty.

Types

type BranchOption added in v0.4.0 

type BranchOption func(*BranchOptions)

func Aggregator 

func Aggregator(aggregator sdk.Aggregator) BranchOption

Aggregator aggregates all outputs into one

func Forwarder 

func Forwarder(forwarder sdk.Forwarder) BranchOption

Forwarder encodes request based on need for children vertex by default the data gets forwarded as it is

func InvokeEdge added in v0.4.0 

func InvokeEdge() BranchOption

InvokeEdge denotes a edge doesn't forwards a data, but rather provides only an execution flow

type BranchOptions added in v0.4.0 

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

BranchOptions options for branching in DAG

type Context 

type Context struct {
	Query url.Values // provides request Query
	State string     // state of the request
	Name  string     // name of the faas-flow

	NodeInput map[string][]byte // stores inputs form each node
	// contains filtered or unexported fields
}

Context execution context and execution state

func CreateContext 

func CreateContext(id string, node string, name string,
	dstore DataStore) *Context

CreateContext create request context (used by template)

func (*Context) Del 

func (context *Context) Del(key string) error

Del deletes a value from the context using DataStore

func (*Context) Get 

func (context *Context) Get(key string) (interface{}, error)

Get retrive a value from the context using DataStore

func (*Context) GetBool 

func (context *Context) GetBool(key string) bool

GetBool retrive a boolean value from the context using DataStore

func (*Context) GetBytes 

func (context *Context) GetBytes(key string) []byte

GetBytes retrive a byte array from the context using DataStore

func (*Context) GetInt 

func (context *Context) GetInt(key string) int

GetInt retrive a integer value from the context using DataStore

func (*Context) GetNode 

func (context *Context) GetNode() string

GetPhase return the node no

func (*Context) GetRequestId 

func (context *Context) GetRequestId() string

GetRequestId returns the request id

func (*Context) GetString 

func (context *Context) GetString(key string) string

GetString retrive a string value from the context using DataStore

func (*Context) Set 

func (context *Context) Set(key string, data interface{}) error

Set put a value in the context using DataStore

type Dag added in v0.4.0 

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

func NewDag added in v0.4.0 

func NewDag() *Dag

NewDag creates a new dag seperately from pipeline

func (*Dag) Append added in v0.4.0 

func (this *Dag) Append(dag *Dag)

Append generalizes a seperate dag by appending its properties into current dag. Provided dag should be mutually exclusive

func (*Dag) ConditionalBranch added in v0.4.0 

func (this *Dag) ConditionalBranch(vertex string, conditions []string, condition sdk.Condition,
	options ...BranchOption) (conditiondags map[string]*Dag)

ConditionalBranch composites multiple dags as a subdag which executes for a conditions matched and returns the set of dags based on the condition passed

func (*Dag) Edge added in v0.4.0 

func (this *Dag) Edge(from, to string, opts ...BranchOption)

Edge adds a directed edge between two vertex as <from>-><to>

func (*Dag) ForEachBranch added in v0.4.0 

func (this *Dag) ForEachBranch(vertex string, foreach sdk.ForEach, options ...BranchOption) (dag *Dag)

ForEachBranch composites a subdag which executes for each value It returns the subdag that will be executed for each value

func (*Dag) Node added in v0.4.0 

func (this *Dag) Node(vertex string, options ...BranchOption) *Node

Node adds a new vertex by id

func (*Dag) SubDag added in v0.4.0 

func (this *Dag) SubDag(vertex string, dag *Dag)

SubDag composites a seperate dag as a node.

type DataStore 

type DataStore interface {
	// Configure the DaraStore with flow name and request ID
	Configure(flowName string, requestId string)
	// Initialize the DataStore (called only once in a request span)
	Init() error
	// Set store a value for key, in failure returns error
	Set(key string, value string) error
	// Get retrives a value by key, if failure returns error
	Get(key string) (string, error)
	// Del delets a value by a key
	Del(key string) error
	// Cleanup all the resorces in DataStore
	Cleanup() error
}

DataStore for Storing Data

type Node added in v0.4.0 

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

func (*Node) Apply added in v0.4.0 

func (node *Node) Apply(function string, opts ...Option) *Node

Apply adds a new function to the given vertex

func (*Node) Callback added in v0.4.0 

func (node *Node) Callback(url string, opts ...Option) *Node

Callback adds a new callback to the given vertex

func (*Node) Modify added in v0.4.0 

func (node *Node) Modify(mod sdk.Modifier) *Node

Modify adds a new modifier to the given vertex

type Option 

type Option func(*Options)
func Header(key, value string) Option

Header Specify a header in a http call

func OnFailure 

func OnFailure(handler sdk.FuncErrorHandler) Option

OnFailure Specify a function failure handler

func OnReponse 

func OnReponse(handler sdk.RespHandler) Option

OnResponse Specify a response handler for function and callback

func Query 

func Query(key string, value ...string) Option

Query Specify a query parameter in a http call

func RequestHandler added in v0.4.0 

func RequestHandler(handler sdk.ReqHandler) Option

RequestHandler Specify a request handler for function and callback request

type Options 

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

Options options for operation execution

type StateStore 

type StateStore interface {
	// Configure the StateStore with flow name and request ID
	Configure(flowName string, requestId string)
	// Initialize the StateStore (called only once in a request span)
	Init() error
	// Set a value (override existing, or create one)
	Set(key string, value string) error
	// Get a value
	Get(key string) (string, error)
	// Compare and Update a value
	Update(key string, oldValue string, newValue string) error
	// Cleanup all the resorces in StateStore (called only once in a request span)
	Cleanup() error
}

StateStore for saving execution state

type Workflow 

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

func GetWorkflow added in v0.4.0 

func GetWorkflow() *Workflow

GetWorkflow initiates a flow with a pipeline

func (*Workflow) Dag added in v0.4.0 

func (flow *Workflow) Dag() *Dag

Dag provides the workflow dag object

func (*Workflow) Finally 

func (flow *Workflow) Finally(handler sdk.PipelineHandler)

Finally sets an execution finish handler routine it will be called once the execution has finished with state either Success/Failure

func (*Workflow) GetPipeline 

func (flow *Workflow) GetPipeline() *sdk.Pipeline

GetPipeline expose the underlying pipeline object

func (*Workflow) OnFailure 

func (flow *Workflow) OnFailure(handler sdk.PipelineErrorHandler)

OnFailure set a failure handler routine for the pipeline

func (*Workflow) SetDag added in v0.4.0 

func (flow *Workflow) SetDag(dag *Dag)

SetDag apply a predefined dag, and override the default dag

func (*Workflow) SyncNode added in v0.4.0 

func (flow *Workflow) SyncNode(options ...BranchOption) *Node

SyncNode adds a new vertex named Sync

Source Files

View all Source files

Directories

Path Synopsis
doc
guide-executesync/example/syncflow
guide-helloworld/example/helloworld
guide-modifier/example/format
guide-modifier/example/makeuserid
guide-modifier/example/titleize
template
faas-flow command
faas-flow/function

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.