Repository: lf-edge/ekuiper
Pull request #2992 (2024-07-08 03:51:40)

feat: support ws client


Patch:
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index 7291ff56..5c94e02b 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -64,6 +64,8 @@ basic:
   gracefulShutdownTimeout: 3s
   connection:
     backoffMaxElapsedDuration: 3m
+  # enableResourceProfiling indicates whether to enable resource profiling for eKuiper. If it is enabled, the cpu usage of the rule will be recorded.
+  enableResourceProfiling: false
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:
diff --git a/go.mod b/go.mod
index 25f9687b..c866d8db 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/PaesslerAG/gval v1.2.2
 	github.com/PaesslerAG/jsonpath v0.1.1
+	github.com/Rookiecom/cpuprofile v1.0.2
 	github.com/alicebob/miniredis/v2 v2.30.0
 	github.com/apple/foundationdb/bindings/go v0.0.0-20240315202255-8943393e84fc
 	github.com/benbjohnson/clock v1.3.5
@@ -18,6 +19,7 @@ require (
 	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
 	github.com/golang-jwt/jwt/v5 v5.2.1
 	github.com/golang/protobuf v1.5.4
+	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.1
@@ -87,7 +89,6 @@ require (
 	github.com/go-playground/validator/v10 v10.19.0 // indirect
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
diff --git a/go.sum b/go.sum
index c7a6d088..7b733b57 100644
--- a/go.sum
+++ b/go.sum
@@ -21,6 +21,8 @@ github.com/PaesslerAG/gval v1.2.2/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbV
 github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
 github.com/PaesslerAG/jsonpath v0.1.1 h1:c1/AToHQMVsduPAa4Vh6xp2U0evy4t8SWp8imEsylIk=
 github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
+github.com/Rookiecom/cpuprofile v1.0.2 h1:wXGOmpSyCfoi7D0OB79srblNSw5Ccn0fTy4c537yWy4=
+github.com/Rookiecom/cpuprofile v1.0.2/go.mod h1:JEBcjNHceN6/g4ppueDEqWlbtbMGMyLy0YlT8QnKFbE=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrdtl/UvroE=
@@ -221,8 +223,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/pprof v0.0.0-20230207041349-798e818bf904/go.mod h1:uglQLonpP8qtYCYyzA+8c/9qtqgA3qsXGYqCPKARAFg=
-github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd h1:LjW4RcTwfcqOYGmD7UpFrn1gfBZ9mgu7QN5mSeFkCog=
-github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik=
+github.com/google/pprof v0.0.0-20240528025155-186aa0362fba h1:ql1qNgCyOB7iAEk8JTNM+zJrgIbnyCKX/wdlyPufP5g=
+github.com/google/pprof v0.0.0-20240528025155-186aa0362fba/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 4065f807..7b24524c 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -195,6 +195,7 @@ type KuiperConf struct {
 		EnableOpenZiti          bool              `yaml:"enableOpenZiti"`
 		AesKey                  string            `yaml:"aesKey"`
 		GracefulShutdownTimeout cast.DurationConf `yaml:"gracefulShutdownTimeout"`
+		EnableResourceProfiling bool              `yaml:"enableResourceProfiling"`
 	}
 	Rule   def.RuleOption
 	Sink   *SinkConf
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index d81ea012..f80bfa03 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 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.
@@ -65,6 +65,34 @@ func (f *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err
 	return f.decodeWithSchema(b, f.schema)
 }
 
+func (f *FastJsonConverter) DecodeField(_ api.StreamContext, b []byte, field string) (any, error) {
+	var p fastjson.Parser
+	v, err := p.ParseBytes(b)
+	if err != nil {
+		return nil, err
+	}
+	switch v.Type() {
+	case fastjson.TypeObject:
+		obj, err := v.Object()
+		if err != nil {
+			return nil, err
+		}
+		vv := obj.Get(field)
+		if vv == nil {
+			return nil, nil
+		}
+		switch vv.Type() {
+		case fastjson.TypeString:
+			return vv.String(), nil
+		case fastjson.TypeNumber:
+			return vv.Float64()
+		case fastjson.TypeTrue, fastjson.TypeFalse:
+			return vv.Bool()
+		}
+	}
+	return nil, nil
+}
+
 func (f *FastJsonConverter) decodeWithSchema(b []byte, schema map[string]*ast.JsonStreamField) (interface{}, error) {
 	var p fastjson.Parser
 	v, err := p.ParseBytes(b)
diff --git a/internal/io/http/httpserver/websocketConn.go b/internal/io/http/httpserver/websocketConn.go
index c106b974..603f911f 100644
--- a/internal/io/http/httpserver/websocketConn.go
+++ b/internal/io/http/httpserver/websocketConn.go
@@ -17,13 +17,22 @@ package httpserver
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 type WebsocketConnection struct {
 	RecvTopic string
 	SendTopic string
-	cfg       *connectionCfg
+	props     map[string]any
+	cfg       *wscConfig
+	isServer  bool
+	client    *WebsocketClient
+}
+
+type wscConfig struct {
+	Datasource string `json:"datasource"`
+	Addr       string `json:"addr"`
 }
 
 func (w *WebsocketConnection) Ping(ctx api.StreamContext) error {
@@ -31,10 +40,14 @@ func (w *WebsocketConnection) Ping(ctx api.StreamContext) error {
 }
 
 func (w *WebsocketConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
-	UnRegisterWebSocketEndpoint(w.cfg.Datasource)
 }
 
 func (w *WebsocketConnection) Close(ctx api.StreamContext) error {
+	if w.isServer {
+		UnRegisterWebSocketEndpoint(w.cfg.Datasource)
+	} else {
+		w.client.Close(ctx)
+	}
 	return nil
 }
 
@@ -43,17 +56,40 @@ func CreateWebsocketConnection(ctx api.StreamContext, props map[string]any) (mod
 }
 
 func createWebsocketServerConnection(ctx api.StreamContext, props map[string]any) (*WebsocketConnection, error) {
-	cfg := &connectionCfg{}
+	cfg := &wscConfig{}
 	if err := cast.MapToStruct(props, cfg); err != nil {
 		return nil, err
 	}
-	rTopic, sTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
-	if err != nil {
-		return nil, err
+	wc := &WebsocketConnection{
+		props:    props,
+		cfg:      cfg,
+		isServer: getWsType(cfg),
+	}
+	if wc.isServer {
+		rTopic, sTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
+		if err != nil {
+			return nil, err
+		}
+		wc.RecvTopic = rTopic
+		wc.SendTopic = sTopic
+	} else {
+		tlsConfig, err := cert.GenTLSConfig(props, "websocket")
+		if err != nil {
+			return nil, err
+		}
+		c := NewWebsocketClient(cfg.Addr, cfg.Datasource, tlsConfig)
+		if err := c.Connect(); err != nil {
+			return nil, err
+		}
+		wc.client = c
+		wc.RecvTopic, wc.SendTopic = c.Run(ctx)
+	}
+	return wc, nil
+}
+
+func getWsType(cfg *wscConfig) bool {
+	if len(cfg.Addr) < 1 {
+		return true
 	}
-	return &WebsocketConnection{
-		RecvTopic: rTopic,
-		SendTopic: sTopic,
-		cfg:       cfg,
-	}, nil
+	return false
 }
diff --git a/internal/io/http/httpserver/websocket_client.go b/internal/io/http/httpserver/websocket_client.go
new file mode 100644
index 00000000..668011ad
--- /dev/null
+++ b/internal/io/http/httpserver/websocket_client.go
@@ -0,0 +1,90 @@
+// Copyright 2024 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.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+)
+
+type WebsocketClient struct {
+	RecvTopic string
+	SendTopic string
+
+	addr      string
+	path      string
+	tlsConfig *tls.Config
+	conn      *websocket.Conn
+	wg        *sync.WaitGroup
+	cancel    context.CancelFunc
+}
+
+func NewWebsocketClient(addr, path string, tlsConfig *tls.Config) *WebsocketClient {
+	return &WebsocketClient{
+		addr:      addr,
+		path:      path,
+		tlsConfig: tlsConfig,
+		wg:        &sync.WaitGroup{},
+	}
+}
+
+func (c *WebsocketClient) Connect() error {
+	d := &websocket.Dialer{
+		HandshakeTimeout: 3 * time.Second,
+		TLSClientConfig:  c.tlsConfig,
+	}
+	if len(c.addr) < 1 {
+		return fmt.Errorf("addr should be defined")
+	}
+	u := url.URL{Scheme: "ws", Host: c.addr, Path: c.path}
+	conn, _, err := d.Dial(u.String(), nil)
+	if err != nil {
+		return err
+	}
+	c.conn = conn
+	return nil
+}
+
+func (c *WebsocketClient) Run(ctx api.StreamContext) (string, string) {
+	c.RecvTopic = recvTopic(c.path, false)
+	c.SendTopic = sendTopic(c.path, false)
+	pubsub.CreatePub(c.RecvTopic)
+	c.handleProcess(ctx)
+	return c.RecvTopic, c.SendTopic
+}
+
+func (c *WebsocketClient) handleProcess(parCtx api.StreamContext) {
+	ctx, cancel := parCtx.WithCancel()
+	c.cancel = cancel
+	c.wg.Add(2)
+	go recvProcess(ctx, c.RecvTopic, c.conn, cancel, c.wg)
+	go sendProcess(ctx, c.SendTopic, "", c.conn, cancel, c.wg)
+}
+
+func (c *WebsocketClient) Close(ctx api.StreamContext) error {
+	pubsub.RemovePub(c.RecvTopic)
+	c.cancel()
+	c.wg.Wait()
+	return nil
+}
diff --git a/internal/io/http/httpserver/websocket_server.go b/internal/io/http/httpserver/websocket_server.go
index fbef1d61..f03fd9a2 100644
--- a/internal/io/http/httpserver/websocket_server.go
+++ b/internal/io/http/httpserver/websocket_server.go
@@ -32,12 +32,20 @@ const (
 	WebsocketTopicPrefix = "$$websocket/"
 )
 
-func recvTopic(endpoint string) string {
-	return fmt.Sprintf("recv/%s/%s", WebsocketTopicPrefix, endpoint)
+func recvTopic(endpoint string, isServer bool) string {
+	if isServer {
+		return fmt.Sprintf("%s/server/recv/%s", WebsocketTopicPrefix, endpoint)
+	} else {
+		return fmt.Sprintf("%s/client/recv/%s", WebsocketTopicPrefix, endpoint)
+	}
 }
 
-func sendTopic(endpoint string) string {
-	return fmt.Sprintf("send/%s/%s", WebsocketTopicPrefix, endpoint)
+func sendTopic(endpoint string, isServer bool) string {
+	if isServer {
+		return fmt.Sprintf("%s/server/send/%s", WebsocketTopicPrefix, endpoint)
+	} else {
+		return fmt.Sprintf("%s/client/send/%s", WebsocketTopicPrefix, endpoint)
+	}
 }
 
 type websocketEndpointContext struct {
@@ -64,21 +72,17 @@ func (m *GlobalServerManager) handleProcess(ctx api.StreamContext, endpoint stri
 	}()
 	subWg := &sync.WaitGroup{}
 	subWg.Add(2)
-	go m.recvProcess(ctx, endpoint, c, cancel, subWg)
-	go m.sendProcess(ctx, endpoint, instanceID, c, cancel, subWg)
+	go recvProcess(ctx, recvTopic(endpoint, true), c, cancel, subWg)
+	go sendProcess(ctx, sendTopic(endpoint, true), fmt.Sprintf("ws/send/%v", instanceID), c, cancel, subWg)
 	subWg.Wait()
 }
 
-func (m *GlobalServerManager) sendProcess(ctx api.StreamContext, endpoint string, instanceID int, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
-	conf.Log.Infof("websocket endpoint %v start sendProcess", endpoint)
-	topic := sendTopic(endpoint)
-	sourceID := fmt.Sprintf("ws/send/%v", instanceID)
+func sendProcess(ctx api.StreamContext, topic, sourceID string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
 	defer func() {
 		pubsub.CloseSourceConsumerChannel(topic, sourceID)
 		cancel()
 		c.Close()
 		wg.Done()
-		conf.Log.Infof("websocket send endpoint %v stop sendProcess", endpoint)
 	}()
 	ch := pubsub.CreateSub(topic, nil, sourceID, 1024)
 	for {
@@ -89,24 +93,19 @@ func (m *GlobalServerManager) sendProcess(ctx api.StreamContext, endpoint string
 			data := d.([]byte)
 			if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
 				if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-					conf.Log.Infof("websocket endpoint %s connection get closed: %v", endpoint, err)
 					return
 				}
-				conf.Log.Warnf("websocket endpoint %v send data meet error: %v", endpoint, err)
 			}
 		}
 	}
 }
 
-func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
+func recvProcess(ctx api.StreamContext, topic string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
 	defer func() {
 		cancel()
 		c.Close()
 		wg.Done()
-		conf.Log.Infof("websocket recv endpoint %v stop recvProcess", endpoint)
 	}()
-	conf.Log.Infof("websocket endpoint %v start recvProcess", endpoint)
-	topic := recvTopic(endpoint)
 	for {
 		select {
 		case <-ctx.Done():
@@ -116,17 +115,14 @@ func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string
 		msgType, data, err := c.ReadMessage()
 		if err != nil {
 			if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-				conf.Log.Infof("websocket endpoint %s connection get closed: %v", endpoint, err)
 				return
 			}
-			conf.Log.Errorf("websocket endpoint %s recv error %s", endpoint, err)
 			continue
 		}
 		switch msgType {
 		case websocket.TextMessage:
 			pubsub.ProduceAny(ctx, topic, data)
 		default:
-			conf.Log.Infof("websocker endpoint %v recv other message typ %v", endpoint, msgType)
 		}
 	}
 }
@@ -135,8 +131,8 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 	conf.Log.Infof("websocket endpoint %v register", endpoint)
 	m.Lock()
 	defer m.Unlock()
-	rTopic := recvTopic(endpoint)
-	sTopic := sendTopic(endpoint)
+	rTopic := recvTopic(endpoint, true)
+	sTopic := sendTopic(endpoint, true)
 	pubsub.CreatePub(rTopic)
 	m.router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
 		c, err := m.upgrader.Upgrade(w, r, nil)
@@ -155,7 +151,7 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 
 func (m *GlobalServerManager) UnRegisterWebSocketEndpoint(endpoint string) *websocketEndpointContext {
 	conf.Log.Infof("websocket endpoint %v unregister", endpoint)
-	pubsub.RemovePub(recvTopic(endpoint))
+	pubsub.RemovePub(recvTopic(endpoint, true))
 	m.Lock()
 	defer m.Unlock()
 	wctx, ok := m.websocketEndpoint[endpoint]
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index 420fa0f6..fee87f69 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -151,7 +151,7 @@ func CreateClient(ctx api.StreamContext, props map[string]any) (*Connection, err
 	opts.OnReconnecting = con.onReconnecting
 	opts.ConnectRetry = true
 	opts.ConnectRetryInterval = connection.DefaultInitialInterval
-	opts.MaxReconnectInterval = connection.DefaultBackoffMaxElapsedDuration
+	opts.MaxReconnectInterval = connection.DefaultMaxInterval
 
 	cli := pahoMqtt.NewClient(opts)
 	token := cli.Connect()
diff --git a/internal/io/websocket/websocket_sink.go b/internal/io/websocket/websocket_sink.go
index 613a21d5..ad56d6ac 100644
--- a/internal/io/websocket/websocket_sink.go
+++ b/internal/io/websocket/websocket_sink.go
@@ -32,6 +32,7 @@ type WebsocketSink struct {
 }
 
 func (w *WebsocketSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	configs = solveProps(configs)
 	cfg := &WebsocketConfig{}
 	if err := cast.MapToStruct(configs, cfg); err != nil {
 		return err
diff --git a/internal/io/websocket/websocket_source.go b/internal/io/websocket/websocket_source.go
index 5dc6d2a1..8e06e27c 100644
--- a/internal/io/websocket/websocket_source.go
+++ b/internal/io/websocket/websocket_source.go
@@ -39,6 +39,7 @@ type WebsocketConfig struct {
 }
 
 func (w *WebsocketSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	configs = solveProps(configs)
 	cfg := &WebsocketConfig{}
 	if err := cast.MapToStruct(configs, cfg); err != nil {
 		return err
@@ -89,6 +90,15 @@ func (w *WebsocketSource) Subscribe(ctx api.StreamContext, ingest api.BytesInges
 	return nil
 }
 
+func solveProps(props map[string]any) map[string]any {
+	v1, ok1 := props["path"]
+	_, ok2 := props["datasource"]
+	if ok1 && !ok2 {
+		props["datasource"] = v1
+	}
+	return props
+}
+
 func GetSource() api.Source {
 	return &WebsocketSource{}
 }
diff --git a/internal/server/pprof_init.go b/internal/server/pprof_init.go
index 6d7ab503..87139a33 100644
--- a/internal/server/pprof_init.go
+++ b/internal/server/pprof_init.go
@@ -21,6 +21,10 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+
+	"github.com/Rookiecom/cpuprofile"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 )
 
 func init() {
@@ -33,6 +37,11 @@ type pprofComp struct {
 
 func (p pprofComp) serve() {
 	go func() {
+		if conf.Config.Basic.EnableResourceProfiling {
+			cpuprofile.WebProfile(":6060")
+			return
+		}
+
 		if err := http.ListenAndServe(":6060", nil); err != nil {
 			log.Fatal(err)
 		}
diff --git a/internal/server/server.go b/internal/server/server.go
index 34847546..e47ee266 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -28,6 +28,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/Rookiecom/cpuprofile"
 	"github.com/sirupsen/logrus"
 	"go.uber.org/automaxprocs/maxprocs"
 
@@ -139,6 +140,11 @@ func StartUp(Version string) {
 	createPaths()
 	conf.InitConf()
 
+	serverCtx, serverCancel := context.WithCancel(context.Background())
+	if conf.Config.Basic.EnableResourceProfiling {
+		startCPUProfiling(serverCtx)
+	}
+
 	undo, _ := maxprocs.Set(maxprocs.Logger(conf.Log.Infof))
 	defer undo()
 
@@ -185,7 +191,8 @@ func StartUp(Version string) {
 		panic(err)
 	}
 	connection.InitConnectionManager()
-	if err := connection.ReloadConnection(); err != nil {
+	// reload with 3s timeout
+	if err := connection.ReloadConnection(3 * time.Second); err != nil {
 		conf.Log.Warn(err)
 	}
 	meta.Bind()
@@ -213,8 +220,7 @@ func StartUp(Version string) {
 			}
 		}
 	}
-	exit := make(chan struct{})
-	go runScheduleRuleChecker(exit)
+	go runScheduleRuleChecker(serverCtx)
 
 	// Start rest service
 	srvRest := createRestServer(conf.Config.Basic.RestIp, conf.Config.Basic.RestPort, conf.Config.Basic.Authentication)
@@ -267,7 +273,7 @@ func StartUp(Version string) {
 		time.Sleep(time.Second)
 		conf.Log.Info("eKuiper stopped by Stop request")
 	}
-	exit <- struct{}{}
+	serverCancel()
 	// wait rule checker exit
 	time.Sleep(10 * time.Millisecond)
 
@@ -368,7 +374,7 @@ func resetAllStreams() error {
 	return nil
 }
 
-func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
+func runScheduleRuleCheckerByInterval(d time.Duration, ctx context.Context) {
 	conf.Log.Infof("start patroling schedule rule state")
 	ticker := time.NewTicker(d)
 	defer func() {
@@ -377,7 +383,7 @@ func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
 	}()
 	for {
 		select {
-		case <-exit:
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
 			rs, err := getAllRulesWithState()
@@ -392,8 +398,8 @@ func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
 	}
 }
 
-func runScheduleRuleChecker(exit <-chan struct{}) {
-	runScheduleRuleCheckerByInterval(time.Duration(conf.Config.Basic.RulePatrolInterval), exit)
+func runScheduleRuleChecker(ctx context.Context) {
+	runScheduleRuleCheckerByInterval(time.Duration(conf.Config.Basic.RulePatrolInterval), ctx)
 }
 
 type RuleStatusMetricsValue int
@@ -474,6 +480,26 @@ func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAc
 	return scheduleRuleActionDoNothing
 }
 
+func startCPUProfiling(ctx context.Context) error {
+	if err := cpuprofile.StartProfilerAndAggregater(ctx, time.Duration(1000)*time.Millisecond); err != nil {
+		return err
+	}
+	receiveChan := make(chan *cpuprofile.DataSetAggregate, 1024)
+	cpuprofile.RegisterTag("rule", receiveChan)
+	go func(ctx context.Context) {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case data := <-receiveChan:
+				// TODO: support query in future
+				conf.Log.Debugf("cpu profile data: %v", data)
+			}
+		}
+	}(ctx)
+	return nil
+}
+
 func waitAllRuleStop(ctx context.Context) error {
 	rules, _ := ruleProcessor.GetAllRules()
 	for _, r := range rules {
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index fdafba69..cb8afee8 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"runtime/pprof"
 	"sync"
 	"text/template"
 	"time"
@@ -55,6 +56,18 @@ type DefaultContext struct {
 	opsWg *sync.WaitGroup
 }
 
+func RuleBackground(ruleName string) *DefaultContext {
+	if !conf.Config.Basic.EnableResourceProfiling {
+		return Background()
+	}
+	ctx := pprof.WithLabels(context.Background(), pprof.Labels("rule", ruleName))
+	pprof.SetGoroutineLabels(ctx)
+	c := &DefaultContext{
+		ctx: ctx,
+	}
+	return c
+}
+
 func Background() *DefaultContext {
 	c := &DefaultContext{
 		ctx: context.Background(),
@@ -62,6 +75,13 @@ func Background() *DefaultContext {
 	return c
 }
 
+func WithContext(ctx context.Context) *DefaultContext {
+	c := &DefaultContext{
+		ctx: ctx,
+	}
+	return c
+}
+
 func WithValue(parent *DefaultContext, key, val interface{}) *DefaultContext {
 	parent.ctx = context.WithValue(parent.ctx, key, val)
 	return parent
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 8857ab75..019c7fd0 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -224,7 +224,7 @@ func (s *Topo) prepareContext() {
 				}
 			}
 		}
-		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+		ctx := kctx.WithValue(kctx.RuleBackground(s.name), kctx.LoggerKey, contextLogger)
 		ctx = kctx.WithValue(ctx, kctx.RuleStartKey, timex.GetNowInMilli())
 		ctx = kctx.WithWg(ctx, s.opsWg)
 		s.ctx, s.cancel = ctx.WithCancel()
diff --git a/pkg/connection/connection_pool.go b/pkg/connection/connection_pool.go
index 59c33688..56a073f3 100644
--- a/pkg/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -15,6 +15,7 @@
 package connection
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -26,7 +27,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	topoContext "github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -236,6 +237,11 @@ func createNamedConnection(ctx api.StreamContext, meta *ConnectionMeta) (modules
 		return nil, fmt.Errorf("unknown connection type")
 	}
 	err = backoff.Retry(func() error {
+		select {
+		case <-ctx.Done():
+			return backoff.Permanent(errors.New("timeout"))
+		default:
+		}
 		conn, err = connRegister(ctx, meta.Props)
 		failpoint.Inject("createConnectionErr", func() {
 			if mockErr {
@@ -299,11 +305,19 @@ func InitConnectionManager() {
 	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
 }
 
-func ReloadConnection() error {
+func ReloadConnection(timeout time.Duration) error {
 	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
 	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer func() {
+		cancel()
+	}()
+	failpoint.Inject("reloadTimeout", func() {
+		time.Sleep(100 * time.Millisecond)
+	})
+
 	for key, props := range cfgs {
 		names := strings.Split(key, ".")
 		if len(names) != 3 {
@@ -316,7 +330,7 @@ func ReloadConnection() error {
 			Typ:   typ,
 			Props: props,
 		}
-		conn, err := createNamedConnection(context.Background(), meta)
+		conn, err := createNamedConnection(topoContext.WithContext(ctx), meta)
 		if err != nil {
 			conf.Log.Warnf("initialize connection:%v failed, err:%v", id, err)
 			globalConnectionManager.failConnection[id] = err.Error()


Test patch:
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 3d012381..5f0bcec5 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 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.
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
@@ -708,3 +709,61 @@ func TestIssue(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, m, 0)
 }
+
+func TestDecodeField(t *testing.T) {
+	testcases := []struct {
+		name    string
+		payload []byte
+		result  any
+		err     string
+	}{
+		{
+			name:    "normal",
+			payload: []byte(`{"id":1, "value":"vv"}`),
+			result:  1.0,
+		},
+		{
+			name:    "empty",
+			payload: []byte(`{"a":[true]}`),
+			result:  nil,
+		},
+		{
+			name:    "composite",
+			payload: []byte(`{"id":[true]}`),
+			result:  nil,
+		},
+		{
+			name:    "invalid",
+			payload: []byte(`{"a":1`),
+			err:     "cannot parse JSON: cannot parse object: unexpected end of object; unparsed tail: \"\"",
+		},
+		{
+			name:    "string",
+			payload: []byte(`{"id":"1", "value":"vv"}`),
+			result:  `"1"`,
+		},
+		{
+			name:    "bool",
+			payload: []byte(`{"id":false, "value":"vv"}`),
+			result:  false,
+		},
+		{
+			name:    "not obj",
+			payload: []byte(`[{"id":false, "value":"vv"}]`),
+			result:  nil,
+		},
+	}
+	ctx := mockContext.NewMockContext("test", "op1")
+	f := NewFastJsonConverter(nil)
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			field, err := f.DecodeField(ctx, tc.payload, "id")
+			if tc.err != "" {
+				assert.EqualError(t, err, tc.err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tc.result, field)
+			}
+		})
+	}
+}
diff --git a/internal/io/http/httpserver/websocketConn_test.go b/internal/io/http/httpserver/websocketConn_test.go
index d0eebf42..3daff97e 100644
--- a/internal/io/http/httpserver/websocketConn_test.go
+++ b/internal/io/http/httpserver/websocketConn_test.go
@@ -15,8 +15,14 @@
 package httpserver
 
 import (
+	"context"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"sync"
 	"testing"
 
+	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/require"
 
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
@@ -35,5 +41,95 @@ func TestWebsocketConn(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, conn.Ping(ctx))
 	conn.DetachSub(ctx, props)
+	require.NoError(t, conn.Close(ctx))
+}
+
+func TestWebsocketClientConn(t *testing.T) {
+	tc := newTC()
+	s := createWServer(tc)
+	defer func() {
+		s.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	props := map[string]any{
+		"datasource": "/ws",
+		"addr":       s.URL[len("http://"):],
+	}
+	conn, err := createWebsocketServerConnection(ctx, props)
+	require.NoError(t, err)
 	require.NoError(t, conn.Ping(ctx))
+	conn.DetachSub(ctx, props)
+	require.NoError(t, conn.Close(ctx))
+}
+
+func newTC() *testcase {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &testcase{
+		ctx:    ctx,
+		cancel: cancel,
+		recvCh: make(chan []byte, 10),
+		sendCh: make(chan []byte, 10),
+	}
+}
+
+type testcase struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	recvCh chan []byte
+	sendCh chan []byte
+}
+
+func createWServer(tc *testcase) *httptest.Server {
+	router := http.NewServeMux()
+	router.HandleFunc("/ws", tc.handler)
+	server := httptest.NewServer(router)
+	return server
+}
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  256,
+	WriteBufferSize: 256,
+	WriteBufferPool: &sync.Pool{},
+}
+
+func (tc *testcase) recvProcess(c *websocket.Conn) {
+	defer func() {
+		tc.cancel()
+		c.Close()
+	}()
+	for {
+		_, message, err := c.ReadMessage()
+		if err != nil {
+			return
+		}
+		tc.recvCh <- message
+	}
+}
+
+func (tc *testcase) sendProcess(c *websocket.Conn) {
+	defer func() {
+		tc.cancel()
+		c.Close()
+	}()
+	for {
+		select {
+		case <-tc.ctx.Done():
+			return
+		case x := <-tc.sendCh:
+			err := c.WriteMessage(websocket.TextMessage, x)
+			if err != nil {
+				return
+			}
+		}
+	}
+}
+
+func (tc *testcase) handler(w http.ResponseWriter, r *http.Request) {
+	c, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Print("upgrade:", err)
+		return
+	}
+	go tc.recvProcess(c)
+	go tc.sendProcess(c)
 }
diff --git a/internal/io/http/httpserver/websocket_client_test.go b/internal/io/http/httpserver/websocket_client_test.go
new file mode 100644
index 00000000..3be51159
--- /dev/null
+++ b/internal/io/http/httpserver/websocket_client_test.go
@@ -0,0 +1,53 @@
+// Copyright 2024 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.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestWebsocketClient(t *testing.T) {
+	tc := newTC()
+	s := createWServer(tc)
+	defer func() {
+		s.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	wc := NewWebsocketClient(s.URL[len("http://"):], "/ws", nil)
+	require.NoError(t, wc.Connect())
+	rt, st := wc.Run(ctx)
+	pubsub.CreatePub(st)
+	defer func() {
+		pubsub.RemovePub(st)
+	}()
+	// wait process start
+	time.Sleep(100 * time.Millisecond)
+	data := []byte("123")
+	pubsub.ProduceAny(ctx, st, data)
+	require.Equal(t, data, <-tc.recvCh)
+	ch := pubsub.CreateSub(rt, nil, "", 1024)
+	defer func() {
+		pubsub.CloseSourceConsumerChannel(rt, "")
+	}()
+	tc.sendCh <- data
+	require.Equal(t, data, <-ch)
+	require.NoError(t, wc.Close(ctx))
+}
diff --git a/internal/io/http/httpserver/websocket_server_test.go b/internal/io/http/httpserver/websocket_server_test.go
index 02f47afd..02c9f97a 100644
--- a/internal/io/http/httpserver/websocket_server_test.go
+++ b/internal/io/http/httpserver/websocket_server_test.go
@@ -84,7 +84,7 @@ func TestWebsocketServerRecvDataOther(t *testing.T) {
 
 func TestWebsocketServerSendData(t *testing.T) {
 	endpoint := "/e1"
-	topic := sendTopic(endpoint)
+	topic := sendTopic(endpoint, true)
 	pubsub.CreatePub(topic)
 	ip := "127.0.0.1"
 	port := 10085
diff --git a/internal/server/server_test.go b/internal/server/server_test.go
index b199b214..d6608465 100644
--- a/internal/server/server_test.go
+++ b/internal/server/server_test.go
@@ -15,6 +15,7 @@
 package server
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -86,10 +87,10 @@ func TestHandleScheduleRule(t *testing.T) {
 }
 
 func TestRunScheduleRuleChecker(t *testing.T) {
-	exit := make(chan struct{})
-	go runScheduleRuleCheckerByInterval(3*time.Second, exit)
+	ctx, cancel := context.WithCancel(context.Background())
+	go runScheduleRuleCheckerByInterval(3*time.Second, ctx)
 	time.Sleep(1 * time.Second)
-	exit <- struct{}{}
+	cancel()
 }
 
 func TestHandleScheduleRuleState(t *testing.T) {
@@ -121,3 +122,13 @@ func TestHandleScheduleRuleState(t *testing.T) {
 	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleStarted))
 	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleWait))
 }
+
+func TestStartCPUProfiling(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	err := startCPUProfiling(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(5 * time.Second)
+	cancel()
+}
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index b9d37667..40390756 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -15,13 +15,17 @@
 package context
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
 	"path"
 	"reflect"
+	"runtime/pprof"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
@@ -228,3 +232,11 @@ func TestParseTemplate(t *testing.T) {
 		}
 	}
 }
+
+func TestRuleBackground(t *testing.T) {
+	conf.InitConf()
+	conf.Config.Basic.EnableResourceProfiling = true
+	c := RuleBackground("test")
+	ctx := pprof.WithLabels(context.Background(), pprof.Labels("rule", "test"))
+	assert.Equal(t, c.ctx, ctx)
+}
diff --git a/pkg/connection/connection_pool_test.go b/pkg/connection/connection_pool_test.go
index b4c199f5..341573e5 100644
--- a/pkg/connection/connection_pool_test.go
+++ b/pkg/connection/connection_pool_test.go
@@ -16,6 +16,7 @@ package connection
 
 import (
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
@@ -114,7 +115,7 @@ func TestConnectionStatus(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
 	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
-	require.NoError(t, ReloadConnection())
+	require.NoError(t, ReloadConnection(999*time.Second))
 	ctx := context.Background()
 	allStatus := GetAllConnectionStatus(ctx)
 	s, ok := allStatus["a1"]
@@ -130,6 +131,16 @@ func TestConnectionStatus(t *testing.T) {
 	}, s)
 }
 
+func TestReloadConnectionErr(t *testing.T) {
+	require.NoError(t, InitConnectionManager4Test())
+	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
+	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/reloadTimeout", "return(true)")
+	require.NoError(t, ReloadConnection(time.Microsecond))
+	require.True(t, len(globalConnectionManager.failConnection) > 0)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/reloadTimeout")
+}
+
 func TestNonStoredConnection(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	ctx := mockContext.NewMockContext("id", "2")