Repository: lf-edge/ekuiper
Pull request #1524 (2022-11-23 01:39:13)

opt(portable): hot reload support
When deleting portable plugin, the rules are kept running. When recreating the plugin, the rule will recover automatically

This is done by record the symbol commands and replay when new plugin instance starts.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
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
diff --git a/internal/topo/rule/ruleState.go b/internal/topo/rule/ruleState.go
index 66cc5596..967274a7 100644
--- a/internal/topo/rule/ruleState.go
+++ b/internal/topo/rule/ruleState.go
@@ -169,7 +169,7 @@ func (rs *RuleState) runTopo(ctx context.Context) {
 					return nil
 				}
 			}
-			if count <= option.Attempts {
+			if count < option.Attempts {
 				if d > option.MaxDelay {
 					d = option.MaxDelay
 				}
@@ -211,42 +211,35 @@ func (rs *RuleState) runTopo(ctx context.Context) {
 func (rs *RuleState) Start() error {
 	rs.Lock()
 	defer rs.Unlock()
-	switch rs.triggered {
-	case 1:
-		return fmt.Errorf("Rule %s is already starting", rs.RuleId)
-	case -1:
-		return fmt.Errorf("Rule %s is already deleted", rs.RuleId)
-	default:
-		// Start from stop
-		if rs.Topology == nil {
-			if tp, err := planner.Plan(rs.Rule); err != nil {
-				return err
-			} else {
-				rs.Topology = tp
-			}
-		} // else start after create
-		rs.triggered = 1
-		rs.ActionCh <- ActionSignalStart
-		return nil
+	if rs.triggered == -1 {
+		return fmt.Errorf("rule %s is already deleted", rs.RuleId)
 	}
+	if rs.Topology == nil {
+		if tp, err := planner.Plan(rs.Rule); err != nil {
+			return err
+		} else {
+			rs.Topology = tp
+		}
+	} // else start after create
+	rs.triggered = 1
+	rs.ActionCh <- ActionSignalStart
+	return nil
 }
 
 // Stop remove the Topology
 func (rs *RuleState) Stop() error {
 	rs.Lock()
 	defer rs.Unlock()
-	switch rs.triggered {
-	case 0:
-		return fmt.Errorf("Rule %s is already stopping", rs.RuleId)
-	case -1:
-		return fmt.Errorf("Rule %s is already deleted", rs.RuleId)
-	default:
-		rs.triggered = 0
+	if rs.triggered == -1 {
+		return fmt.Errorf("rule %s is already deleted", rs.RuleId)
+	}
+	rs.triggered = 0
+	if rs.Topology != nil {
 		rs.Topology.Cancel()
 		rs.Topology = nil
-		rs.ActionCh <- ActionSignalStop
-		return nil
 	}
+	rs.ActionCh <- ActionSignalStop
+	return nil
 }
 
 func (rs *RuleState) Close() error {


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/test/change_rule_status.jmx b/test/change_rule_status.jmx
index 5e947b0e..ec0c04ad 100644
--- a/test/change_rule_status.jmx
+++ b/test/change_rule_status.jmx
@@ -241,7 +241,7 @@
           <hashTree>
             <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
               <collectionProp name="Asserion.test_strings">
-                <stringProp name="-652969125">Rule rule1 is already starting</stringProp>
+                <stringProp name="-652969125">Rule rule1 was started</stringProp>
               </collectionProp>
               <stringProp name="Assertion.custom_message"></stringProp>
               <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
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 {