feat: support function config (#192)

Author: RobertIndie

clients/gofs/gofs.go

@@ -73,6 +73,22 @@ type moduleWrapper struct {
 	registerErr error
 }
 
+func (m *moduleWrapper) AddInitFunc(initFunc func(context.Context) error) *moduleWrapper {
+	parentInit := m.initFunc
+	if parentInit != nil {
+		m.initFunc = func(ctx context.Context) error {
+			err := parentInit(ctx)
+			if err != nil {
+				return err
+			}
+			return initFunc(ctx)
+		}
+	} else {
+		m.initFunc = initFunc
+	}
+	return m
+}
+
 func (c *fsClient) Register(module string, wrapper *moduleWrapper) FSClient {
 	if c.err != nil {
 		return c
@@ -200,23 +216,29 @@ func Custom(init func(ctx context.Context) error, execute func(ctx context.Conte
 }
 
 type FunctionContext struct {
-	c *fsClient
+	c      *fsClient
+	name   string
+	module string
 }
 
 func (c *FunctionContext) GetState(ctx context.Context, key string) ([]byte, error) {
-	return c.c.rpc.GetState(ctx, key)
+	return c.c.rpc.GetState(c.warpContext(ctx), key)
 }
 
 func (c *FunctionContext) PutState(ctx context.Context, key string, value []byte) error {
-	return c.c.rpc.PutState(ctx, key, value)
+	return c.c.rpc.PutState(c.warpContext(ctx), key, value)
 }
 
 func (c *FunctionContext) Write(ctx context.Context, payload []byte) error {
-	return c.c.rpc.Write(ctx, payload)
+	return c.c.rpc.Write(c.warpContext(ctx), payload)
 }
 
 func (c *FunctionContext) Read(ctx context.Context) ([]byte, error) {
-	return c.c.rpc.Read(ctx)
+	return c.c.rpc.Read(c.warpContext(ctx))
+}
+
+func (c *FunctionContext) GetConfig(ctx context.Context) (map[string]string, error) {
+	return c.c.rpc.GetConfig(c.warpContext(ctx))
 }
 
 type funcCtxKey struct{}
@@ -241,15 +263,6 @@ func (c *fsClient) Run() error {
 	if funcName == "" {
 		return fmt.Errorf("%s is not set", FSFunctionName)
 	}
-	funcCtx := &FunctionContext{c: c}
-	if c.rpc == nil {
-		rpc, err := newFSRPCClient()
-		if err != nil {
-			return err
-		}
-		c.rpc = rpc
-	}
-	ctx := c.rpc.GetContext(context.WithValue(context.Background(), funcCtxKey{}, funcCtx), funcName)
 	module := os.Getenv(FSModuleName)
 	if module == "" {
 		module = DefaultModule
@@ -258,6 +271,15 @@ func (c *fsClient) Run() error {
 	if !ok {
 		return fmt.Errorf("module %s not found", module)
 	}
+	funcCtx := &FunctionContext{c: c, name: funcName, module: module}
+	if c.rpc == nil {
+		rpc, err := newFSRPCClient()
+		if err != nil {
+			return err
+		}
+		c.rpc = rpc
+	}
+	ctx := funcCtx.warpContext(context.WithValue(context.Background(), funcCtxKey{}, funcCtx))
 	m.fsClient = c
 	err := m.initFunc(ctx)
 	if err != nil {

clients/gofs/gofs_socket.go

@@ -25,12 +25,19 @@ import (
 	"os"
 	"strings"
 
+	"google.golang.org/grpc/metadata"
+
 	"github.com/functionstream/function-stream/fs/runtime/external/model"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/metadata"
 )
 
+func (c *FunctionContext) warpContext(parent context.Context) context.Context {
+	return metadata.NewOutgoingContext(parent, metadata.New(map[string]string{
+		"name": c.name,
+	}))
+}
+
 type fsRPCClient struct {
 	grpcCli model.FunctionClient
 }
@@ -65,13 +72,6 @@ func newFSRPCClient() (*fsRPCClient, error) {
 	return &fsRPCClient{grpcCli: client}, nil
 }
 
-func (c *fsRPCClient) GetContext(parent context.Context, funcName string) context.Context {
-	md := metadata.New(map[string]string{
-		"name": funcName,
-	})
-	return metadata.NewOutgoingContext(parent, md)
-}
-
 func (c *fsRPCClient) RegisterSchema(ctx context.Context, schema string) error {
 	_, err := c.grpcCli.RegisterSchema(ctx, &model.RegisterSchemaRequest{Schema: schema})
 	if err != nil {
@@ -124,6 +124,14 @@ func (c *fsRPCClient) ListStates(ctx context.Context, path string) ([]string, er
 	return res.Keys, nil
 }
 
+func (c *fsRPCClient) GetConfig(ctx context.Context) (map[string]string, error) {
+	res, err := c.grpcCli.GetConfig(ctx, &model.GetConfigRequest{})
+	if err != nil {
+		return nil, err
+	}
+	return res.Config, nil
+}
+
 func (c *fsRPCClient) loadModule(_ *moduleWrapper) {
 	// no-op
 }

clients/gofs/gofs_wasmfs.go

@@ -48,17 +48,17 @@ func process() {
 	}
 }
 
+func (c *FunctionContext) warpContext(parent context.Context) context.Context {
+	return parent
+}
+
 type fsRPCClient struct {
 }
 
 func newFSRPCClient() (*fsRPCClient, error) {
 	return &fsRPCClient{}, nil
 }
 
-func (c *fsRPCClient) GetContext(parent context.Context, funcName string) context.Context {
-	return context.Background()
-}
-
 func (c *fsRPCClient) RegisterSchema(ctx context.Context, schema string) error {
 	_, err := syscall.Write(registerSchemaFd, []byte(schema))
 	if err != nil {
@@ -68,19 +68,23 @@ func (c *fsRPCClient) RegisterSchema(ctx context.Context, schema string) error {
 }
 
 func (c *fsRPCClient) Write(ctx context.Context, payload []byte) error {
-	panic("rpc write not implemented")
+	panic("rpc write not supported")
 }
 
 func (c *fsRPCClient) Read(ctx context.Context) ([]byte, error) {
-	panic("rpc read not implemented")
+	panic("rpc read not supported")
 }
 
 func (c *fsRPCClient) GetState(ctx context.Context, key string) ([]byte, error) {
-	panic("rpc get state not implemented")
+	panic("rpc get state not supported")
 }
 
 func (c *fsRPCClient) PutState(ctx context.Context, key string, value []byte) error {
-	panic("rpc put state not implemented")
+	panic("rpc put state not supported")
+}
+
+func (c *fsRPCClient) GetConfig(ctx context.Context) (map[string]string, error) {
+	panic("rpc get config not supported")
 }
 
 func (c *fsRPCClient) loadModule(m *moduleWrapper) {

fs/api/func_ctx.go

@@ -28,4 +28,5 @@ type FunctionContext interface {
 	ListStates(ctx context.Context, startInclusive string, endExclusive string) ([]string, error)
 	DeleteState(ctx context.Context, key string) error
 	Write(record contube.Record) error
+	GetConfig() map[string]string
 }

fs/func_ctx_impl.go

@@ -19,6 +19,8 @@ package fs
 import (
 	"context"
 
+	"github.com/functionstream/function-stream/common/model"
+
 	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/pkg/errors"
@@ -28,12 +30,13 @@ var ErrStateStoreNotLoaded = errors.New("state store not loaded")
 
 type funcCtxImpl struct {
 	api.FunctionContext
+	function   *model.Function
 	stateStore api.StateStore
 	sink       chan<- contube.Record
 }
 
-func newFuncCtxImpl(store api.StateStore) *funcCtxImpl {
-	return &funcCtxImpl{stateStore: store}
+func newFuncCtxImpl(function *model.Function, store api.StateStore) *funcCtxImpl {
+	return &funcCtxImpl{function: function, stateStore: store}
 }
 
 func (f *funcCtxImpl) checkStateStore() error {
@@ -79,6 +82,10 @@ func (f *funcCtxImpl) Write(record contube.Record) error {
 	return nil
 }
 
+func (f *funcCtxImpl) GetConfig() map[string]string {
+	return f.function.Config
+}
+
 func (f *funcCtxImpl) setSink(sink chan<- contube.Record) {
 	f.sink = sink
 }

fs/func_ctx_impl_test.go

@@ -24,7 +24,7 @@ import (
 )
 
 func TestFuncCtx_NilStore(t *testing.T) {
-	f := newFuncCtxImpl(nil)
+	f := newFuncCtxImpl(nil, nil)
 	assert.ErrorIs(t, f.PutState(context.Background(), "key", []byte("value")), ErrStateStoreNotLoaded)
 	_, err := f.GetState(context.Background(), "key")
 	assert.ErrorIs(t, err, ErrStateStoreNotLoaded)

fs/manager.go

@@ -243,7 +243,7 @@ func (fm *functionManagerImpl) StartFunction(f *model.Function) error { // TODO:
 			return fmt.Errorf("failed to create state store: %w", err)
 		}
 
-		funcCtx := newFuncCtxImpl(store)
+		funcCtx := newFuncCtxImpl(f, store)
 		instanceLogger := fm.log.SubLogger("functionName", f.Name, "instanceIndex", int(i), "runtimeType", runtimeConfig.Type)
 		instance := fm.options.instanceFactory.NewFunctionInstance(f, funcCtx, i, instanceLogger)
 		fm.functionsLock.Lock()