From 971c5984b2511224a8864e26fa223d22cdaa5bee Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Wed, 15 May 2024 19:18:54 +0800 Subject: [PATCH 1/4] feat: support tls --- fs/contube/contube.go | 4 +++- server/config.go | 4 ++++ server/function_store.go | 4 ++++ server/server.go | 22 +++++++++++++++++++--- 4 files changed, 30 insertions(+), 4 deletions(-) diff --git a/fs/contube/contube.go b/fs/contube/contube.go index 8d221276..1ff7b91c 100644 --- a/fs/contube/contube.go +++ b/fs/contube/contube.go @@ -25,7 +25,9 @@ import ( ) var ( - ErrSinkTubeNotImplemented = errors.New("sink tube not implemented") + ErrTubeNotImplemented = errors.New("tube not implemented") + ErrSinkTubeNotImplemented = errors.Wrap(ErrTubeNotImplemented, "sink tube not implemented") + ErrSourceTubeNotImplemented = errors.Wrap(ErrTubeNotImplemented, "source tube not implemented") ) type Record interface { diff --git a/server/config.go b/server/config.go index cb1078c3..231ecab4 100644 --- a/server/config.go +++ b/server/config.go @@ -53,6 +53,10 @@ type Config struct { // FunctionStore is the path to the function store FunctionStore string `mapstructure:"function_store"` + + EnableTLS bool `mapstructure:"enable_tls"` + TLSCertFile string `mapstructure:"tls_cert_file"` + TLSKeyFile string `mapstructure:"tls_key_file"` } func init() { diff --git a/server/function_store.go b/server/function_store.go index acfa7fcb..ec9892ae 100644 --- a/server/function_store.go +++ b/server/function_store.go @@ -50,6 +50,10 @@ func (f *FunctionStoreImpl) Load() error { f.loadingFunctions = make(map[string]*model.Function) info, err := os.Stat(f.path) if err != nil { + if os.IsNotExist(err) { + slog.Info("the path to the function store does not exist. skip loading functions") + return nil + } return errors.Wrapf(err, "the path to the function store %s is invalid", f.path) } if !info.IsDir() { diff --git a/server/server.go b/server/server.go index 96bbb575..2638dbed 100644 --- a/server/server.go +++ b/server/server.go @@ -65,6 +65,9 @@ type serverOptions struct { runtimeLoader RuntimeLoaderType stateStoreLoader StateStoreLoaderType functionStore string + enableTls bool + tlsCertFile string + tlsKeyFile string } type ServerOption interface { @@ -205,6 +208,14 @@ func WithConfig(config *Config) ServerOption { return nil, err } o.httpListener = ln + o.enableTls = config.EnableTLS + if o.enableTls { + if config.TLSCertFile == "" || config.TLSKeyFile == "" { + return nil, errors.New("TLS certificate and key file must be provided") + } + o.tlsCertFile = config.TLSCertFile + o.tlsKeyFile = config.TLSKeyFile + } err = initFactories[contube.TubeFactory](config.TubeFactory, o.tubeLoader, func(n string, f contube.TubeFactory) { o.managerOpts = append(o.managerOpts, fs.WithTubeFactory(n, f)) }) @@ -326,7 +337,8 @@ func (s *Server) Run(context context.Context) { func (s *Server) startRESTHandlers() error { statusSvr := new(restful.WebService) - statusSvr.Route(statusSvr.GET("/api/v1/status").To(func(request *restful.Request, response *restful.Response) { + statusSvr.Path("/api/v1/status") + statusSvr.Route(statusSvr.GET("/").To(func(request *restful.Request, response *restful.Response) { response.WriteHeader(http.StatusOK) }). Doc("Get the status of the Function Stream"). @@ -360,7 +372,11 @@ func (s *Server) startRESTHandlers() error { } s.httpSvr.Store(httpSvr) - return httpSvr.Serve(s.options.httpListener) + if s.options.enableTls { + return httpSvr.ServeTLS(s.options.httpListener, s.options.tlsCertFile, s.options.tlsKeyFile) + } else { + return httpSvr.Serve(s.options.httpListener) + } } func enrichSwaggerObject(swo *spec.Swagger) { @@ -432,6 +448,7 @@ func (s *Server) WaitForReady(ctx context.Context) <-chan struct{} { if err != nil { s.log.InfoContext(ctx, "Detect connection to server failed", slog.Any("error", err)) } + s.log.Info("Server is ready", slog.String("address", s.options.httpListener.Addr().String())) return true } go func() { @@ -447,7 +464,6 @@ func (s *Server) WaitForReady(ctx context.Context) <-chan struct{} { return case <-time.After(1 * time.Second): if detect() { - s.log.Info("Server is ready", slog.String("address", s.options.httpListener.Addr().String())) return } } From ea3ca2798b163fa98542262019e5679c0a63dce1 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Sat, 18 May 2024 19:22:34 +0800 Subject: [PATCH 2/4] feat: improve config --- conf/function-stream.yaml | 6 ++---- server/config.go | 44 +++++++++++++++++++++++---------------- server/config_test.go | 30 ++++++-------------------- server/server.go | 39 ++++++++++++++-------------------- tests/test_config.json | 18 +++++++++------- tests/test_config.yaml | 12 +++++------ 6 files changed, 67 insertions(+), 82 deletions(-) diff --git a/conf/function-stream.yaml b/conf/function-stream.yaml index 001c50dc..e759e9b6 100644 --- a/conf/function-stream.yaml +++ b/conf/function-stream.yaml @@ -15,15 +15,13 @@ listen_addr: ":7300" tube_factory: pulsar: - type: "pulsar" config: pulsar_url: "pulsar://localhost:6650" memory: - Type: "memory" default: ref: "pulsar" runtime_factory: wasm: - type: "wasm" default: - ref: "wasm" \ No newline at end of file + ref: "wasm" +function_store: "./functions" \ No newline at end of file diff --git a/server/config.go b/server/config.go index 231ecab4..176bb123 100644 --- a/server/config.go +++ b/server/config.go @@ -26,9 +26,9 @@ import ( "github.com/spf13/viper" ) -type FactoryConfig struct { +type TypeConfig struct { + Name string `mapstructure:"name"` Ref *string `mapstructure:"ref"` - Type *string `mapstructure:"type"` Config *common.ConfigMap `mapstructure:"config"` } @@ -41,11 +41,13 @@ type Config struct { // ListenAddr is the address that the function stream REST service will listen on. ListenAddr string `mapstructure:"listen_addr"` - // TubeFactory is the list of tube factories that the function stream server will use. - TubeFactory map[string]*FactoryConfig `mapstructure:"tube_factory"` + TubeTypes []*TypeConfig `mapstructure:"tube_types"` + + tubeTypesMap map[string]*TypeConfig + + RuntimeTypes []*TypeConfig `mapstructure:"runtime_types"` - // RuntimeFactory is the list of runtime factories that the function stream server will use. - RuntimeFactory map[string]*FactoryConfig `mapstructure:"runtime_factory"` + runtimeTypesMap map[string]*TypeConfig // StateStore is the configuration for the state store that the function stream server will use. // Optional @@ -64,25 +66,16 @@ func init() { viper.SetDefault("function_store", "./functions") } -func preprocessFactoriesConfig(n string, m map[string]*FactoryConfig) error { +func preprocessFactoriesConfig(n string, m map[string]*TypeConfig) error { for name, factory := range m { if ref := factory.Ref; ref != nil && *ref != "" { referred, ok := m[strings.ToLower(*ref)] if !ok { return errors.Errorf("%s factory %s refers to non-existent factory %s", n, name, *ref) } - if factory.Type == nil { - factory.Type = referred.Type - } factory.Config = common.MergeConfig(referred.Config, factory.Config) } } - - for name, factory := range m { - if factory.Type == nil { - return errors.Errorf("%s factory %s has no type", n, name) - } - } return nil } @@ -90,11 +83,11 @@ func (c *Config) preprocessConfig() error { if c.ListenAddr == "" { return errors.New("ListenAddr shouldn't be empty") } - err := preprocessFactoriesConfig("Tube", c.TubeFactory) + err := preprocessFactoriesConfig("Tube", c.tubeTypesMap) if err != nil { return err } - return preprocessFactoriesConfig("Runtime", c.RuntimeFactory) + return preprocessFactoriesConfig("Runtime", c.runtimeTypesMap) } func loadConfig() (*Config, error) { @@ -102,6 +95,20 @@ func loadConfig() (*Config, error) { if err := viper.Unmarshal(&c); err != nil { return nil, err } + c.tubeTypesMap = make(map[string]*TypeConfig) + for _, t := range c.TubeTypes { + if t.Config == nil { + t.Config = &common.ConfigMap{} + } + c.tubeTypesMap[strings.ToLower(t.Name)] = t + } + c.runtimeTypesMap = make(map[string]*TypeConfig) + for _, t := range c.RuntimeTypes { + if t.Config == nil { + t.Config = &common.ConfigMap{} + } + c.runtimeTypesMap[strings.ToLower(t.Name)] = t + } if err := c.preprocessConfig(); err != nil { return nil, err } @@ -118,6 +125,7 @@ func LoadConfigFromFile(filePath string) (*Config, error) { return loadConfig() } +// Deprecate func LoadConfigFromEnv() (*Config, error) { for _, env := range os.Environ() { if strings.HasPrefix(env, "FS_") { diff --git a/server/config_test.go b/server/config_test.go index f735cb56..f0a14e9e 100644 --- a/server/config_test.go +++ b/server/config_test.go @@ -17,10 +17,8 @@ package server import ( - "os" "testing" - "github.com/spf13/viper" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -37,37 +35,21 @@ func TestLoadConfigFromJson(t *testing.T) { assertConfig(t, c) } -func TestLoadConfigFromEnv(t *testing.T) { - assert.Nil(t, os.Setenv("FS_LISTEN_ADDR", ":17300")) - assert.Nil(t, os.Setenv("FS_TUBE_FACTORY__MY_PULSAR__TYPE", "pulsar")) - assert.Nil(t, os.Setenv("FS_TUBE_FACTORY__MY_PULSAR__CONFIG__PULSAR_URL", "pulsar://localhost:6651")) - assert.Nil(t, os.Setenv("FS_TUBE_FACTORY__MY_MEMORY__TYPE", "memory")) - assert.Nil(t, os.Setenv("FS_TUBE_FACTORY__DEFAULT__REF", "my_pulsar")) - - viper.AutomaticEnv() - - c, err := LoadConfigFromEnv() - require.Nil(t, err) - assertConfig(t, c) -} - func assertConfig(t *testing.T, c *Config) { assert.Equal(t, ":17300", c.ListenAddr) - require.Contains(t, c.TubeFactory, "my_pulsar") - assert.Equal(t, "pulsar", *c.TubeFactory["my_pulsar"].Type) + require.Contains(t, c.tubeTypesMap, "my_pulsar") - if config := c.TubeFactory["my_pulsar"].Config; config != nil { + if config := c.tubeTypesMap["my_pulsar"].Config; config != nil { assert.Equal(t, "pulsar://localhost:6651", (*config)["pulsar_url"]) } else { t.Fatal("pulsar config is nil") } - require.Contains(t, c.TubeFactory, "my_memory") - assert.Equal(t, "memory", *c.TubeFactory["my_memory"].Type) + require.Contains(t, c.tubeTypesMap, "my_memory") - require.Contains(t, c.TubeFactory, "default") - assert.Equal(t, "my_pulsar", *c.TubeFactory["default"].Ref) - if config := c.TubeFactory["default"].Config; config != nil { + require.Contains(t, c.tubeTypesMap, "default") + assert.Equal(t, "my_pulsar", *c.tubeTypesMap["default"].Ref) + if config := c.tubeTypesMap["default"].Config; config != nil { assert.Equal(t, "pulsar://localhost:6651", (*config)["pulsar_url"]) } else { t.Fatal("pulsar config is nil") diff --git a/server/server.go b/server/server.go index 2638dbed..7b82bb33 100644 --- a/server/server.go +++ b/server/server.go @@ -53,8 +53,8 @@ type Server struct { FunctionStore FunctionStore } -type TubeLoaderType func(c *FactoryConfig) (contube.TubeFactory, error) -type RuntimeLoaderType func(c *FactoryConfig) (api.FunctionRuntimeFactory, error) +type TubeLoaderType func(name string, c *TypeConfig) (contube.TubeFactory, error) +type RuntimeLoaderType func(name string, c *TypeConfig) (api.FunctionRuntimeFactory, error) type StateStoreLoaderType func(c *StateStoreConfig) (api.StateStore, error) type serverOptions struct { @@ -131,7 +131,7 @@ func WithStateStoreLoader(loader func(c *StateStoreConfig) (api.StateStore, erro }) } -func getRefFactory(m map[string]*FactoryConfig, name string, visited set.Set[string]) (string, error) { +func getRefFactory(m map[string]*TypeConfig, name string, visited set.Set[string]) (string, error) { if visited.Has(name) { return "", errors.Errorf("circular reference of factory %s", name) } @@ -146,7 +146,7 @@ func getRefFactory(m map[string]*FactoryConfig, name string, visited set.Set[str return name, nil } -func initFactories[T any](m map[string]*FactoryConfig, newFactory func(c *FactoryConfig) (T, error), +func initFactories[T any](m map[string]*TypeConfig, newFactory func(name string, c *TypeConfig) (T, error), setup func(n string, f T)) error { factoryMap := make(map[string]T) @@ -160,10 +160,7 @@ func initFactories[T any](m map[string]*FactoryConfig, newFactory func(c *Factor if !exist { return errors.Errorf("factory %s not found, which the factory %s is pointed to", refName, name) } - if fc.Type == nil { - return errors.Errorf("factory %s type is not set", refName) - } - f, err := newFactory(fc) + f, err := newFactory(refName, fc) if err != nil { return err } @@ -175,22 +172,22 @@ func initFactories[T any](m map[string]*FactoryConfig, newFactory func(c *Factor return nil } -func DefaultTubeLoader(c *FactoryConfig) (contube.TubeFactory, error) { - switch strings.ToLower(*c.Type) { +func DefaultTubeLoader(name string, c *TypeConfig) (contube.TubeFactory, error) { + switch strings.ToLower(name) { case common.PulsarTubeType: return contube.NewPulsarEventQueueFactory(context.Background(), contube.ConfigMap(*c.Config)) case common.MemoryTubeType: return contube.NewMemoryQueueFactory(context.Background()), nil } - return nil, errors.WithMessagef(ErrUnsupportedTubeType, "unsupported tube type :%s", *c.Type) + return nil, errors.WithMessagef(ErrUnsupportedTubeType, "unsupported tube type :%s", name) } -func DefaultRuntimeLoader(c *FactoryConfig) (api.FunctionRuntimeFactory, error) { - switch strings.ToLower(*c.Type) { +func DefaultRuntimeLoader(name string, c *TypeConfig) (api.FunctionRuntimeFactory, error) { + switch strings.ToLower(name) { case common.WASMRuntime: return wazero.NewWazeroFunctionRuntimeFactory(), nil } - return nil, errors.WithMessagef(ErrUnsupportedTRuntimeType, "unsupported runtime type: %s", *c.Type) + return nil, errors.WithMessagef(ErrUnsupportedTRuntimeType, "unsupported runtime type: %s", name) } func DefaultStateStoreLoader(c *StateStoreConfig) (api.StateStore, error) { @@ -216,13 +213,13 @@ func WithConfig(config *Config) ServerOption { o.tlsCertFile = config.TLSCertFile o.tlsKeyFile = config.TLSKeyFile } - err = initFactories[contube.TubeFactory](config.TubeFactory, o.tubeLoader, func(n string, f contube.TubeFactory) { + err = initFactories[contube.TubeFactory](config.tubeTypesMap, o.tubeLoader, func(n string, f contube.TubeFactory) { o.managerOpts = append(o.managerOpts, fs.WithTubeFactory(n, f)) }) if err != nil { return nil, err } - err = initFactories[api.FunctionRuntimeFactory](config.RuntimeFactory, o.runtimeLoader, + err = initFactories[api.FunctionRuntimeFactory](config.runtimeTypesMap, o.runtimeLoader, func(n string, f api.FunctionRuntimeFactory) { o.managerOpts = append(o.managerOpts, fs.WithRuntimeFactory(n, f)) }) @@ -245,7 +242,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { options := &serverOptions{} httpTubeFact := contube.NewHttpTubeFactory(context.Background()) options.managerOpts = []fs.ManagerOption{ - fs.WithDefaultTubeFactory(contube.NewMemoryQueueFactory(context.Background())), fs.WithTubeFactory("http", httpTubeFact), } options.httpTubeFact = httpTubeFact @@ -295,9 +291,8 @@ func NewServer(opts ...ServerOption) (*Server, error) { func NewDefaultServer() (*Server, error) { defaultConfig := &Config{ ListenAddr: ":7300", - TubeFactory: map[string]*FactoryConfig{ + tubeTypesMap: map[string]*TypeConfig{ "pulsar": { - Type: common.OptionalStr(common.PulsarTubeType), Config: &common.ConfigMap{ contube.PulsarURLKey: "pulsar://localhost:6650", }, @@ -306,10 +301,8 @@ func NewDefaultServer() (*Server, error) { Ref: common.OptionalStr("pulsar"), }, }, - RuntimeFactory: map[string]*FactoryConfig{ - "wasm": { - Type: common.OptionalStr(common.WASMRuntime), - }, + runtimeTypesMap: map[string]*TypeConfig{ + "wasm": {}, "default": { Ref: common.OptionalStr("wasm"), }, diff --git a/tests/test_config.json b/tests/test_config.json index 8c89ab5d..c3b96f5a 100644 --- a/tests/test_config.json +++ b/tests/test_config.json @@ -1,17 +1,21 @@ { "listen_addr": ":17300", - "tube_factory": { - "my_pulsar": { - "type": "pulsar", + "tube_types": [ + { + "name": "my_pulsar", "config": { "pulsar_url": "pulsar://localhost:6651" } }, - "my_memory": { - "type": "memory" + { + "name": "my_memory", + "config": { + "test": 1 + } }, - "default": { + { + "name": "default", "ref": "my_pulsar" } - } + ] } \ No newline at end of file diff --git a/tests/test_config.yaml b/tests/test_config.yaml index a9a0dd12..2c55dc42 100644 --- a/tests/test_config.yaml +++ b/tests/test_config.yaml @@ -13,12 +13,12 @@ # limitations under the License. listen_addr: ":17300" -tube_factory: - my_pulsar: - type: "pulsar" +tube_types: + - name: my_pulsar config: pulsar_url: "pulsar://localhost:6651" - my_memory: - type: "memory" - default: + - name: my_memory + config: + test: 1 + - name: default ref: "my_pulsar" \ No newline at end of file From 171cd5ca6e06fc031afc9bbb7840d432a83fb1ec Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Tue, 21 May 2024 22:30:32 +0800 Subject: [PATCH 3/4] feat: improve config --- conf/function-stream.yaml | 4 ++-- conf/standalone.yaml | 6 ++---- server/config.go | 35 ++++++++++++++++--------------- server/server.go | 43 +++++++++++++++++++++++---------------- 4 files changed, 46 insertions(+), 42 deletions(-) diff --git a/conf/function-stream.yaml b/conf/function-stream.yaml index e759e9b6..245d8da1 100644 --- a/conf/function-stream.yaml +++ b/conf/function-stream.yaml @@ -13,14 +13,14 @@ # limitations under the License. listen_addr: ":7300" -tube_factory: +tube_types: pulsar: config: pulsar_url: "pulsar://localhost:6650" memory: default: ref: "pulsar" -runtime_factory: +runtime_types: wasm: default: ref: "wasm" diff --git a/conf/standalone.yaml b/conf/standalone.yaml index 24bfcb85..3c22849f 100644 --- a/conf/standalone.yaml +++ b/conf/standalone.yaml @@ -13,13 +13,11 @@ # limitations under the License. listen_addr: ":7300" -tube_factory: +tube_types: memory: - Type: "memory" default: ref: "memory" -runtime_factory: +runtime_types: wasm: - type: "wasm" default: ref: "wasm" \ No newline at end of file diff --git a/server/config.go b/server/config.go index 176bb123..01ba9c58 100644 --- a/server/config.go +++ b/server/config.go @@ -38,17 +38,16 @@ type StateStoreConfig struct { } type Config struct { + tubeTypesMap map[string]*TypeConfig + runtimeTypesMap map[string]*TypeConfig + // ListenAddr is the address that the function stream REST service will listen on. ListenAddr string `mapstructure:"listen_addr"` TubeTypes []*TypeConfig `mapstructure:"tube_types"` - tubeTypesMap map[string]*TypeConfig - RuntimeTypes []*TypeConfig `mapstructure:"runtime_types"` - runtimeTypesMap map[string]*TypeConfig - // StateStore is the configuration for the state store that the function stream server will use. // Optional StateStore *StateStoreConfig `mapstructure:"state_store"` @@ -80,6 +79,20 @@ func preprocessFactoriesConfig(n string, m map[string]*TypeConfig) error { } func (c *Config) preprocessConfig() error { + c.tubeTypesMap = make(map[string]*TypeConfig) + for _, t := range c.TubeTypes { + if t.Config == nil { + t.Config = &common.ConfigMap{} + } + c.tubeTypesMap[strings.ToLower(t.Name)] = t + } + c.runtimeTypesMap = make(map[string]*TypeConfig) + for _, t := range c.RuntimeTypes { + if t.Config == nil { + t.Config = &common.ConfigMap{} + } + c.runtimeTypesMap[strings.ToLower(t.Name)] = t + } if c.ListenAddr == "" { return errors.New("ListenAddr shouldn't be empty") } @@ -95,20 +108,6 @@ func loadConfig() (*Config, error) { if err := viper.Unmarshal(&c); err != nil { return nil, err } - c.tubeTypesMap = make(map[string]*TypeConfig) - for _, t := range c.TubeTypes { - if t.Config == nil { - t.Config = &common.ConfigMap{} - } - c.tubeTypesMap[strings.ToLower(t.Name)] = t - } - c.runtimeTypesMap = make(map[string]*TypeConfig) - for _, t := range c.RuntimeTypes { - if t.Config == nil { - t.Config = &common.ConfigMap{} - } - c.runtimeTypesMap[strings.ToLower(t.Name)] = t - } if err := c.preprocessConfig(); err != nil { return nil, err } diff --git a/server/server.go b/server/server.go index 7b82bb33..24f6d514 100644 --- a/server/server.go +++ b/server/server.go @@ -53,8 +53,8 @@ type Server struct { FunctionStore FunctionStore } -type TubeLoaderType func(name string, c *TypeConfig) (contube.TubeFactory, error) -type RuntimeLoaderType func(name string, c *TypeConfig) (api.FunctionRuntimeFactory, error) +type TubeLoaderType func(c *TypeConfig) (contube.TubeFactory, error) +type RuntimeLoaderType func(c *TypeConfig) (api.FunctionRuntimeFactory, error) type StateStoreLoaderType func(c *StateStoreConfig) (api.StateStore, error) type serverOptions struct { @@ -146,7 +146,7 @@ func getRefFactory(m map[string]*TypeConfig, name string, visited set.Set[string return name, nil } -func initFactories[T any](m map[string]*TypeConfig, newFactory func(name string, c *TypeConfig) (T, error), +func initFactories[T any](m map[string]*TypeConfig, newFactory func(c *TypeConfig) (T, error), setup func(n string, f T)) error { factoryMap := make(map[string]T) @@ -160,7 +160,7 @@ func initFactories[T any](m map[string]*TypeConfig, newFactory func(name string, if !exist { return errors.Errorf("factory %s not found, which the factory %s is pointed to", refName, name) } - f, err := newFactory(refName, fc) + f, err := newFactory(fc) if err != nil { return err } @@ -172,22 +172,24 @@ func initFactories[T any](m map[string]*TypeConfig, newFactory func(name string, return nil } -func DefaultTubeLoader(name string, c *TypeConfig) (contube.TubeFactory, error) { - switch strings.ToLower(name) { +func DefaultTubeLoader(c *TypeConfig) (contube.TubeFactory, error) { + switch strings.ToLower(c.Name) { case common.PulsarTubeType: return contube.NewPulsarEventQueueFactory(context.Background(), contube.ConfigMap(*c.Config)) case common.MemoryTubeType: return contube.NewMemoryQueueFactory(context.Background()), nil + case common.HttpTubeType: + return contube.NewHttpTubeFactory(context.Background()), nil } - return nil, errors.WithMessagef(ErrUnsupportedTubeType, "unsupported tube type :%s", name) + return nil, errors.WithMessagef(ErrUnsupportedTubeType, "unsupported tube type :%s", c.Name) } -func DefaultRuntimeLoader(name string, c *TypeConfig) (api.FunctionRuntimeFactory, error) { - switch strings.ToLower(name) { +func DefaultRuntimeLoader(c *TypeConfig) (api.FunctionRuntimeFactory, error) { + switch strings.ToLower(c.Name) { case common.WASMRuntime: return wazero.NewWazeroFunctionRuntimeFactory(), nil } - return nil, errors.WithMessagef(ErrUnsupportedTRuntimeType, "unsupported runtime type: %s", name) + return nil, errors.WithMessagef(ErrUnsupportedRuntimeType, "unsupported runtime type: %s", c.Name) } func DefaultStateStoreLoader(c *StateStoreConfig) (api.StateStore, error) { @@ -291,20 +293,25 @@ func NewServer(opts ...ServerOption) (*Server, error) { func NewDefaultServer() (*Server, error) { defaultConfig := &Config{ ListenAddr: ":7300", - tubeTypesMap: map[string]*TypeConfig{ - "pulsar": { + TubeTypes: []*TypeConfig{ + { + Name: "pulsar", Config: &common.ConfigMap{ contube.PulsarURLKey: "pulsar://localhost:6650", }, }, - "default": { - Ref: common.OptionalStr("pulsar"), + { + Name: "default", + Ref: common.OptionalStr("pulsar"), }, }, - runtimeTypesMap: map[string]*TypeConfig{ - "wasm": {}, - "default": { - Ref: common.OptionalStr("wasm"), + RuntimeTypes: []*TypeConfig{ + { + Name: "wasm", + }, + { + Name: "default", + Ref: common.OptionalStr("wasm"), }, }, } From 8f88a7396bfdcc10b0aa028694545c4474eb1b85 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Tue, 21 May 2024 22:31:30 +0800 Subject: [PATCH 4/4] feat: validate config --- server/server.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/server/server.go b/server/server.go index 24f6d514..bff2d5b9 100644 --- a/server/server.go +++ b/server/server.go @@ -202,6 +202,10 @@ func DefaultStateStoreLoader(c *StateStoreConfig) (api.StateStore, error) { func WithConfig(config *Config) ServerOption { return serverOptionFunc(func(o *serverOptions) (*serverOptions, error) { + err := config.preprocessConfig() + if err != nil { + return nil, err + } ln, err := net.Listen("tcp", config.ListenAddr) if err != nil { return nil, err