refactor: yomo run zipper from a config file (#496)

# Description

1. Support Zipper launch from a config file.

2. Remove the `-a -m` flag from `yomo serve`, as all configurations are
now covered by the configuration file.

3. If you want to develop a Zipper, use `yomo.NewZipper()`. If you want
to run a Zipper from a configuration, use `yomo.RunZipper()`.

Author: woorui

cli/README.md

@@ -74,7 +74,7 @@ cd example/stream-fn-db && yomo build && ./sl.yomo
 
 ### 4. YoMo-Zipper
 
-#### Configure YoMo-Zipper `workflow.yaml`
+#### Configure YoMo-Zipper `config.yaml`
 
 ```yaml
 name: Service
@@ -88,7 +88,7 @@ functions:
 #### Run
 
 ```sh
-yomo serve --config workflow.yaml
+yomo serve --config config.yaml
 ```
 
 ## Example

cli/README_CN.md

@@ -63,7 +63,7 @@ cd example/stream-fn-db && yomo build && ./sl.yomo
 ```
 
 ### 4. YoMo-Zipper 应用编排
-#### 编写工作流配置文件 `workflow.yaml`
+#### 编写工作流配置文件 `config.yaml`
 
 ```yaml
 name: Service
@@ -76,7 +76,7 @@ functions:
 #### 运行 YoMo-Zipper 应用程序
 
 ```shell
-yomo serve --config workflow.yaml
+yomo serve --config config.yaml
 ```
 
 ## 示例

cli/run.go

@@ -96,7 +96,7 @@ func init() {
 	rootCmd.AddCommand(runCmd)
 
 	runCmd.Flags().StringVarP(&url, "zipper", "z", "localhost:9000", "YoMo-Zipper endpoint addr")
-	runCmd.Flags().StringVarP(&opts.Name, "name", "n", "", "yomo stream function name. It should match the specific service name in YoMo-Zipper config (workflow.yaml)")
+	runCmd.Flags().StringVarP(&opts.Name, "name", "n", "", "yomo stream function name. It should match the specific service name in YoMo-Zipper config (config.yaml)")
 	runCmd.Flags().StringVarP(&opts.ModFile, "modfile", "m", "", "custom go.mod")
 	runCmd.Flags().StringVarP(&opts.Credential, "credential", "d", "", "client credential payload, eg: `token:dBbBiRE7`")
 	runCmd.Flags().StringVarP(&opts.Runtime, "runtime", "r", "", "serverless runtime type")

cli/serve.go

@@ -16,59 +16,28 @@ limitations under the License.
 package cli
 
 import (
+	"context"
 	"os"
-	"strings"
 
 	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
 	"github.com/yomorun/yomo"
 	"github.com/yomorun/yomo/pkg/log"
 )
 
-var meshConfURL string
-var v *viper.Viper
-
 // serveCmd represents the serve command
 var serveCmd = &cobra.Command{
 	Use:   "serve",
 	Short: "Run a YoMo-Zipper",
 	Long:  "Run a YoMo-Zipper",
 	Run: func(cmd *cobra.Command, args []string) {
 		if config == "" {
-			log.FailureStatusEvent(os.Stdout, "Please input the file name of workflow config")
-			return
-		}
-		// printYoMoServerConf(conf)
-
-		// endpoint := fmt.Sprintf("%s:%d", conf.Host, conf.Port)
-
-		zipper, err := yomo.NewZipper(config)
-		if err != nil {
-			log.FailureStatusEvent(os.Stdout, err.Error())
-			return
-		}
-		// auth
-		auth := v.GetString("auth")
-		if len(auth) > 0 {
-			idx := strings.Index(auth, ":")
-			if idx != -1 {
-				authName := auth[:idx]
-				idx++
-				args := auth[idx:]
-				authArgs := strings.Split(args, ",")
-				// log.InfoStatusEvent(os.Stdout, "authName=%s, authArgs=%s, idx=%d", authName, authArgs, idx)
-				zipper.InitOptions(yomo.WithAuth(authName, authArgs...))
-			}
-		}
-		// mesh
-		err = zipper.ConfigMesh(meshConfURL)
-		if err != nil {
-			log.FailureStatusEvent(os.Stdout, err.Error())
+			log.FailureStatusEvent(os.Stdout, "Please input the file name of config")
 			return
 		}
 
 		log.InfoStatusEvent(os.Stdout, "Running YoMo-Zipper...")
-		err = zipper.ListenAndServe()
+
+		err := yomo.RunZipper(context.Background(), config)
 		if err != nil {
 			log.FailureStatusEvent(os.Stdout, err.Error())
 			return
@@ -79,13 +48,5 @@ var serveCmd = &cobra.Command{
 func init() {
 	rootCmd.AddCommand(serveCmd)
 
-	serveCmd.Flags().StringVarP(&config, "config", "c", "", "Workflow config file")
-	serveCmd.Flags().StringVarP(&meshConfURL, "mesh-config", "m", "", "The URL of mesh config")
-	// auth string
-	serveCmd.Flags().StringP("auth", "a", "", "authentication name and arguments, eg: `token:yomo`")
-	v = viper.New()
-	v.AutomaticEnv()
-	v.SetEnvPrefix("YOMO")
-	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
-	v.BindPFlag("auth", serveCmd.Flags().Lookup("auth"))
+	serveCmd.Flags().StringVarP(&config, "config", "c", "", "config file")
 }

core/client.go

@@ -320,7 +320,16 @@ func (c *Client) SetErrorHandler(fn func(err error)) {
 	c.logger.Debug("the error handler has been set")
 }
 
-// ClientID return the client ID
-func (c *Client) ClientID() string {
-	return c.clientID
+// ClientID returns the ID of client.
+func (c *Client) ClientID() string { return c.clientID }
+
+// Name returns the name of client.
+func (c *Client) Name() string { return c.name }
+
+// FrameWriterConnection represents a frame writer that can connect to an addr.
+type FrameWriterConnection interface {
+	frame.Writer
+	Name() string
+	Close() error
+	Connect(context.Context, string) error
 }

core/client_test.go

@@ -42,22 +42,21 @@ func TestFrameRoundTrip(t *testing.T) {
 	)
 
 	server := NewServer("zipper",
-		WithAddr(testaddr),
 		WithAuth("token", "auth-token"),
 		WithServerQuicConfig(DefalutQuicConfig),
 		WithServerTLSConfig(nil),
 		WithServerLogger(discardingLogger),
 	)
 	server.ConfigMetadataDecoder(metadata.DefaultDecoder())
-	server.ConfigRouter(router.Default([]config.App{{Name: "sfn-1"}, {Name: "close-early-sfn"}}))
+	server.ConfigRouter(router.Default([]config.Function{{Name: "sfn-1"}, {Name: "close-early-sfn"}}))
 
 	// test server hooks
 	ht := &hookTester{t}
 	server.SetStartHandlers(ht.startHandler)
 	server.SetBeforeHandlers(ht.beforeHandler)
 	server.SetAfterHandlers(ht.afterHandler)
 
-	recorder := newFrameWriterRecorder()
+	recorder := newFrameWriterRecorder("mockClient")
 	server.AddDownstreamServer("mockAddr", recorder)
 
 	go func() {
@@ -189,14 +188,19 @@ func createTestStreamFunction(name string, obversedTag frame.Tag) *Client {
 
 // frameWriterRecorder frames be writen.
 type frameWriterRecorder struct {
-	mu  sync.Mutex
-	buf *bytes.Buffer
+	name string
+	mu   sync.Mutex
+	buf  *bytes.Buffer
 }
 
-func newFrameWriterRecorder() *frameWriterRecorder {
-	return &frameWriterRecorder{buf: bytes.NewBuffer([]byte{})}
+func newFrameWriterRecorder(name string) *frameWriterRecorder {
+	return &frameWriterRecorder{name: name, buf: bytes.NewBuffer([]byte{})}
 }
 
+func (w *frameWriterRecorder) Name() string                              { return w.name }
+func (w *frameWriterRecorder) Close() error                              { return nil }
+func (w *frameWriterRecorder) Connect(_ context.Context, _ string) error { return nil }
+
 func (w *frameWriterRecorder) WriteFrame(frm frame.Frame) error {
 	w.mu.Lock()
 	defer w.mu.Unlock()

core/router/default.go

@@ -18,7 +18,7 @@ type DefaultRouter struct {
 }
 
 // Default return the DefaultRouter.
-func Default(functions []config.App) Router {
+func Default(functions []config.Function) Router {
 	return &DefaultRouter{r: newRoute(functions)}
 }
 
@@ -38,12 +38,12 @@ func (r *DefaultRouter) Clean() {
 }
 
 type defaultRoute struct {
-	functions []config.App
+	functions []config.Function
 	data      map[frame.Tag]map[string]string
 	mu        sync.RWMutex
 }
 
-func newRoute(functions []config.App) *defaultRoute {
+func newRoute(functions []config.Function) *defaultRoute {
 	return &defaultRoute{
 		functions: functions,
 		data:      make(map[frame.Tag]map[string]string),

core/router/default_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestRouter(t *testing.T) {
-	router := Default([]config.App{{Name: "sfn-1"}})
+	router := Default([]config.Function{{Name: "sfn-1"}})
 
 	m := &metadata.Default{}
 

core/server.go

@@ -35,7 +35,7 @@ type Server struct {
 	router                  router.Router
 	metadataDecoder         metadata.Decoder
 	counterOfDataFrame      int64
-	downstreams             map[string]frame.Writer
+	downstreams             map[string]FrameWriterConnection
 	mu                      sync.Mutex
 	opts                    *serverOptions
 	startHandlers           []FrameHandler
@@ -58,7 +58,7 @@ func NewServer(name string, opts ...ServerOption) *Server {
 
 	s := &Server{
 		name:        name,
-		downstreams: make(map[string]frame.Writer),
+		downstreams: make(map[string]FrameWriterConnection),
 		logger:      logger,
 		opts:        options,
 	}
@@ -68,10 +68,6 @@ func NewServer(name string, opts ...ServerOption) *Server {
 
 // ListenAndServe starts the server.
 func (s *Server) ListenAndServe(ctx context.Context, addr string) error {
-	if addr == "" {
-		addr = DefaultListenAddr
-	}
-
 	udpAddr, err := net.ResolveUDPAddr("udp", addr)
 	if err != nil {
 		return err
@@ -81,6 +77,13 @@ func (s *Server) ListenAndServe(ctx context.Context, addr string) error {
 		return err
 	}
 
+	s.logger = s.logger.With("zipper_addr", addr)
+
+	// connect to all downstreams.
+	for addr, client := range s.downstreams {
+		go client.Connect(ctx, addr)
+	}
+
 	return s.Serve(ctx, conn)
 }
 
@@ -104,13 +107,7 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error {
 	}
 	s.listener = listener
 
-	s.logger.Info(
-		"zipper is up and running",
-		"pid", os.Getpid(),
-		"quic", listener.Versions(),
-		"auth_name", s.authNames(),
-		"zipper_addr", s.opts.addr,
-	)
+	s.logger.Info("zipper is up and running", "pid", os.Getpid(), "quic", listener.Versions(), "auth_name", s.authNames())
 
 	for {
 		conn, err := s.listener.Accept(ctx)
@@ -175,6 +172,10 @@ func (s *Server) Logger() *slog.Logger { return s.logger }
 
 // Close will shutdown the server.
 func (s *Server) Close() error {
+	// downstreams
+	for _, ds := range s.downstreams {
+		ds.Close()
+	}
 	// connector
 	if s.connector != nil {
 		s.connector.Close()
@@ -379,12 +380,19 @@ func (s *Server) StatsFunctions() map[string]string {
 
 // StatsCounter returns how many DataFrames pass through server.
 func (s *Server) StatsCounter() int64 {
-	return s.counterOfDataFrame
+	return atomic.LoadInt64(&s.counterOfDataFrame)
 }
 
 // Downstreams return all the downstream servers.
-func (s *Server) Downstreams() map[string]frame.Writer {
-	return s.downstreams
+func (s *Server) Downstreams() map[string]string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	snapshotOfDownstream := make(map[string]string, len(s.downstreams))
+	for addr, client := range s.downstreams {
+		snapshotOfDownstream[addr] = client.Name()
+	}
+	return snapshotOfDownstream
 }
 
 // ConfigRouter is used to set router by zipper
@@ -413,7 +421,7 @@ func (s *Server) ConfigAlpnHandler(h func(string) error) {
 
 // AddDownstreamServer add a downstream server to this server. all the DataFrames will be
 // dispatch to all the downstreams.
-func (s *Server) AddDownstreamServer(addr string, c frame.Writer) {
+func (s *Server) AddDownstreamServer(addr string, c FrameWriterConnection) {
 	s.mu.Lock()
 	s.downstreams[addr] = c
 	s.mu.Unlock()
@@ -487,6 +495,9 @@ func (s *Server) authNames() []string {
 	return result
 }
 
+// Name returns the name of server.
+func (s *Server) Name() string { return s.name }
+
 func (s *Server) doConnectionCloseHandlers(qconn quic.Connection) {
 	for _, h := range s.connectionCloseHandlers {
 		h(qconn)

core/server_options.go

@@ -9,19 +9,13 @@ import (
 	"golang.org/x/exp/slog"
 )
 
-const (
-	// DefaultListenAddr is the default address to listen.
-	DefaultListenAddr = "0.0.0.0:9000"
-)
-
 // ServerOption is the option for server.
 type ServerOption func(*serverOptions)
 
 // ServerOptions are the options for YoMo server.
 type serverOptions struct {
 	quicConfig  *quic.Config
 	tlsConfig   *tls.Config
-	addr        string
 	auths       map[string]auth.Authentication
 	logger      *slog.Logger
 	alpnHandler func(proto string) error
@@ -33,7 +27,6 @@ func defaultServerOptions() *serverOptions {
 	opts := &serverOptions{
 		quicConfig: DefalutQuicConfig,
 		tlsConfig:  nil,
-		addr:       DefaultListenAddr,
 		auths:      map[string]auth.Authentication{},
 		logger:     logger,
 	}
@@ -44,13 +37,6 @@ func defaultServerOptions() *serverOptions {
 	return opts
 }
 
-// WithAddr sets the server address.
-func WithAddr(addr string) ServerOption {
-	return func(o *serverOptions) {
-		o.addr = addr
-	}
-}
-
 // WithAuth sets the server authentication method.
 func WithAuth(name string, args ...string) ServerOption {
 	return func(o *serverOptions) {