Add web server & auto handler parser for show commands (#169)

Signed-off-by: Congqi Xia <[email protected]>

bapps/webserver.go

@@ -0,0 +1,272 @@
+package bapps
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"reflect"
+	"strconv"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/milvus-io/birdwatcher/common"
+	"github.com/milvus-io/birdwatcher/configs"
+	"github.com/milvus-io/birdwatcher/framework"
+	"github.com/milvus-io/birdwatcher/models"
+	"github.com/milvus-io/birdwatcher/states"
+	etcdversion "github.com/milvus-io/birdwatcher/states/etcd/version"
+)
+
+type WebServerApp struct {
+	port   int
+	config *configs.Config
+}
+
+type InstanceInfo struct {
+	EtcdAddr string `form:"etcd"`
+	RootPath string `form:"rootPath"`
+}
+
+func (app *WebServerApp) Run(states.State) {
+	r := gin.Default()
+	etcdversion.SetVersion(models.GTEVersion2_2)
+
+	r.GET("/version", func(c *gin.Context) {
+		c.JSON(http.StatusOK, gin.H{"version": common.Version})
+	})
+
+	app.ParseRouter(r, &states.InstanceState{})
+
+	r.Run(fmt.Sprintf(":%d", app.port))
+}
+
+func (app *WebServerApp) ParseRouter(r *gin.Engine, s states.State) {
+	v := reflect.ValueOf(s)
+	tp := v.Type()
+
+	for i := 0; i < v.NumMethod(); i++ {
+		mt := tp.Method(i)
+
+		// parse method like with pattern %Command
+		if !strings.HasSuffix(mt.Name, "Command") {
+			continue
+		}
+
+		// fmt.Println("parsing method", mt.Name)
+		app.parseMethod(r, mt, mt.Name)
+	}
+}
+
+func (app *WebServerApp) parseMethod(r *gin.Engine, mt reflect.Method, name string) {
+	// v := reflect.ValueOf(s)
+	t := mt.Type
+	var use string
+	var paramType reflect.Type
+
+	if t.NumIn() == 0 {
+		// shall not be reached
+		return
+	}
+	if t.NumIn() > 1 {
+		// should be context.Context
+		in := t.In(1)
+		if !in.Implements(reflect.TypeOf((*context.Context)(nil)).Elem()) {
+			return
+		}
+	}
+	if t.NumIn() > 2 {
+		// should be CmdParam
+		in := t.In(2)
+		if !in.Implements(reflect.TypeOf((*framework.CmdParam)(nil)).Elem()) {
+			return
+		}
+		cp, ok := reflect.New(in.Elem()).Interface().(framework.CmdParam)
+		if !ok {
+			fmt.Println("conversion failed", in.Name())
+		} else {
+			paramType = in
+			use, _ = cp.Desc()
+		}
+	}
+	if t.NumOut() == 0 {
+		fmt.Printf("%s not output\n", name)
+		return
+	}
+
+	if t.NumOut() > 0 {
+		// should be ResultSet
+		out := t.Out(0)
+		if !out.Implements(reflect.TypeOf((*framework.ResultSet)(nil)).Elem()) {
+			fmt.Printf("%s output not ResultSet\n", name)
+			return
+		}
+	}
+
+	//fmt.Println(mt.Name)
+	cp := reflect.New(paramType.Elem()).Interface().(framework.CmdParam)
+	fUse, _ := states.GetCmdFromFlag(cp)
+	if len(use) == 0 {
+		use = fUse
+	}
+
+	if len(use) == 0 {
+		fnName := mt.Name
+		use = strings.ToLower(fnName[:len(fnName)-8])
+	}
+	uses := states.ParseUseSegments(use)
+	lastKw := uses[len(uses)-1]
+	// hard code, show xxx command only
+	if uses[0] != "show" {
+		return
+	}
+
+	// fmt.Printf("path: /%s\n", lastKw)
+
+	r.GET(fmt.Sprintf("/%s", lastKw), func(c *gin.Context) {
+
+		info := &InstanceInfo{}
+		c.ShouldBind(info)
+
+		start := states.Start(app.config)
+		s, err := start.Process(fmt.Sprintf("connect --etcd=%s --rootPath=%s", info.EtcdAddr, info.RootPath))
+
+		if err != nil {
+			c.Error(err)
+			return
+		}
+
+		v := reflect.ValueOf(s)
+		cp := reflect.New(paramType.Elem()).Interface().(framework.CmdParam)
+		setupDefaultValue(cp)
+		if err := app.BindCmdParam(c, cp); err != nil {
+			c.Error(err)
+			return
+		}
+
+		m := v.MethodByName(mt.Name)
+		results := m.Call([]reflect.Value{
+			reflect.ValueOf(c),
+			reflect.ValueOf(cp),
+		})
+
+		// reverse order, check error first
+		for i := 0; i < len(results); i++ {
+			result := results[len(results)-i-1]
+			switch {
+			case result.Type().Implements(reflect.TypeOf((*error)(nil)).Elem()):
+				// error nil, skip
+				if result.IsNil() {
+					continue
+				}
+				err := result.Interface().(error)
+				c.Error(err)
+				return
+			case result.Type().Implements(reflect.TypeOf((*framework.ResultSet)(nil)).Elem()):
+				if result.IsNil() {
+					continue
+				}
+				rs := result.Interface().(framework.ResultSet)
+				c.JSON(http.StatusOK, rs.Entities())
+				return
+			}
+		}
+
+		c.Error(errors.New("unexpected branch reached, no result set found"))
+	})
+}
+
+func (app *WebServerApp) BindCmdParam(c *gin.Context, cp framework.CmdParam) error {
+	v := reflect.ValueOf(cp)
+	if v.Kind() != reflect.Pointer {
+		return errors.New("param is not pointer")
+	}
+
+	for v.Kind() != reflect.Struct {
+		v = v.Elem()
+	}
+	tp := v.Type()
+
+	for i := 0; i < v.NumField(); i++ {
+		f := tp.Field(i)
+		if !f.IsExported() {
+			continue
+		}
+		name := f.Tag.Get("name")
+		rawStr, ok := c.GetQuery(name)
+		if !ok {
+			continue
+		}
+		switch f.Type.Kind() {
+		case reflect.Int64:
+			var dv int64
+			if v, err := strconv.ParseInt(rawStr, 10, 64); err == nil {
+				dv = v
+			}
+			v.Field(i).SetInt(dv)
+			fmt.Println("set default", f.Name, dv)
+		case reflect.String:
+			v.Field(i).SetString(rawStr)
+		case reflect.Bool:
+			var dv bool
+			if v, err := strconv.ParseBool(rawStr); err == nil {
+				dv = v
+			}
+			v.Field(i).SetBool(dv)
+		case reflect.Struct:
+			continue
+		default:
+			return fmt.Errorf("field %s with kind %s not supported yet", f.Name, f.Type.Kind())
+		}
+	}
+	return nil
+}
+
+func setupDefaultValue(p framework.CmdParam) {
+	v := reflect.ValueOf(p)
+	if v.Kind() != reflect.Pointer {
+		fmt.Println("param is not pointer")
+		return
+	}
+
+	for v.Kind() != reflect.Struct {
+		v = v.Elem()
+	}
+	tp := v.Type()
+
+	for i := 0; i < v.NumField(); i++ {
+		f := tp.Field(i)
+		if !f.IsExported() {
+			continue
+		}
+		defaultStr := f.Tag.Get("default")
+		switch f.Type.Kind() {
+		case reflect.Int64:
+			var dv int64
+			if v, err := strconv.ParseInt(defaultStr, 10, 64); err == nil {
+				dv = v
+			}
+			v.Field(i).SetInt(dv)
+			fmt.Println("set default", f.Name, dv)
+		case reflect.String:
+			v.Field(i).SetString(defaultStr)
+		case reflect.Bool:
+			var dv bool
+			if v, err := strconv.ParseBool(defaultStr); err == nil {
+				dv = v
+			}
+			v.Field(i).SetBool(dv)
+		case reflect.Struct:
+			continue
+		default:
+			fmt.Printf("field %s with kind %s not supported yet\n", f.Name, f.Type.Kind())
+		}
+	}
+}
+
+func NewWebServerApp(port int, config *configs.Config) *WebServerApp {
+	return &WebServerApp{
+		port:   port,
+		config: config,
+	}
+}

framework/resultset.go

@@ -0,0 +1,69 @@
+package framework
+
+type Format int32
+
+const (
+	FormatDefault Format = iota + 1
+	FormatPlain
+	FormatJSON
+	FormatTable
+)
+
+var (
+	name2Format = map[string]Format{
+		"default": FormatDefault,
+		"plain":   FormatPlain,
+		"json":    FormatJSON,
+		"table":   FormatTable,
+	}
+)
+
+// ResultSet is the interface for command result set.
+type ResultSet interface {
+	PrintAs(Format) string
+	Entities() any
+}
+
+// PresetResultSet implements Stringer and "memorize" output format.
+type PresetResultSet struct {
+	ResultSet
+	format Format
+}
+
+func (rs *PresetResultSet) String() string {
+	if rs.format < FormatDefault {
+		return rs.PrintAs(FormatDefault)
+	}
+	return rs.PrintAs(rs.format)
+}
+
+// NameFormat name to format mapping tool function.
+func NameFormat(name string) Format {
+	f, ok := name2Format[name]
+	if !ok {
+		return FormatDefault
+	}
+	return f
+}
+
+type ListResultSet[T any] struct {
+	Data []T
+}
+
+func (rs *ListResultSet[T]) Entities() any {
+	return rs.Data
+}
+
+func (rs *ListResultSet[T]) SetData(data []T) {
+	rs.Data = data
+}
+
+func NewListResult[LRS any, P interface {
+	*LRS
+	SetData([]E)
+}, E any](data []E) *LRS {
+	var t LRS
+	var p P = &t
+	p.SetData(data)
+	return &t
+}

main.go

@@ -17,6 +17,8 @@ import (
 var (
 	oneLineCommand = flag.String("olc", "", "one line command execution mode")
 	simple         = flag.Bool("simple", false, "use simple ui without suggestion and history")
+	restServer     = flag.Bool("rest", false, "rest server address")
+	webPort        = flag.Int("port", 8002, "listening port for web server")
 	printVersion   = flag.Bool("version", false, "print version")
 )
 
@@ -34,6 +36,8 @@ func main() {
 		appFactory = func(*configs.Config) bapps.BApp { return bapps.NewSimpleApp() }
 	case len(*oneLineCommand) > 0:
 		appFactory = func(*configs.Config) bapps.BApp { return bapps.NewOlcApp(*oneLineCommand) }
+	case *restServer:
+		appFactory = func(config *configs.Config) bapps.BApp { return bapps.NewWebServerApp(*webPort, config) }
 	default:
 		defer handleExit()
 		// open file and create if non-existent

models/channel_watch.go

@@ -59,6 +59,6 @@ func getVChannelInfo[info interface {
 		UnflushedSegmentIds: vchan.GetUnflushedSegmentIds(),
 		FlushedSegmentIds:   vchan.GetFlushedSegmentIds(),
 		DroppedSegmentIds:   vchan.GetDroppedSegmentIds(),
-		SeekPosition:        newMsgPosition(vchan.GetSeekPosition()),
+		SeekPosition:        NewMsgPosition(vchan.GetSeekPosition()),
 	}
 }

models/data_type.go

@@ -13,6 +13,8 @@ const (
 	DataTypeDouble       DataType = 11
 	DataTypeString       DataType = 20
 	DataTypeVarChar      DataType = 21
+	DataTypeArray        DataType = 22
+	DataTypeJSON         DataType = 23
 	DataTypeBinaryVector DataType = 100
 	DataTypeFloatVector  DataType = 101
 )
@@ -28,6 +30,8 @@ var DataTypename = map[int32]string{
 	11:  "Double",
 	20:  "String",
 	21:  "VarChar",
+	22:  "Array",
+	23:  "JSON",
 	100: "BinaryVector",
 	101: "FloatVector",
 }
@@ -43,6 +47,8 @@ var DataTypevalue = map[string]int32{
 	"Double":       11,
 	"String":       20,
 	"VarChar":      21,
+	"Array":        22,
+	"JSON":         23,
 	"BinaryVector": 100,
 	"FloatVector":  101,
 }