Issue 2348

sqle/driver/plugin_adapter_v2.go

@@ -5,6 +5,7 @@ import (
 	sqlDriver "database/sql/driver"
 	"errors"
 	"fmt"
+	"os"
 	"sync"
 
 	driverV2 "github.com/actiontech/sqle/sqle/driver/v2"
@@ -18,11 +19,12 @@ import (
 )
 
 type PluginProcessorV2 struct {
-	cfg     func(cmdBase string, cmdArgs []string) *goPlugin.ClientConfig
-	cmdBase string
-	cmdArgs []string
-	client  *goPlugin.Client
-	meta    *driverV2.DriverMetas
+	cfg               func(cmdBase string, cmdArgs []string) *goPlugin.ClientConfig
+	cmdBase           string
+	cmdArgs           []string
+	client            *goPlugin.Client
+	meta              *driverV2.DriverMetas
+	pluginPidFilePath string
 	sync.Mutex
 }
 
@@ -142,6 +144,7 @@ func (d *PluginProcessorV2) Stop() error {
 	if d.client != nil {
 		d.client.Kill()
 	}
+	os.Remove(d.pluginPidFilePath)
 	d.Unlock()
 	return nil
 }

sqle/driver/plugin_manager.go

@@ -5,7 +5,9 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strconv"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/actiontech/sqle/sqle/config"
@@ -179,6 +181,19 @@ func (pm *pluginManager) Start(pluginDir string, pluginConfigList []config.Plugi
 
 	// register plugin
 	for _, p := range plugins {
+
+		pluginPidFilePath := filepath.Join(pluginDir, "pidfile", p.Name()+".pid")
+		process, err := GetProcessByPidFile(pluginPidFilePath)
+		if err != nil {
+			log.NewEntry().Warnf("get plugin %s process failed, error: %v", pluginPidFilePath, err)
+		}
+		if process != nil {
+			err = KillProcess(process)
+			if err != nil {
+				log.NewEntry().Warnf("kill plugin process [%v] failed, error: %v", process.Pid, err)
+			}
+		}
+
 		cmdBase := filepath.Join(pluginDir, p.Name())
 		cmdArgs := make([]string, 0)
 
@@ -197,17 +212,20 @@ func (pm *pluginManager) Start(pluginDir string, pluginConfigList []config.Plugi
 		}
 
 		client := goPlugin.NewClient(getClientConfig(cmdBase, cmdArgs))
-		_, err := client.Client()
+		_, err = client.Client()
 		if err != nil {
 			return fmt.Errorf("plugin %v failed to start, error: %v Please check the sqled.log for more details", p.Name(), err)
 		}
-
+		err = WritePidFile(pluginPidFilePath, int64(client.ReattachConfig().Pid))
+		if err != nil {
+			return fmt.Errorf("write plugin %s pid file failed, error: %v", pluginPidFilePath, err)
+		}
 		var pp PluginProcessor
 		switch client.NegotiatedVersion() {
 		case driverV1.ProtocolVersion:
 			pp = &PluginProcessorV1{cfg: getClientConfig, cmdBase: cmdBase, cmdArgs: cmdArgs, client: client}
 		case driverV2.ProtocolVersion:
-			pp = &PluginProcessorV2{cfg: getClientConfig, cmdBase: cmdBase, cmdArgs: cmdArgs, client: client}
+			pp = &PluginProcessorV2{cfg: getClientConfig, cmdBase: cmdBase, cmdArgs: cmdArgs, client: client, pluginPidFilePath: pluginPidFilePath}
 		}
 		if err := pm.register(pp); err != nil {
 			stopErr := pp.Stop()
@@ -243,3 +261,90 @@ func (pm *pluginManager) OpenPlugin(l *logrus.Entry, pluginName string, cfg *dri
 	}
 	return pm.pluginProcessors[pluginName].Open(l, cfg)
 }
+
+// 根据pid文件获取进程信息
+func GetProcessByPidFile(pluginPidFile string) (*os.Process, error) {
+	if _, err := os.Stat(pluginPidFile); err != nil {
+		if !os.IsNotExist(err) {
+			return nil, err
+		}
+	} else {
+		pidContent, err := os.ReadFile(pluginPidFile)
+		if err != nil {
+			return nil, err
+		}
+		if len(pidContent) == 0 {
+			return nil, nil
+		}
+		pid, err := strconv.Atoi(string(pidContent))
+		if err != nil {
+			return nil, err
+		}
+		// 获取进程
+		process, err := GetProcessByPid(pid)
+		if err != nil {
+			return nil, err
+		}
+		return process, nil
+	}
+	return nil, nil
+}
+
+// 根据pid获取进程信息，若进程已退出则返回nil
+func GetProcessByPid(pid int) (*os.Process, error) {
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		return nil, err
+	}
+	// 检查进程是否存在的方式
+	err = process.Signal(syscall.Signal(0))
+	if err != nil {
+		if errors.Is(err, os.ErrProcessDone) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return process, nil
+}
+
+// 退出进程
+func KillProcess(process *os.Process) error {
+	doneChan := time.NewTicker(2 * time.Second)
+	defer doneChan.Stop()
+	for {
+		select {
+		case <-doneChan.C:
+			log.NewEntry().Warnf("stop plugin process [%v] failed, just kill it ", process.Pid)
+			err := process.Kill()
+			if err != nil {
+				return err
+			}
+			return nil
+		default:
+			err := process.Signal(syscall.SIGTERM)
+			if errors.Is(err, os.ErrProcessDone) {
+				return nil
+			}
+		}
+	}
+}
+
+func WritePidFile(pidFilePath string, pid int64) error {
+	_, err := os.Stat(pidFilePath)
+	if os.IsNotExist(err) {
+		if err := os.MkdirAll(filepath.Dir(pidFilePath), 0644); err != nil {
+			return err
+		}
+	}
+	file, err := os.OpenFile(pidFilePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	_, err = fmt.Fprintf(file, "%d", pid)
+	if err != nil {
+		return err
+	}
+	return nil
+}