Repository: lf-edge/ekuiper
Pull request #1526 (2022-11-24 01:16:41)

chore: fix upload artifact error in github workflow


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index 60e824c0..1c53c514 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -68,7 +68,7 @@ jobs:
     - uses: actions/upload-artifact@v3
       with:
         name: packages
-        path: _packages/.
+        path: _packages/
 
   build-on-mac:
     runs-on: macos-latest
@@ -91,7 +91,7 @@ jobs:
     - uses: actions/upload-artifact@v3
       with:
         name: packages-mac
-        path: _packages/.
+        path: _packages/
 
   build-docker-images:
     runs-on: ubuntu-latest
@@ -274,7 +274,7 @@ jobs:
     - uses: actions/upload-artifact@v3
       with:
         name: plugins
-        path: "_plugins/"
+        path: _plugins/
 
   build-kubernetes-tool:
     runs-on: ubuntu-latest
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index d7609bd9..9fc0a351 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -143,8 +143,8 @@ func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 			}
 		}
 	}
-
 	conf.Log.Infof("Installed portable plugin %s successfully", name)
+	runtime.GetPluginInsManager().CreateIns(&pi.PluginMeta)
 	return nil
 }
 
@@ -340,11 +340,6 @@ func (m *Manager) Delete(name string) error {
 	if !ok {
 		return fmt.Errorf("portable plugin %s is not found", name)
 	}
-	pm := runtime.GetPluginInsManager()
-	err := pm.Kill(name)
-	if err != nil {
-		conf.Log.Errorf("fail to kill portable plugin %s process, please try to kill it manually", name)
-	}
 	// unregister the plugin
 	m.reg.Delete(name)
 	// delete files and uninstall metas
@@ -367,5 +362,11 @@ func (m *Manager) Delete(name string) error {
 		os.Remove(p)
 	}
 	_ = os.RemoveAll(path.Join(m.pluginDir, name))
+	// Kill the process in the end, and return error if it cannot be deleted
+	pm := runtime.GetPluginInsManager()
+	err := pm.Kill(name)
+	if err != nil {
+		return fmt.Errorf("fail to kill portable plugin %s process, please try to kill it manually", name)
+	}
 	return nil
 }
diff --git a/internal/plugin/portable/runtime/connection.go b/internal/plugin/portable/runtime/connection.go
index 22f6baa0..2d74ad16 100644
--- a/internal/plugin/portable/runtime/connection.go
+++ b/internal/plugin/portable/runtime/connection.go
@@ -62,6 +62,15 @@ func (r *NanomsgReqChannel) SendCmd(arg []byte) error {
 	r.Lock()
 	defer r.Unlock()
 	if err := r.sock.Send(arg); err != nil {
+		if err == mangos.ErrProtoState {
+			_, err = r.sock.Recv()
+			if err == nil {
+				err = r.sock.Send(arg)
+				if err == nil {
+					return nil
+				}
+			}
+		}
 		return fmt.Errorf("can't send message on control rep socket: %s", err.Error())
 	}
 	if msg, err := r.sock.Recv(); err != nil {
@@ -85,7 +94,7 @@ func (r *NanomsgReqChannel) Handshake() error {
 		return err
 	}
 	_, err = r.sock.Recv()
-	if err != nil {
+	if err != nil && err != mangos.ErrProtoState {
 		return err
 	}
 	err = r.sock.SetOption(mangos.OptionRecvDeadline, t)
@@ -124,6 +133,15 @@ func (r *NanomsgReqRepChannel) Req(arg []byte) ([]byte, error) {
 	r.Lock()
 	defer r.Unlock()
 	if err := r.sock.Send(arg); err != nil {
+		if err == mangos.ErrProtoState { // resend if protocol state wrong, because of plugin restart or other problems
+			err = r.Handshake()
+			if err == nil {
+				err = r.sock.Send(arg)
+				if err == nil {
+					return r.sock.Recv()
+				}
+			}
+		}
 		return nil, fmt.Errorf("can't send message on function rep socket: %s", err.Error())
 	}
 	return r.sock.Recv()
@@ -132,7 +150,10 @@ func (r *NanomsgReqRepChannel) Req(arg []byte) ([]byte, error) {
 // Handshake should only be called once
 func (r *NanomsgReqRepChannel) Handshake() error {
 	_, err := r.sock.Recv()
-	return err
+	if err != nil && err != mangos.ErrProtoState {
+		return err
+	}
+	return nil
 }
 
 func CreateSourceChannel(ctx api.StreamContext) (DataInChannel, error) {
diff --git a/internal/plugin/portable/runtime/function.go b/internal/plugin/portable/runtime/function.go
index 81176b20..6d45ea99 100644
--- a/internal/plugin/portable/runtime/function.go
+++ b/internal/plugin/portable/runtime/function.go
@@ -36,7 +36,7 @@ func NewPortableFunc(symbolName string, reg *PluginMeta) (*PortableFunc, error)
 	// Setup channel and route the data
 	conf.Log.Infof("Start running portable function meta %+v", reg)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(reg, PortbleConf)
+	ins, err := pm.getOrStartProcess(reg, PortbleConf, false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index 3c210cf3..0e6cad0f 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -35,25 +35,40 @@ var PortbleConf = &PortableConfig{
 	SendTimeout: 1000,
 }
 
+// PluginIns created at two scenarios
+// 1. At runtime, plugin is created/updated: in order to be able to reload rules that already uses previous ins
+// 2. At system start/restart, when plugin is used by a rule
+// Once created, never deleted until delete plugin command or system shutdown
 type PluginIns struct {
-	process      *os.Process
-	ctrlChan     ControlChannel
-	runningCount int
-	name         string
+	sync.RWMutex
+	name     string
+	ctrlChan ControlChannel // the same lifecycle as pluginIns, once created keep listening
+	// audit the commands, so that when restarting the plugin, we can replay the commands
+	commands map[Meta][]byte
+	process  *os.Process // created when used by rule and deleted when no rule uses it
 }
 
 func NewPluginIns(name string, ctrlChan ControlChannel, process *os.Process) *PluginIns {
-	// if process is not passed, it is run in simulator mode. Then do not count running.
-	// so that it won't be automatically close.
-	rc := 0
-	if process == nil {
-		rc = 1
+	return &PluginIns{
+		process:  process,
+		ctrlChan: ctrlChan,
+		name:     name,
+		commands: make(map[Meta][]byte),
 	}
+}
+
+func NewPluginInsForTest(name string, ctrlChan ControlChannel) *PluginIns {
+	commands := make(map[Meta][]byte)
+	commands[Meta{
+		RuleId:     "test",
+		OpId:       "test",
+		InstanceId: 0,
+	}] = []byte{}
 	return &PluginIns{
-		process:      process,
-		ctrlChan:     ctrlChan,
-		runningCount: rc,
-		name:         name,
+		process:  nil,
+		ctrlChan: ctrlChan,
+		name:     name,
+		commands: commands,
 	}
 }
 
@@ -72,7 +87,9 @@ func (i *PluginIns) StartSymbol(ctx api.StreamContext, ctrl *Control) error {
 	}
 	err = i.ctrlChan.SendCmd(jsonArg)
 	if err == nil {
-		i.runningCount++
+		i.Lock()
+		i.commands[ctrl.Meta] = jsonArg
+		i.Unlock()
 		ctx.GetLogger().Infof("started symbol %s", ctrl.SymbolName)
 	}
 	return err
@@ -92,24 +109,30 @@ func (i *PluginIns) StopSymbol(ctx api.StreamContext, ctrl *Control) error {
 		return err
 	}
 	err = i.ctrlChan.SendCmd(jsonArg)
-	i.runningCount--
-	ctx.GetLogger().Infof("stopped symbol %s", ctrl.SymbolName)
-	if i.runningCount == 0 {
-		err := GetPluginInsManager().Kill(i.name)
-		if err != nil {
-			ctx.GetLogger().Infof("fail to stop plugin %s: %v", i.name, err)
-			return err
+	if err == nil {
+		referred := false
+		i.Lock()
+		delete(i.commands, ctrl.Meta)
+		referred = len(i.commands) > 0
+		i.Unlock()
+		ctx.GetLogger().Infof("stopped symbol %s", ctrl.SymbolName)
+		if !referred {
+			err := GetPluginInsManager().Kill(i.name)
+			if err != nil {
+				ctx.GetLogger().Infof("fail to stop plugin %s: %v", i.name, err)
+				return err
+			}
+			ctx.GetLogger().Infof("stop plugin %s", i.name)
 		}
-		ctx.GetLogger().Infof("stop plugin %s", i.name)
 	}
 	return err
 }
 
+// Stop intentionally
 func (i *PluginIns) Stop() error {
 	var err error
-	if i.ctrlChan != nil {
-		err = i.ctrlChan.Close()
-	}
+	i.RLock()
+	defer i.RUnlock()
 	if i.process != nil { // will also trigger process exit clean up
 		err = i.process.Kill()
 	}
@@ -138,6 +161,7 @@ func (p *pluginInsManager) getPluginIns(name string) (*PluginIns, bool) {
 	return ins, ok
 }
 
+// deletePluginIns should only run when there is no state aka. commands
 func (p *pluginInsManager) deletePluginIns(name string) {
 	p.Lock()
 	defer p.Unlock()
@@ -151,27 +175,56 @@ func (p *pluginInsManager) AddPluginIns(name string, ins *PluginIns) {
 	p.instances[name] = ins
 }
 
+// CreateIns Run when plugin is created/updated
+func (p *pluginInsManager) CreateIns(pluginMeta *PluginMeta) {
+	p.Lock()
+	defer p.Unlock()
+	if ins, ok := p.instances[pluginMeta.Name]; ok {
+		if len(ins.commands) != 0 {
+			go p.getOrStartProcess(pluginMeta, PortbleConf, true)
+		}
+	}
+}
+
 // getOrStartProcess Control the plugin process lifecycle.
 // Need to manage the resources: instances map, control socket, plugin process
+// May be called at plugin creation or restart with previous state(ctrlCh, commands)
+// PluginIns is created by plugin manager but started by rule/funcop.
+// During plugin delete/update, if the commands is not empty, keep the ins for next creation and restore
 // 1. During creation, clean up those resources for any errors in defer immediately after the resource is created.
 // 2. During plugin running, when detecting plugin process exit, clean up those resources for the current ins.
-func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *PortableConfig) (*PluginIns, error) {
+func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *PortableConfig, pluginCreation bool) (*PluginIns, error) {
 	p.Lock()
 	defer p.Unlock()
-	if ins, ok := p.instances[pluginMeta.Name]; ok {
-		return ins, nil
+	var (
+		ins *PluginIns
+		ok  bool
+	)
+	// run initialization for firstly creating plugin instance
+	ins, ok = p.instances[pluginMeta.Name]
+	if !ok {
+		ins = NewPluginIns(pluginMeta.Name, nil, nil)
+		p.instances[pluginMeta.Name] = ins
 	}
-
-	conf.Log.Infof("create control channel")
-	ctrlChan, err := CreateControlChannel(pluginMeta.Name)
-	if err != nil {
-		return nil, fmt.Errorf("can't create new control channel: %s", err.Error())
+	// ins process has not run yet
+	if !pluginCreation && len(ins.commands) != 0 {
+		return ins, nil
 	}
-	defer func() {
+	// should only happen for first start, then the ctrl channel will keep running
+	if ins.ctrlChan == nil {
+		conf.Log.Infof("create control channel")
+		ctrlChan, err := CreateControlChannel(pluginMeta.Name)
 		if err != nil {
-			_ = ctrlChan.Close()
+			return nil, fmt.Errorf("can't create new control channel: %s", err.Error())
 		}
-	}()
+		defer func() {
+			if err != nil {
+				_ = ctrlChan.Close()
+			}
+		}()
+		ins.ctrlChan = ctrlChan
+	}
+	// init or restart all need to run the process
 	conf.Log.Infof("executing plugin")
 	jsonArg, err := json.Marshal(pconf)
 	if err != nil {
@@ -216,23 +269,39 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 			conf.Log.Printf("plugin executable %s stops with error %v", pluginMeta.Executable, err)
 		}
 		// must make sure the plugin ins is not cleaned up yet by checking the process identity
+		// clean up for stop unintentionally
 		if ins, ok := p.getPluginIns(pluginMeta.Name); ok && ins.process == cmd.Process {
-			if ins.ctrlChan != nil {
-				_ = ins.ctrlChan.Close()
+			ins.Lock()
+			if len(ins.commands) == 0 {
+				if ins.ctrlChan != nil {
+					_ = ins.ctrlChan.Close()
+				}
+				p.deletePluginIns(pluginMeta.Name)
 			}
-			p.deletePluginIns(pluginMeta.Name)
+			ins.process = nil
+			ins.Unlock()
 		}
 		return nil
 	})
-
 	conf.Log.Println("waiting handshake")
-	err = ctrlChan.Handshake()
+	err = ins.ctrlChan.Handshake()
 	if err != nil {
 		return nil, fmt.Errorf("plugin %s control handshake error: %v", pluginMeta.Executable, err)
 	}
-	ins := NewPluginIns(pluginMeta.Name, ctrlChan, process)
+	ins.process = process
 	p.instances[pluginMeta.Name] = ins
 	conf.Log.Println("plugin start running")
+	// restore symbols by sending commands when restarting plugin
+	conf.Log.Info("restore plugin symbols")
+	for m, c := range ins.commands {
+		go func(key Meta, jsonArg []byte) {
+			e := ins.ctrlChan.SendCmd(jsonArg)
+			if e != nil {
+				conf.Log.Errorf("send command to %v error: %v", key, e)
+			}
+		}(m, c)
+	}
+
 	return ins, nil
 }
 
@@ -242,9 +311,9 @@ func (p *pluginInsManager) Kill(name string) error {
 	var err error
 	if ins, ok := p.instances[name]; ok {
 		err = ins.Stop()
-		delete(p.instances, name)
 	} else {
-		return fmt.Errorf("instance %s not found", name)
+		conf.Log.Warnf("instance %s not found when deleting", name)
+		return nil
 	}
 	return err
 }
@@ -255,7 +324,6 @@ func (p *pluginInsManager) KillAll() error {
 	for _, ins := range p.instances {
 		_ = ins.Stop()
 	}
-	p.instances = make(map[string]*PluginIns)
 	return nil
 }
 
diff --git a/internal/plugin/portable/runtime/shared.go b/internal/plugin/portable/runtime/shared.go
index 38c84f14..55e6f9e8 100644
--- a/internal/plugin/portable/runtime/shared.go
+++ b/internal/plugin/portable/runtime/shared.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -33,7 +33,7 @@ type FuncMeta struct {
 
 type Control struct {
 	SymbolName string                 `json:"symbolName"`
-	Meta       *Meta                  `json:"meta,omitempty"`
+	Meta       Meta                   `json:"meta"`
 	PluginType string                 `json:"pluginType"`
 	DataSource string                 `json:"dataSource,omitempty"`
 	Config     map[string]interface{} `json:"config,omitempty"`
diff --git a/internal/plugin/portable/runtime/sink.go b/internal/plugin/portable/runtime/sink.go
index fbc01e03..e8ecaceb 100644
--- a/internal/plugin/portable/runtime/sink.go
+++ b/internal/plugin/portable/runtime/sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -43,7 +43,7 @@ func (ps *PortableSink) Configure(props map[string]interface{}) error {
 func (ps *PortableSink) Open(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Start running portable sink %s with conf %+v", ps.symbolName, ps.props)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf)
+	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf, false)
 	if err != nil {
 		return err
 	}
@@ -51,7 +51,7 @@ func (ps *PortableSink) Open(ctx api.StreamContext) error {
 
 	// Control: send message to plugin to ask starting symbol
 	c := &Control{
-		Meta: &Meta{
+		Meta: Meta{
 			RuleId:     ctx.GetRuleId(),
 			OpId:       ctx.GetOpId(),
 			InstanceId: ctx.GetInstanceId(),
diff --git a/internal/plugin/portable/runtime/source.go b/internal/plugin/portable/runtime/source.go
index 33632f09..5346c926 100644
--- a/internal/plugin/portable/runtime/source.go
+++ b/internal/plugin/portable/runtime/source.go
@@ -44,7 +44,7 @@ func NewPortableSource(symbolName string, reg *PluginMeta) *PortableSource {
 func (ps *PortableSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	ctx.GetLogger().Infof("Start running portable source %s with datasource %s and conf %+v", ps.symbolName, ps.topic, ps.props)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf)
+	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf, false)
 	if err != nil {
 		infra.DrainError(ctx, err, errCh)
 		return
@@ -60,7 +60,7 @@ func (ps *PortableSource) Open(ctx api.StreamContext, consumer chan<- api.Source
 
 	// Control: send message to plugin to ask starting symbol
 	c := &Control{
-		Meta: &Meta{
+		Meta: Meta{
 			RuleId:     ctx.GetRuleId(),
 			OpId:       ctx.GetOpId(),
 			InstanceId: ctx.GetInstanceId(),
@@ -72,6 +72,7 @@ func (ps *PortableSource) Open(ctx api.StreamContext, consumer chan<- api.Source
 	}
 	err = ins.StartSymbol(ctx, c)
 	if err != nil {
+		ctx.GetLogger().Error(err)
 		infra.DrainError(ctx, err, errCh)
 		_ = dataCh.Close()
 		return


Test patch:
diff --git a/internal/plugin/portable/runtime/connection_test.go b/internal/plugin/portable/runtime/connection_test.go
index 57df589a..0579fff2 100644
--- a/internal/plugin/portable/runtime/connection_test.go
+++ b/internal/plugin/portable/runtime/connection_test.go
@@ -112,8 +112,8 @@ func TestControlCh(t *testing.T) {
 
 	// 5. no handshake
 	err = ch.SendCmd(okMsg)
-	if err == nil || err.Error() != "can't send message on control rep socket: incorrect protocol state" {
-		t.Errorf("5th process: send command should have error but got %v", err)
+	if err != nil {
+		t.Errorf("5th process: send command should auto handshake but got %v", err)
 	}
 	err = ch.Handshake()
 	if err != nil {
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager_test.go b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
index b1250b5a..4444d9af 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager_test.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
@@ -50,11 +50,7 @@ func TestPluginInstance(t *testing.T) {
 		t.Errorf("can't ack handshake: %s", err.Error())
 		return
 	}
-	ins := &PluginIns{
-		name:     "test",
-		process:  nil,
-		ctrlChan: ch,
-	}
+	ins := NewPluginIns("test", ch, nil)
 	var tests = []struct {
 		c  *Control
 		sj string
@@ -63,7 +59,7 @@ func TestPluginInstance(t *testing.T) {
 		{
 			c: &Control{
 				SymbolName: "symbol1",
-				Meta: &Meta{
+				Meta: Meta{
 					RuleId:     "rule1",
 					OpId:       "op1",
 					InstanceId: 0,
@@ -77,7 +73,7 @@ func TestPluginInstance(t *testing.T) {
 		}, {
 			c: &Control{
 				SymbolName: "symbol2",
-				Meta: &Meta{
+				Meta: Meta{
 					RuleId:     "rule1",
 					OpId:       "op2",
 					InstanceId: 0,
@@ -89,7 +85,7 @@ func TestPluginInstance(t *testing.T) {
 		}, {
 			c: &Control{
 				SymbolName: "symbol3",
-				Meta: &Meta{
+				Meta: Meta{
 					RuleId:     "rule1",
 					OpId:       "op3",
 					InstanceId: 0,
diff --git a/tools/plugin_server/plugin_test_server.go b/tools/plugin_server/plugin_test_server.go
index 0191c50b..9fc2a467 100644
--- a/tools/plugin_server/plugin_test_server.go
+++ b/tools/plugin_server/plugin_test_server.go
@@ -102,7 +102,7 @@ func startPluginIns(info *portable.PluginInfo) (*runtime.PluginIns, error) {
 		return nil, fmt.Errorf("plugin %s control handshake error: %v", info.Name, err)
 	}
 	conf.Log.Println("plugin start running")
-	return runtime.NewPluginIns(info.Name, ctrlChan, nil), nil
+	return runtime.NewPluginInsForTest(info.Name, ctrlChan), nil
 }
 
 func createRestServer(ip string, port int) *http.Server {