fnproject/flow-lib-go

Workflow orchestration for your Go functions using Fn Flow

Serverless Workflows with Go

Easily create serverless workflows directly in Go with the power of Fn Flow.

Quick Intro

Simply import this library into your go function, build and deploy onto Fn. Flows use the fdk-go to handle interacting with Fn, below is an example flow:

package main

import (
	"fmt"
	"strings"
	"time"

  	fdk "github.com/fnproject/fdk-go"
  	flows "github.com/fnproject/flow-lib-go"
)

func init() {
  	flows.RegisterAction(strings.ToUpper)
  	flows.RegisterAction(strings.ToLower)
}

func main() {
	fdk.Handle(flows.WithFlow(
    		fdk.HandlerFunc(func(ctx context.Context, r io.Reader, w io.Writer) {
      			cf := flows.CurrentFlow().CompletedValue("foo")
      			valueCh, errorCh := cf.ThenApply(strings.ToUpper).ThenApply(strings.ToLower).Get()
      			select {
      			case value := <-valueCh:
        			fmt.Fprintf(w, "Flow succeeded with value %v", value)
      			case err := <-errorCh:
        			fmt.Fprintf(w, "Flow failed with error %v", err)
      			case <-time.After(time.Minute * 1):
        			fmt.Fprintf(w, "Timed out!")
      			}
    		}),
  	)
}

Where do I go from here?

A variety of example use-cases is provided here.

FAQs

How are values serialized?

Go's gob serialization mechanism is used to encode/decode values for communication with the completer.

What kinds of values can be serialized?

Booleans, string, structs, arrays and slices are supported. Functions, closures and channels are not.

How are continuations serialized?

Since Go does not support serializing closures/functions due to its statically compiled nature, they are in fact not serialized at all. Go functions implementing a continuation need to be explicitly registered by calling flows.RegisterAction(actionFunction) typically inside the handler's init function. Registering actions assigns a unique and stable key that can be serialized and used to look up a pointer to the function during a continuation invocation.

Why do actions need to be registered?

See above.

Can I use closures or method receivers in my continuations?

No. Only continuation actions implemented with functions are supported, since they are stateless. No state will be serialized with a continuation. Although possible, invoking a method receiver is not currently supported inside continuations.

How does error-handling work?

Go allows functions to return error types in addition to a result via its support for multiple return values. If a continuation function returns a (non-nil) error as its second return value, its error message will be serialized and form the failed value of that stage.

If a panic occurs while invoking the continuation function, the panic value will be captured and the stage failed with the same value.

Can I invoke other fn functions?

Yes. flows.CurrentFlow().InvokeFunction("your_function_id", req), where the function ID is a value like 01CQV4NEGMNG8G00GZJ0000002 and can be resolved with the following command:

fn inspect function your_app your_function | grep fnproject.io/fn/invokeEndpoint

See here for a full example.

flow-lib-go's Issues

Build example failed

Error output

Step 6/10 : RUN cd /go/src/func/ && go build -o func
---> Running in 0dc0f0ce1e73
func/vendor/github.com/fnproject/flow-lib-go
vendor/github.com/fnproject/flow-lib-go/codecs.go:37:20: cannot convert c.ctx (type context.Context) to type fdk.Context:
context.Context does not implement fdk.Context (missing AppID method)
vendor/github.com/fnproject/flow-lib-go/codecs.go:37:34: invalid operation: fdk.Context(c.ctx).Config[appName] (type func() map[string]string does not support indexing)
vendor/github.com/fnproject/flow-lib-go/codecs.go:41:20: cannot convert c.ctx (type context.Context) to type fdk.Context:
context.Context does not implement fdk.Context (missing AppID method)
vendor/github.com/fnproject/flow-lib-go/codecs.go:41:34: invalid operation: fdk.Context(c.ctx).Config[fnID] (type func() map[string]string does not support indexing)
vendor/github.com/fnproject/flow-lib-go/codecs.go:61:18: cannot convert c.ctx (type context.Context) to type fdk.Context:
context.Context does not implement fdk.Context (missing AppID method)
vendor/github.com/fnproject/flow-lib-go/codecs.go:61:32: fdk.Context(c.ctx).Header.Get undefined (type func() http.Header has no field or method Get)

Triage

fdk-go has changed Context from a function to a interface.
https://github.com/fnproject/fdk-go/blob/master/fdk.go:
// Context contains all configuration for a function invocation
type Context interface {

But flow-lib-go is not updated accordingly.

Support invoking method receivers in continuations

It would be useful to be able to invoke method receivers on gob-serialized structs as part of a stage's continuation action. This mechanism could be used to carry some context state that is not currently possible using go functions.

Flow continuation fails when combining fdk handler to accept http input to function

With the provided flow example, add code from the fdk boiler plate to accept http input to the function as in the attached file. The stage continuation fails with the following output:

Enabled debugging
Awakened flow 26607dbd-66be-4eab-a2af-f09c43b43904
Handling continuation
panic: Failed to decode stage invocation request: EOF

goroutine 1 [running]:
func/vendor/github.com/fnproject/flow-lib-go.handleInvocation(0x8e79e0, 0xc42031cbe0)
/go/src/func/vendor/github.com/fnproject/flow-lib-go/invocations.go:86 +0x1ab
func/vendor/github.com/fnproject/flow-lib-go.WithFlow(0x8af880)
/go/src/func/vendor/github.com/fnproject/flow-lib-go/flows.go:108 +0x126
main.stringExample()
/go/src/func/func.go:58 +0x2d
main.myHandler(0x8e73a0, 0xc420070600, 0x8e3e60, 0xc42000e010, 0x8e3e80, 0xc42000e018)
/go/src/func/func.go:51 +0xcb
func/vendor/github.com/fnproject/fdk-go.HandlerFunc.Serve(0x8af878, 0x8e73a0, 0xc420070600, 0x8e3e60, 0xc42000e010, 0x8e3e80, 0xc42000e018)
/go/src/func/vendor/github.com/fnproject/fdk-go/fdk.go:19 +0x62
func/vendor/github.com/fnproject/fdk-go/utils.DoDefault(0x7f1e08c827c0, 0x8af878, 0x8e73a0, 0xc420070600, 0x8e3e60, 0xc42000e010, 0x8e3e80, 0xc42000e018)
/go/src/func/vendor/github.com/fnproject/fdk-go/utils/utils.go:63 +0xf9
func/vendor/github.com/fnproject/fdk-go/utils.Do(0x7f1e08c827c0, 0x8af878, 0xc4200240aa, 0x7, 0x8e3e60, 0xc42000e010, 0x8e3e80, 0xc42000e018)
/go/src/func/vendor/github.com/fnproject/fdk-go/utils/utils.go:49 +0x251
func/vendor/github.com/fnproject/fdk-go.Handle(0x8e4380, 0x8af878)
/go/src/func/vendor/github.com/fnproject/fdk-go/fdk.go:81 +0xce
main.main()
/go/src/func/func.go:31 +0x39

func.go.zip

Calling a stage without arguments causes the flow application to fail

The following code fails when attempting to execute the ThenRun stage, since it takes no arguments.

			cf := flows.CurrentFlow()
			s1 := cf.Delay(5 * time.Second)
			s2 := cf.Delay(2 * time.Second)
			s3 := cf.AnyOf(s1, s2).ThenRun(EmptyFunc)

Add support for various encodings (or pick another)

currently, gob encoding is used, but a new encoder/decoder is built for each conversion. From xp, and the gob docs themselves, there is a lot of overhead to build the encoder/decoder and it works much faster on streams (see docs: https://golang.org/pkg/encoding/gob/). I believe we're using this because gob can serialize funcs, but feel free to clarify this point. at least, we should build a encoder/decoder and re-use them across conversions to reduce latency, the impact is significant though performance wise. if possible, it may be better to just move to json which is slightly faster but not great (but stdlib), or something else.