feat(conf): allow to set portable init timeout value

Some plugin, especially when loaded AI models, need a lot of init time

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

docs/en_US/configuration/global_configurations.md

@@ -191,6 +191,8 @@ This section configures the portable plugin runtime.
       # The executable of python. Specify this if you have multiple python instances in your system
       # or other circumstance where the python executable cannot be successfully invoked through the default command.
       pythonBin: python
+      # control init timeout in ms. If the init time is longer than this value, the plugin will be terminated.
+      initTimeout: 5000
 ```
 
 ## Ruleset Provision

docs/zh_CN/configuration/global_configurations.md

@@ -191,6 +191,8 @@ GET http://localhost:9081/plugins/functions/prebuild
       # 配置 python 可执行文件的位置或命令。
       # 若系统中有多个 python 版本，可通过此配置指定具体的 python 地址。
       pythonBin: python
+      # 控制插件初始化超时时间，单位为毫秒。eKuiper portable 插件运行时会等待插件初始化以完成握手，若超时则终止插件进程
+      initTimeout: 5000
 ```
 
 ## 初始化规则集

etc/kuiper.yaml

@@ -99,4 +99,6 @@ store:
 portable:
   # The executable of python. Specify this if you have multiple python instances in your system
   # or other circumstance where the python executable cannot be successfully invoked through the default command.
-  pythonBin: python
+  pythonBin: python
+  # control init timeout in ms. If the init time is longer than this value, the plugin will be terminated.
+  initTimeout: 5000

internal/conf/conf.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -147,7 +147,8 @@ type KuiperConf struct {
 		}
 	}
 	Portable struct {
-		PythonBin string `yaml:"pythonBin"`
+		PythonBin   string `yaml:"pythonBin"`
+		InitTimeout int    `yaml:"initTimeout"`
 	}
 }
 
@@ -226,6 +227,9 @@ func InitConf() {
 	if Config.Portable.PythonBin == "" {
 		Config.Portable.PythonBin = "python"
 	}
+	if Config.Portable.InitTimeout <= 0 {
+		Config.Portable.InitTimeout = 5000
+	}
 	if Config.Source == nil {
 		Config.Source = &SourceConf{}
 	}

internal/plugin/portable/runtime/connection.go

@@ -89,7 +89,7 @@ func (r *NanomsgReqChannel) Handshake() error {
 	if err != nil {
 		return err
 	}
-	err = r.sock.SetOption(mangos.OptionRecvDeadline, 5*time.Second)
+	err = r.sock.SetOption(mangos.OptionRecvDeadline, time.Duration(conf.Config.Portable.InitTimeout)*time.Millisecond)
 	if err != nil {
 		return err
 	}

internal/plugin/portable/runtime/connection_test.go

@@ -17,6 +17,7 @@ package runtime
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -29,6 +30,10 @@ import (
 	"time"
 )
 
+func init() {
+	testx.InitEnv()
+}
+
 var okMsg = []byte("ok")
 
 func TestControlCh(t *testing.T) {