feat: support neuron/kafka ping

extensions/sinks/kafka/ext/kafka.go

@@ -52,6 +52,19 @@ type kafkaConf struct {
 	Headers      interface{} `json:"headers"`
 }
 
+func (m *kafkaSink) Ping(_ string, props map[string]interface{}) error {
+	if err := m.Configure(props); err != nil {
+		return err
+	}
+	for _, broker := range strings.Split(m.c.Brokers, ",") {
+		err := ping(m.tlsConfig, broker)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (m *kafkaSink) Configure(props map[string]interface{}) error {
 	c := &sinkConf{
 		Brokers: "localhost:9092",
@@ -279,3 +292,13 @@ func (m *kafkaSink) parseHeaders(ctx api.StreamContext, data interface{}) ([]kaf
 	}
 	return nil, nil
 }
+
+func ping(tlsConfig *tls.Config, address string) error {
+	d := &kafkago.Dialer{TLS: tlsConfig}
+	c, err := d.Dial("tcp", address)
+	if err != nil {
+		return err
+	}
+	defer c.Close()
+	return nil
+}

extensions/sources/kafka/ext/kafka.go

@@ -33,6 +33,7 @@ type KafkaSource struct {
 	reader    *kafkago.Reader
 	offset    int64
 	tlsConfig *tls.Config
+	sc        *kafkaSourceConf
 }
 
 type kafkaSourceConf struct {
@@ -44,6 +45,19 @@ type kafkaSourceConf struct {
 	MaxBytes    int    `json:"maxBytes"`
 }
 
+func (s *KafkaSource) Ping(d string, props map[string]interface{}) error {
+	if err := s.Configure(d, props); err != nil {
+		return err
+	}
+	for _, broker := range strings.Split(s.sc.Brokers, ",") {
+		err := ping(s.tlsConfig, broker)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *kafkaSourceConf) validate() error {
 	if len(strings.Split(c.Brokers, ",")) == 0 {
 		return fmt.Errorf("brokers can not be empty")
@@ -116,6 +130,7 @@ func (s *KafkaSource) Configure(topic string, props map[string]interface{}) erro
 	}
 	reader := kafkago.NewReader(readerConfig)
 	s.reader = reader
+	s.sc = kConf
 	if err := s.reader.SetOffset(kafkago.LastOffset); err != nil {
 		return err
 	}
@@ -183,3 +198,13 @@ func (s *KafkaSource) GetOffset() (interface{}, error) {
 func GetSource() api.Source {
 	return &KafkaSource{}
 }
+
+func ping(tlsConfig *tls.Config, address string) error {
+	d := &kafkago.Dialer{TLS: tlsConfig}
+	c, err := d.Dial("tcp", address)
+	if err != nil {
+		return err
+	}
+	defer c.Close()
+	return nil
+}

internal/io/neuron/sink.go

@@ -16,13 +16,11 @@ package neuron
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
+	"net/url"
 	"sort"
-	"sync/atomic"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -53,18 +51,11 @@ func (s *sink) Ping(_ string, props map[string]interface{}) error {
 	if err := s.Configure(props); err != nil {
 		return err
 	}
-	ctx := context.Background()
-	cli, err := createOrGetConnection(ctx, s.c.Url)
+	u, err := url.Parse(s.c.Url)
 	if err != nil {
 		return err
 	}
-	defer func() {
-		closeConnection(ctx, s.c.Url)
-	}()
-	if atomic.LoadInt32(&cli.opened) == 1 {
-		return nil
-	}
-	return errors.New("neuron sink ping failed")
+	return ping(u)
 }
 
 func (s *sink) Configure(props map[string]interface{}) error {

internal/io/neuron/sink_test.go

@@ -19,6 +19,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 )
@@ -118,3 +120,20 @@ func sinkConnExpTest(t *testing.T) {
 		t.Errorf("error mismatch:\n\nexp=%s\n\ngot=%s\n\n", expErrStr, err.Error())
 	}
 }
+
+func TestSinkPing(t *testing.T) {
+	config := map[string]interface{}{
+		"url": "non-tcp",
+		"raw": true,
+	}
+	s := &sink{}
+	err := s.Ping("", config)
+	require.Error(t, err)
+	config = map[string]interface{}{
+		"url": "tcp://127.0.0.1:4313",
+		"raw": true,
+	}
+	s = &sink{}
+	err = s.Ping("", config)
+	require.Error(t, err)
+}

internal/io/neuron/source.go

@@ -15,14 +15,14 @@
 package neuron
 
 import (
-	"errors"
 	"fmt"
-	"sync/atomic"
+	"net/http"
+	"net/url"
 
 	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
-	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
@@ -39,16 +39,11 @@
 	if err := s.Configure(dataSource, props); err != nil {
 		return err
 	}
-	ctx := context.Background()
-	cli, err := createOrGetConnection(ctx, s.c.Url)
+	u, err := url.Parse(s.c.Url)
 	if err != nil {
 		return err
 	}
-	defer closeConnection(ctx, s.c.Url)
-	if atomic.LoadInt32(&cli.opened) == 1 {
-		return nil
-	}
-	return errors.New("neuron source ping failed")
+	return ping(u)
 }
 
 func (s *source) Configure(_ string, props map[string]interface{}) error {
@@ -94,3 +89,18 @@
 func GetSource() *source {
 	return &source{}
 }
+
+func ping(u *url.URL) error {
+	if u.Scheme == "tcp" {
+		r, err := http.Get(fmt.Sprintf("http://%v/api/v2/ping", u.Host))
+		if err != nil {
+			return err
+		}
+		if r.StatusCode == http.StatusOK {
+			return nil
+		}
+		return fmt.Errorf("neuron ping failed, code:%v", r.StatusCode)
+	}
+
+	return errorx.New("only tcp neuron url support ping")
+}

internal/io/neuron/source_test.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/benbjohnson/clock"
+	"github.com/stretchr/testify/require"
 	_ "go.nanomsg.org/mangos/v3/transport/ipc"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -75,3 +76,20 @@ func connectFailTest(t *testing.T) {
 	server.Close()
 	time.Sleep(1 * time.Second)
 }
+
+func TestSourcePing(t *testing.T) {
+	config := map[string]interface{}{
+		"url": "non-tcp",
+		"raw": true,
+	}
+	s := &source{}
+	err := s.Ping("", config)
+	require.Error(t, err)
+	config = map[string]interface{}{
+		"url": "tcp://127.0.0.1:4313",
+		"raw": true,
+	}
+	s = &source{}
+	err = s.Ping("", config)
+	require.Error(t, err)
+}

internal/topo/node/source_connector_node_test.go

@@ -192,7 +192,7 @@ func TestSubError(t *testing.T) {
 	var errResult error
 	go func() {
 		defer wg.Done()
-		ticker := time.After(2 * time.Second)
+		ticker := time.After(5 * time.Second)
 		for {
 			select {
 			case sg := <-errCh: