Repository: lf-edge/ekuiper
Pull request #3050 (2024-07-24 08:15:16)

chore: merge 1.14 influx changes

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `21.87500%` with `50 lines` in your changes missing coverage. Please review.
> Project coverage is 67.83%. Comparing base [(`167232c`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/167232c279c6f5dcbdeecd6b57823a22de459e06?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`12ea4d0`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/12ea4d0bc8532719b0846c97349c7996a368f7e1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [extensions/impl/influx/influx.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&filepath=extensions%2Fimpl%2Finflux%2Finflux.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL2luZmx1eC9pbmZsdXguZ28=) | 0.00% | [20 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/impl/influx2/influx2.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&filepath=extensions%2Fimpl%2Finflux2%2Finflux2.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL2luZmx1eDIvaW5mbHV4Mi5nbw==) | 12.50% | [14 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvc2luay5nbw==) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/lookup.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2Flookup.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvbG9va3VwLmdv) | 84.62% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3050      +/-   ##
==========================================
- Coverage   67.89%   67.83%   -0.07%     
==========================================
  Files         388      388              
  Lines       36980    37042      +62     
==========================================
+ Hits        25107    25125      +18     
- Misses       9603     9646      +43     
- Partials     2270     2271       +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3050?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/extensions/impl/influx/influx.go b/extensions/impl/influx/influx.go
index 8bc6420e..934d7dcd 100644
--- a/extensions/impl/influx/influx.go
+++ b/extensions/impl/influx/influx.go
@@ -114,6 +114,34 @@ func (m *influxSink) Connect(ctx api.StreamContext) (err error) {
 	return err
 }
 
+func (m *influxSink) Ping(_ string, props map[string]interface{}) (err error) {
+	if err = m.Provision(nil, props); err != nil {
+		return err
+	}
+	var insecureSkip bool
+	if m.tlsconf != nil {
+		insecureSkip = m.tlsconf.InsecureSkipVerify
+	}
+	m.cli, err = client.NewHTTPClient(client.HTTPConfig{
+		Addr:               m.conf.Addr,
+		Username:           m.conf.Username,
+		Password:           m.conf.Password,
+		InsecureSkipVerify: insecureSkip,
+		TLSConfig:          m.tlsconf,
+	})
+	if err != nil {
+		return fmt.Errorf("error creating influx client: %s", err)
+	}
+	defer func() {
+		if m.cli != nil {
+			m.cli.Close()
+		}
+	}()
+	// Test connection. Put it here to avoid server connection when running test in Configure
+	_, _, err = m.cli.Ping(time.Second * 10)
+	return err
+}
+
 func (m *influxSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
 	return m.collect(ctx, item.ToMap())
 }
diff --git a/extensions/impl/influx2/influx2.go b/extensions/impl/influx2/influx2.go
index c1337b37..49e5a66b 100644
--- a/extensions/impl/influx2/influx2.go
+++ b/extensions/impl/influx2/influx2.go
@@ -15,6 +15,7 @@
 package influx2
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"strings"
@@ -57,7 +58,28 @@ type influxSink2 struct {
 	cli client.Client
 }
 
-func (m *influxSink2) Provision(ctx api.StreamContext, props map[string]any) error {
+func (m *influxSink2) Ping(_ string, props map[string]interface{}) error {
+	if err := m.Provision(nil, props); err != nil {
+		return err
+	}
+	options := client.DefaultOptions().SetPrecision(m.conf.Precision).SetBatchSize(uint(m.conf.BatchSize))
+	if m.tlsconf != nil {
+		options = options.SetTLSConfig(m.tlsconf)
+	}
+	m.cli = client.NewClientWithOptions(m.conf.Addr, m.conf.Token, options)
+	defer func() {
+		if m.cli != nil {
+			m.cli.Close()
+		}
+	}()
+	pingable, err := m.cli.Ping(context.Background())
+	if err != nil || !pingable {
+		return fmt.Errorf("error connecting to influxdb2: %v", err)
+	}
+	return nil
+}
+
+func (m *influxSink2) Provision(_ api.StreamContext, props map[string]any) error {
 	m.conf = c{
 		PrecisionStr: "ms",
 		WriteOptions: tspoint.WriteOptions{
@@ -89,7 +111,7 @@ func (m *influxSink2) Provision(ctx api.StreamContext, props map[string]any) err
 	default:
 		return fmt.Errorf("precision %s is not supported", m.conf.PrecisionStr)
 	}
-	if len(m.conf.Measurement) == 0 {
+	if len(m.conf.Measurement) == 0 && !m.conf.UseLineProtocol {
 		return fmt.Errorf("measurement is required")
 	}
 	err = cast.MapToStruct(props, &m.conf.WriteOptions)
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index b2c8be2c..71974bcc 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -51,7 +51,7 @@ func init() {
 	modules.RegisterSink("websocket", func() api.Sink { return websocket.GetSink() })
 
 	modules.RegisterLookupSource("memory", memory.GetLookupSource)
-	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
+	modules.RegisterLookupSource("httppull", http.GetLookUpSource)
 
 	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
 	modules.RegisterConnection("nng", nng.CreateConnection)
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index f42cbefd..3e21dea5 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -60,20 +60,28 @@ func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]a
 }
 
 func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, error) {
-	headers, err := hps.parseHeaders(ctx, hps.tokens)
+	result, latestMD5, err := doPull(ctx, hps.ClientConf, hps.lastMD5)
 	if err != nil {
 		return nil, err
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, headers, true, []byte(hps.config.Body))
+	hps.lastMD5 = latestMD5
+	return result, nil
+}
+
+func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]any, string, error) {
+	headers, err := c.parseHeaders(ctx, c.tokens)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
-	results, newMD5, err := hps.parseResponse(ctx, resp, hps.lastMD5)
+	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, true, []byte(c.config.Body))
 	if err != nil {
-		return nil, err
+		return nil, "", err
+	}
+	results, newMD5, err := c.parseResponse(ctx, resp, lastMD5)
+	if err != nil {
+		return nil, "", err
 	}
-	hps.lastMD5 = newMD5
-	return results, nil
+	return results, newMD5, nil
 }
 
 func GetSource() api.Source {
diff --git a/internal/io/http/lookup_source.go b/internal/io/http/lookup_source.go
new file mode 100644
index 00000000..d4f5ddd7
--- /dev/null
+++ b/internal/io/http/lookup_source.go
@@ -0,0 +1,96 @@
+// 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 http
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type HttpLookupSource struct {
+	*ClientConf
+}
+
+func (hls *HttpLookupSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	pc := &pullSourceConfig{}
+	if err := cast.MapToStruct(configs, pc); err != nil {
+		return err
+	}
+	if hls.ClientConf == nil {
+		hls.ClientConf = &ClientConf{}
+	}
+	return hls.InitConf(pc.Path, configs)
+}
+
+func (hls *HttpLookupSource) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (hls *HttpLookupSource) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (hls *HttpLookupSource) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error) {
+	resps, _, err := doPull(ctx, hls.ClientConf, "")
+	if err != nil {
+		return nil, err
+	}
+	resps = pruneData(resps, fields)
+	return lookupJoin(resps, keys, values), nil
+}
+
+func pruneData(data []map[string]any, fields []string) []map[string]any {
+	for index, row := range data {
+		for key := range row {
+			if !findColumn(key, fields) {
+				delete(row, key)
+			}
+		}
+		data[index] = row
+	}
+	return data
+}
+
+func findColumn(column string, fields []string) bool {
+	for _, field := range fields {
+		if field == column {
+			return true
+		}
+	}
+	return false
+}
+
+func lookupJoin(dataMap []map[string]interface{}, keys []string, values []interface{}) []map[string]any {
+	var resps []map[string]interface{}
+	for _, resp := range dataMap {
+		match := true
+		for i, k := range keys {
+			if val, ok := resp[k]; !ok || val != values[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			resps = append(resps, resp)
+		}
+	}
+	return resps
+}
+
+func GetLookUpSource() api.Source {
+	return &HttpLookupSource{}
+}
+
+var _ api.LookupSource = &HttpLookupSource{}
diff --git a/internal/io/redis/lookup.go b/internal/io/redis/lookup.go
index 02b30a2b..0e9660a6 100644
--- a/internal/io/redis/lookup.go
+++ b/internal/io/redis/lookup.go
@@ -15,6 +15,7 @@
 package redis
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -42,6 +43,22 @@ type lookupSource struct {
 	cli *redis.Client
 }
 
+func (s *lookupSource) Ping(dataSource string, props map[string]interface{}) error {
+	err := s.Validate(props)
+	if err != nil {
+		return err
+	}
+	s.cli = redis.NewClient(&redis.Options{
+		Addr:     s.c.Addr,
+		Username: s.c.Username,
+		Password: s.c.Password,
+		DB:       s.db, // use default DB
+	})
+	defer s.cli.Close()
+	_, err = s.cli.Ping(context.Background()).Result()
+	return err
+}
+
 func (s *lookupSource) Provision(ctx api.StreamContext, props map[string]any) error {
 	return s.Validate(props)
 }
diff --git a/internal/io/redis/sink.go b/internal/io/redis/sink.go
index bff42352..1322a1eb 100644
--- a/internal/io/redis/sink.go
+++ b/internal/io/redis/sink.go
@@ -15,6 +15,7 @@
 package redis
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -93,6 +94,23 @@ func (r *RedisSink) Validate(props map[string]any) error {
 	return nil
 }
 
+func (r *RedisSink) Ping(dataSource string, props map[string]interface{}) error {
+	if err := r.Validate(props); err != nil {
+		return err
+	}
+	cli := redis.NewClient(&redis.Options{
+		Addr:     r.c.Addr,
+		Username: r.c.Username,
+		Password: r.c.Password,
+		DB:       r.c.Db, // use default DB
+	})
+	_, err := cli.Ping(context.Background()).Result()
+	defer func() {
+		cli.Close()
+	}()
+	return err
+}
+
 func (r *RedisSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
 	return r.save(ctx, item.ToMap())
 }
diff --git a/internal/server/rpc.go b/internal/server/rpc.go
index 2d229868..84f7eef5 100644
--- a/internal/server/rpc.go
+++ b/internal/server/rpc.go
@@ -296,6 +296,7 @@ func (t *Server) Export(file string, reply *string) error {
 	if err != nil {
 		return err
 	}
+	defer f.Close()
 	exported, counts, err := rulesetProcessor.Export()
 	if err != nil {
 		return err


Test patch:
diff --git a/extensions/impl/influx2/influx2_test.go b/extensions/impl/influx2/influx2_test.go
index d059baa5..795232af 100644
--- a/extensions/impl/influx2/influx2_test.go
+++ b/extensions/impl/influx2/influx2_test.go
@@ -15,6 +15,7 @@
 package influx2
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
@@ -104,13 +105,36 @@ func TestConfig(t *testing.T) {
 			},
 			error: "precision abc is not supported",
 		},
+		{
+			name: "no error",
+			conf: map[string]interface{}{
+				"addr":            "http://192.168.0.3:8086",
+				"org":             "abc",
+				"bucket":          "bucket_one",
+				"precision":       "ns",
+				"useLineProtocol": true,
+			},
+			expected: c{
+				Addr:            "http://192.168.0.3:8086",
+				Org:             "abc",
+				Bucket:          "bucket_one",
+				PrecisionStr:    "ns",
+				Precision:       time.Nanosecond,
+				UseLineProtocol: true,
+				WriteOptions: tspoint.WriteOptions{
+					PrecisionStr: "ns",
+				},
+				BatchSize: 1,
+			},
+		},
 		{
 			name: "measurement missing error",
 			conf: map[string]interface{}{
-				"addr":      "http://192.168.0.3:8086",
-				"org":       "abc",
-				"bucket":    "bucket_one",
-				"precision": "ns",
+				"addr":            "http://192.168.0.3:8086",
+				"org":             "abc",
+				"bucket":          "bucket_one",
+				"precision":       "ns",
+				"useLineProtocol": false,
 			},
 			error: "measurement is required",
 		},
@@ -428,10 +452,11 @@ func TestCollectPointsError(t *testing.T) {
 func TestCollectLines(t *testing.T) {
 	timex.Set(10)
 	tests := []struct {
-		name   string
-		conf   c
-		data   any
-		result []string
+		name    string
+		conf    c
+		data    any
+		result  []string
+		result2 []string
 	}{
 		{
 			name: "normal",
@@ -507,7 +532,8 @@ func TestCollectLines(t *testing.T) {
 				"humidity": 50,
 				"ts":       100,
 			},
-			result: []string{"test5,tag2=50 humidity=50,ts=100 100"},
+			result:  []string{"test5,tag2=50 humidity=50,ts=100 100"},
+			result2: []string{"test5,tag2=50 ts=100,humidity=50 100"},
 		},
 	}
 
@@ -519,7 +545,11 @@ func TestCollectLines(t *testing.T) {
 			ctx := mockContext.NewMockContext(test.name, "op")
 			lines, err := ifsink.transformLines(ctx, test.data)
 			assert.NoError(t, err)
-			assert.Equal(t, test.result, lines)
+			if test.result2 == nil {
+				assert.Equal(t, test.result, lines)
+			} else {
+				assert.True(t, reflect.DeepEqual(test.result, lines) || reflect.DeepEqual(test.result2, lines))
+			}
 		})
 	}
 }
diff --git a/internal/io/http/lookup_source_test.go b/internal/io/http/lookup_source_test.go
new file mode 100644
index 00000000..c0f86ea1
--- /dev/null
+++ b/internal/io/http/lookup_source_test.go
@@ -0,0 +1,46 @@
+// 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 http
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestLookup(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	hls := &HttpLookupSource{}
+	require.NoError(t, hls.Provision(ctx, map[string]any{
+		"url":        server.URL,
+		"datasource": "/get",
+		"method":     "get",
+	}))
+	require.NoError(t, hls.Connect(ctx))
+	got, err := hls.Lookup(ctx, []string{"code"}, []string{"code"}, []any{float64(200)})
+	require.NoError(t, err)
+	require.Equal(t, []map[string]any{
+		{
+			"code": float64(200),
+		},
+	}, got)
+	require.NoError(t, hls.Close(ctx))
+}
diff --git a/internal/io/redis/lookup_test.go b/internal/io/redis/lookup_test.go
index bd606765..7bef1f44 100644
--- a/internal/io/redis/lookup_test.go
+++ b/internal/io/redis/lookup_test.go
@@ -134,3 +134,13 @@ func TestLookupSourceDB(t *testing.T) {
 	require.Error(t, err)
 	require.Equal(t, "redis lookup source db should be in range 0-15", err.Error())
 }
+
+func TestLookUpPingRedis(t *testing.T) {
+	s := &lookupSource{}
+	prop := map[string]interface{}{
+		"datasource": "1",
+		"addr":       addr,
+		"datatype":   "string",
+	}
+	require.NoError(t, s.Ping("1", prop))
+}