Repository: lf-edge/ekuiper
Pull request #1417 (2022-09-28 03:18:39)

feat(lookup): lookup cache
Add cache support to lookup node


Patch:
diff --git a/extensions/sources/sql/sql.yaml b/extensions/sources/sql/sql.yaml
index 6a5093be..78f6ed17 100644
--- a/extensions/sources/sql/sql.yaml
+++ b/extensions/sources/sql/sql.yaml
@@ -8,6 +8,10 @@ default:
     indexValue: "2022-04-21 10:23:55"
     indexFieldType: "DATETIME"
     dateTimeFormat: "YYYY-MM-dd HH:mm:ss"
+  lookup:
+    cache: true
+    cacheTtl: 600
+    cacheMissingKey: true
 
 sqlserver_config:
   url: sqlserver://username:password@140.210.204.147/testdb
diff --git a/internal/topo/lookup/cache/cache.go b/internal/topo/lookup/cache/cache.go
new file mode 100644
index 00000000..f8c06a32
--- /dev/null
+++ b/internal/topo/lookup/cache/cache.go
@@ -0,0 +1,106 @@
+// Copyright 2022 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 cache
+
+import (
+	"context"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"sync"
+)
+
+type item struct {
+	data       []api.SourceTuple
+	expiration int64
+}
+
+type Cache struct {
+	expireTime      int
+	cacheMissingKey bool
+	cancel          context.CancelFunc
+	items           map[string]*item
+	sync.RWMutex
+}
+
+func NewCache(expireTime int, cacheMissingKey bool) *Cache {
+	c := &Cache{
+		expireTime:      expireTime,
+		cacheMissingKey: cacheMissingKey,
+		items:           make(map[string]*item),
+	}
+	if expireTime > 0 {
+		ctx, cancel := context.WithCancel(context.Background())
+		c.cancel = cancel
+		go c.run(ctx)
+	}
+	return c
+}
+
+func (c *Cache) run(ctx context.Context) {
+	ticker := conf.GetTicker(c.expireTime * 2000)
+	for {
+		select {
+		case <-ticker.C:
+			c.deleteExpired()
+		case <-ctx.Done():
+			conf.Log.Infof("Lookup cache is stopped")
+			ticker.Stop()
+			return
+		}
+	}
+}
+
+func (c *Cache) deleteExpired() {
+	now := conf.GetNowInMilli()
+	c.Lock()
+	for k, v := range c.items {
+		if v.expiration > 0 && now > v.expiration {
+			delete(c.items, k)
+		}
+	}
+	c.Unlock()
+}
+
+func (c *Cache) Set(key string, value []api.SourceTuple) {
+	if (value == nil || len(value) == 0) && !c.cacheMissingKey {
+		return
+	}
+	c.Lock()
+	defer c.Unlock()
+	if c.expireTime > 0 {
+		c.items[key] = &item{data: value, expiration: conf.GetNowInMilli() + int64(c.expireTime*1000)}
+	} else {
+		c.items[key] = &item{data: value}
+	}
+}
+
+func (c *Cache) Get(key string) ([]api.SourceTuple, bool) {
+	c.RLock()
+	defer c.RUnlock()
+	if v, ok := c.items[key]; ok {
+		if v.expiration > 0 && conf.GetNowInMilli() > v.expiration {
+			return nil, false
+		}
+		return v.data, true
+	}
+	return nil, false
+}
+
+func (c *Cache) Close() {
+	if c.cancel != nil {
+		c.cancel()
+	}
+	c.items = nil
+}
diff --git a/internal/topo/lookup/table.go b/internal/topo/lookup/table.go
index 55fdc924..bb7e79aa 100644
--- a/internal/topo/lookup/table.go
+++ b/internal/topo/lookup/table.go
@@ -67,7 +67,7 @@ func CreateInstance(name string, sourceType string, options *ast.Options) error
 	defer lock.Unlock()
 	contextLogger := conf.Log.WithField("table", name)
 	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
-	props := nodeConf.GetSourceConf(ctx, sourceType, options)
+	props := nodeConf.GetSourceConf(sourceType, options)
 	ctx.GetLogger().Infof("open lookup table with props %v", conf.Printable(props))
 	// Create the lookup source according to the source options
 	ns, err := io.LookupSource(sourceType)
diff --git a/internal/topo/node/conf/source.go b/internal/topo/node/conf/source.go
index e79a8882..689288e6 100644
--- a/internal/topo/node/conf/source.go
+++ b/internal/topo/node/conf/source.go
@@ -16,14 +16,12 @@ package conf
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"strings"
 )
 
-func GetSourceConf(ctx api.StreamContext, sourceType string, options *ast.Options) map[string]interface{} {
+func GetSourceConf(sourceType string, options *ast.Options) map[string]interface{} {
 	confkey := options.CONF_KEY
-	logger := ctx.GetLogger()
 	confPath := "sources/" + sourceType + ".yaml"
 	if sourceType == "mqtt" {
 		confPath = "mqtt_source.yaml"
@@ -32,11 +30,11 @@ func GetSourceConf(ctx api.StreamContext, sourceType string, options *ast.Option
 	cfg := make(map[string]interface{})
 	err := conf.LoadConfigByName(confPath, &cfg)
 	if err != nil {
-		logger.Warnf("fail to parse yaml for source %s. Return an empty configuration", sourceType)
+		conf.Log.Warnf("fail to parse yaml for source %s. Return an empty configuration", sourceType)
 	} else {
 		def, ok := cfg["default"]
 		if !ok {
-			logger.Warnf("default conf is not found", confkey)
+			conf.Log.Warnf("default conf %s is not found", confkey)
 		} else {
 			if def1, ok1 := def.(map[string]interface{}); ok1 {
 				props = def1
@@ -56,7 +54,7 @@ func GetSourceConf(ctx api.StreamContext, sourceType string, options *ast.Option
 		f = "json"
 	}
 	props["format"] = strings.ToLower(f)
-	logger.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
+	conf.Log.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
 	return props
 }
 
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index 70ffb672..46b92b50 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -18,13 +18,22 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/lookup"
+	"github.com/lf-edge/ekuiper/internal/topo/lookup/cache"
+	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
+type LookupConf struct {
+	Cache           bool `json:"cache"`
+	CacheTTL        int  `json:"cacheTtl"`
+	CacheMissingKey bool `json:"cacheMissingKey"`
+}
+
 // LookupNode will look up the data from the external source when receiving an event
 type LookupNode struct {
 	*defaultSinkNode
@@ -34,6 +43,7 @@ type LookupNode struct {
 	vals        []ast.Expr
 
 	srcOptions *ast.Options
+	conf       *LookupConf
 	fields     []string
 	keys       []string
 }
@@ -43,10 +53,19 @@ func NewLookupNode(name string, fields []string, keys []string, joinType ast.Joi
 	if t == "" {
 		return nil, fmt.Errorf("source type is not specified")
 	}
+	props := nodeConf.GetSourceConf(t, srcOptions)
+	lookupConf := &LookupConf{}
+	if lc, ok := props["lookup"].(map[string]interface{}); ok {
+		err := cast.MapToStruct(lc, lookupConf)
+		if err != nil {
+			return nil, err
+		}
+	}
 	n := &LookupNode{
 		fields:     fields,
 		keys:       keys,
 		srcOptions: srcOptions,
+		conf:       lookupConf,
 		sourceType: t,
 		joinType:   joinType,
 		vals:       vals,
@@ -85,6 +104,11 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 			}
 			defer lookup.Detach(n.name)
 			fv, _ := xsql.NewFunctionValuersForOp(ctx)
+			var c *cache.Cache
+			if n.conf.Cache {
+				c = cache.NewCache(n.conf.CacheTTL, n.conf.CacheMissingKey)
+				defer c.Close()
+			}
 			// Start the lookup source loop
 			for {
 				log.Debugf("LookupNode %s is looping", n.name)
@@ -109,7 +133,7 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 						log.Debugf("Lookup Node receive tuple input %s", d)
 						n.statManager.ProcessTimeStart()
 						sets := &xsql.JoinTuples{Content: make([]*xsql.JoinTuple, 0)}
-						err := n.lookup(ctx, d, fv, ns, sets)
+						err := n.lookup(ctx, d, fv, ns, sets, c)
 						if err != nil {
 							n.Broadcast(err)
 							n.statManager.IncTotalExceptions(err.Error())
@@ -128,7 +152,7 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 							if !ok {
 								return false, fmt.Errorf("Invalid window element, must be a tuple row but got %v", r)
 							}
-							err := n.lookup(ctx, tr, fv, ns, sets)
+							err := n.lookup(ctx, tr, fv, ns, sets, c)
 							if err != nil {
 								return false, err
 							}
@@ -160,7 +184,8 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.TupleRow, fv *xsql.FunctionValuer, ns api.LookupSource, tuples *xsql.JoinTuples) error {
+// lookup will lookup the cache firstly, if expires, read the external source
+func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.TupleRow, fv *xsql.FunctionValuer, ns api.LookupSource, tuples *xsql.JoinTuples, c *cache.Cache) error {
 	ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(d, fv)}
 	cvs := make([]interface{}, len(n.vals))
 	hasNil := false
@@ -171,11 +196,24 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.TupleRow, fv *xsql.Fun
 		}
 	}
 	var (
-		r []api.SourceTuple
-		e error
+		r  []api.SourceTuple
+		e  error
+		ok bool
 	)
 	if !hasNil { // if any of the value is nil, the lookup will always return empty result
-		r, e = ns.Lookup(ctx, n.fields, n.keys, cvs)
+		if c != nil {
+			k := fmt.Sprintf("%v", cvs)
+			r, ok = c.Get(k)
+			if !ok {
+				r, e = ns.Lookup(ctx, n.fields, n.keys, cvs)
+				if e != nil {
+					return e
+				}
+				c.Set(k, r)
+			}
+		} else {
+			r, e = ns.Lookup(ctx, n.fields, n.keys, cvs)
+		}
 	}
 	if e != nil {
 		return e
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index b3fb4cb6..117c014f 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -73,7 +73,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 	logger.Infof("open source node %s with option %v", m.name, m.options)
 	go func() {
 		panicOrError := infra.SafeRun(func() error {
-			props := nodeConf.GetSourceConf(ctx, m.sourceType, m.options)
+			props := nodeConf.GetSourceConf(m.sourceType, m.options)
 			m.props = props
 			if c, ok := props["concurrency"]; ok {
 				if t, err := cast.ToInt(c, cast.STRICT); err != nil || t <= 0 {


Test patch:
diff --git a/internal/topo/lookup/cache/cache_test.go b/internal/topo/lookup/cache/cache_test.go
new file mode 100644
index 00000000..49de1457
--- /dev/null
+++ b/internal/topo/lookup/cache/cache_test.go
@@ -0,0 +1,120 @@
+// Copyright 2022 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 cache
+
+import (
+	"github.com/benbjohnson/clock"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+	"time"
+)
+
+func TestExpiration(t *testing.T) {
+	c := NewCache(20, false)
+	defer c.Close()
+	clock := conf.Clock.(*clock.Mock)
+	expects := [][]api.SourceTuple{
+		{api.NewDefaultSourceTuple(map[string]interface{}{"a": 1}, nil)},
+		{api.NewDefaultSourceTuple(map[string]interface{}{"a": 2}, nil), api.NewDefaultSourceTuple(map[string]interface{}{"a": 3}, nil)},
+		{},
+	}
+	c.Set("a", expects[0])
+	clock.Add(10 * time.Second)
+	c.Set("b", expects[1])
+	c.Set("c", expects[2])
+	clock.Add(1 * time.Second)
+	r1, ok := c.Get("a")
+	if !ok {
+		t.Error("a should exist")
+		return
+	}
+	if !reflect.DeepEqual(r1, expects[0]) {
+		t.Errorf("expect %v but get %v", expects[0], r1)
+	}
+	r2, ok := c.Get("b")
+	if !ok {
+		t.Error("b should exist")
+		return
+	}
+	if !reflect.DeepEqual(r2, expects[1]) {
+		t.Errorf("expect %v but get %v", expects[1], r2)
+	}
+	_, ok = c.Get("c")
+	if ok {
+		t.Error("c should not exist")
+		return
+	}
+
+	clock.Add(10 * time.Second)
+	r1, ok = c.Get("a")
+	if ok {
+		t.Error("a should not exist after expiration")
+		return
+	}
+	r2, ok = c.Get("b")
+	if !ok {
+		t.Error("b should exist")
+		return
+	}
+	if !reflect.DeepEqual(r2, expects[1]) {
+		t.Errorf("expect %v but get %v", expects[1], r2)
+	}
+	clock.Add(10 * time.Second)
+	r2, ok = c.Get("b")
+	if ok {
+		t.Error("b should not exist after expiration")
+		return
+	}
+}
+
+func TestNoExpiration(t *testing.T) {
+	c := NewCache(0, true)
+	defer c.Close()
+	clock := conf.Clock.(*clock.Mock)
+	expects := [][]api.SourceTuple{
+		{api.NewDefaultSourceTuple(map[string]interface{}{"a": 1}, nil)},
+		{api.NewDefaultSourceTuple(map[string]interface{}{"a": 2}, nil), api.NewDefaultSourceTuple(map[string]interface{}{"a": 3}, nil)},
+		{},
+	}
+	c.Set("a", expects[0])
+	clock.Add(10 * time.Second)
+	c.Set("b", expects[1])
+	c.Set("c", expects[2])
+
+	clock.Add(100 * time.Second)
+	r1, ok := c.Get("a")
+	if !ok {
+		t.Error("a should exist")
+		return
+	}
+	if !reflect.DeepEqual(r1, expects[0]) {
+		t.Errorf("expect %v but get %v", expects[0], r1)
+	}
+	r2, ok := c.Get("b")
+	if !ok {
+		t.Error("b should exist")
+		return
+	}
+	if !reflect.DeepEqual(r2, expects[1]) {
+		t.Errorf("expect %v but get %v", expects[1], r2)
+	}
+	_, ok = c.Get("c")
+	if !ok {
+		t.Error("c should exist")
+		return
+	}
+}
diff --git a/internal/topo/node/lookup_node_test.go b/internal/topo/node/lookup_node_test.go
index b50695b3..51d82782 100644
--- a/internal/topo/node/lookup_node_test.go
+++ b/internal/topo/node/lookup_node_test.go
@@ -16,6 +16,7 @@ package node
 
 import (
 	"fmt"
+	"github.com/benbjohnson/clock"
 	"github.com/lf-edge/ekuiper/internal/binder"
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -30,6 +31,7 @@ import (
 )
 
 type mockLookupSrc struct {
+	data []api.SourceTuple // new mock data
 }
 
 func (m *mockLookupSrc) Open(_ api.StreamContext) error {
@@ -41,7 +43,17 @@ func (m *mockLookupSrc) Configure(_ string, _ map[string]interface{}) error {
 }
 
 // Lookup accept int value as the first array value
-func (m *mockLookupSrc) Lookup(_ api.StreamContext, _ []string, _ []string, values []interface{}) ([]api.SourceTuple, error) {
+func (m *mockLookupSrc) Lookup(_ api.StreamContext, fields []string, _ []string, values []interface{}) ([]api.SourceTuple, error) {
+	if len(fields) > 0 { // if fields is not empty, the value will be kept
+		if m.data != nil {
+			return m.data, nil
+		} else {
+			m.data = []api.SourceTuple{api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": 1000,
+				"newB": 1000,
+			}, nil)}
+		}
+	}
 	a1, ok := values[0].(int)
 	if ok {
 		var result []api.SourceTuple
@@ -341,3 +353,177 @@ func TestLookup(t *testing.T) {
 		}
 	}
 }
+
+func TestCachedLookup(t *testing.T) {
+	options := &ast.Options{
+		DATASOURCE:        "mock",
+		TYPE:              "mock",
+		STRICT_VALIDATION: true,
+		KIND:              "lookup",
+	}
+	lookup.CreateInstance("mock", "mock", options)
+	contextLogger := conf.Log.WithField("rule", "TestLookup")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	l, _ := NewLookupNode("mock", []string{"fixed"}, []string{"a"}, ast.INNER_JOIN, []ast.Expr{&ast.FieldRef{
+		StreamName: "",
+		Name:       "a",
+	}}, options, &api.RuleOption{
+		IsEventTime:        false,
+		LateTol:            0,
+		Concurrency:        0,
+		BufferLength:       0,
+		SendMetaToSink:     false,
+		SendError:          false,
+		Qos:                0,
+		CheckpointInterval: 0,
+	})
+	l.conf = &LookupConf{
+		Cache:           true,
+		CacheTTL:        20,
+		CacheMissingKey: false,
+	}
+	errCh := make(chan error)
+	outputCh := make(chan interface{}, 1)
+	l.outputs["mock"] = outputCh
+	l.Exec(ctx, errCh)
+	input := &xsql.Tuple{
+		Emitter: "demo",
+		Message: map[string]interface{}{
+			"a": 6,
+			"b": "aaaa",
+		},
+	}
+	outputBefore := &xsql.JoinTuples{
+		Content: []*xsql.JoinTuple{
+			{
+				Tuples: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "demo",
+						Message: map[string]interface{}{
+							"a": 6,
+							"b": "aaaa",
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "mock",
+						Message: map[string]interface{}{
+							"newA": 1,
+							"newB": 2,
+						},
+						Timestamp: 11000,
+					},
+				},
+			}, {
+				Tuples: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "demo",
+						Message: map[string]interface{}{
+							"a": 6,
+							"b": "aaaa",
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "mock",
+						Message: map[string]interface{}{
+							"newA": 6,
+							"newB": 12,
+						},
+						Timestamp: 11000,
+					},
+				},
+			},
+		},
+	}
+	outputAfter := &xsql.JoinTuples{
+		Content: []*xsql.JoinTuple{
+			{
+				Tuples: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "demo",
+						Message: map[string]interface{}{
+							"a": 6,
+							"b": "aaaa",
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "mock",
+						Message: map[string]interface{}{
+							"newA": 1000,
+							"newB": 1000,
+						},
+						Timestamp: 22000,
+					},
+				},
+			},
+		},
+	}
+	// First run and the set mock result
+	clock := conf.Clock.(*clock.Mock)
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case l.input <- input:
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case <-outputCh:
+		//if !reflect.DeepEqual(output, outputBefore) {
+		//	t.Errorf("\nfirst case: expect %v but got %v", output, outputBefore)
+		//}
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+	// Get cache
+	clock.Add(11 * time.Second)
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case l.input <- input:
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case output := <-outputCh:
+		if !reflect.DeepEqual(output, outputBefore) {
+			t.Errorf("\ncached case: expect %v but got %v", output, outputBefore)
+		}
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+	// Cache expired
+	clock.Add(11 * time.Second)
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case l.input <- input:
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+	select {
+	case err := <-errCh:
+		t.Error(err)
+		return
+	case output := <-outputCh:
+		if !reflect.DeepEqual(output, outputAfter) {
+			t.Errorf("\nexpired case: expect %v but got %v", output, outputAfter)
+		}
+	case <-time.After(1 * time.Second):
+		t.Error("send message timeout")
+		return
+	}
+}
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index c3f09b35..f50afabf 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -15,8 +15,6 @@
 package node
 
 import (
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/context"
 	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -43,9 +41,7 @@ func TestGetConf_Apply(t *testing.T) {
 		DATASOURCE: "/feed",
 		TYPE:       "httppull",
 	}, false)
-	contextLogger := conf.Log.WithField("rule", "test")
-	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	conf := nodeConf.GetSourceConf(ctx, n.sourceType, n.options)
+	conf := nodeConf.GetSourceConf(n.sourceType, n.options)
 	if !reflect.DeepEqual(result, conf) {
 		t.Errorf("result mismatch:\n\nexp=%s\n\ngot=%s\n\n", result, conf)
 	}
@@ -71,9 +67,7 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 		TYPE:       "httppull",
 		CONF_KEY:   "application_conf",
 	}, false)
-	contextLogger := conf.Log.WithField("rule", "test")
-	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	conf := nodeConf.GetSourceConf(ctx, n.sourceType, n.options)
+	conf := nodeConf.GetSourceConf(n.sourceType, n.options)
 	if !reflect.DeepEqual(result, conf) {
 		t.Errorf("result mismatch:\n\nexp=%s\n\ngot=%s\n\n", result, conf)
 		return