Repository: lf-edge/ekuiper
Pull request #2843 (2024-05-11 02:58:23)

chore: update codecov action

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2843?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
All modified and coverable lines are covered by tests :white_check_mark:
> :exclamation: No coverage uploaded for pull request base (`master-v2@9382d71`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit).


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


```diff
@@             Coverage Diff              @@
##             master-v2    #2843   +/-   ##
============================================
  Coverage             ?   68.54%           
============================================
  Files                ?      271           
  Lines                ?    25708           
  Branches             ?        0           
============================================
  Hits                 ?    17621           
  Misses               ?     6498           
  Partials             ?     1589           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2843?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/internal/topo/node/contract.go b/internal/topo/node/contract.go
index d098dd1b..284d32b7 100644
--- a/internal/topo/node/contract.go
+++ b/internal/topo/node/contract.go
@@ -76,12 +76,12 @@ type MergeableTopo interface {
 	// SubMetrics return the metrics of the sub nodes
 	SubMetrics() ([]string, []any)
 	// Close notifies subtopo to deref
-	Close(ruleId string)
+	Close(ctx api.StreamContext, ruleId string)
 }
 
 type SchemaNode interface {
 	// AttachSchema attach the schema to the node. The parameters are ruleId, sourceName, schema, whether is wildcard
 	AttachSchema(api.StreamContext, string, map[string]*ast.JsonStreamField, bool)
 	// DetachSchema detach the schema from the node. The parameters are ruleId
-	DetachSchema(string)
+	DetachSchema(api.StreamContext, string)
 }
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 0bbb80d9..0360bf99 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -18,7 +18,6 @@ import (
 	"fmt"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/converter"
 	schemaLayer "github.com/lf-edge/ekuiper/v2/internal/converter/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -35,23 +34,25 @@ type DecodeOp struct {
 }
 
 func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
-	ctx.GetLogger().Infof("attach schema to shared stream")
 	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("attach schema to shared stream")
 		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
 			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
 		} else {
+			ctx.GetLogger().Infof("attach schema become %+v", o.sLayer.GetSchema())
 			fastDecoder.ResetSchema(o.sLayer.GetSchema())
 		}
 	}
 }
 
-func (o *DecodeOp) DetachSchema(ruleId string) {
-	conf.Log.Infof("detach schema for shared stream rule %v", ruleId)
+func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("detach schema for shared stream rule %v", ruleId)
 		if err := o.sLayer.DetachSchema(ruleId); err != nil {
-			conf.Log.Warnf("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
+			ctx.GetLogger().Infof("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
 		} else {
 			fastDecoder.ResetSchema(o.sLayer.GetSchema())
+			ctx.GetLogger().Infof("detach schema become %+v", o.sLayer.GetSchema())
 		}
 	}
 }
diff --git a/internal/topo/subtopo.go b/internal/topo/subtopo.go
index e41d27c3..8ce506f2 100644
--- a/internal/topo/subtopo.go
+++ b/internal/topo/subtopo.go
@@ -171,7 +171,7 @@ func (s *SrcSubTopo) StoreSchema(ruleID, dataSource string, schema map[string]*a
 	}
 }
 
-func (s *SrcSubTopo) Close(ruleId string) {
+func (s *SrcSubTopo) Close(ctx api.StreamContext, ruleId string) {
 	if _, ok := s.refRules.LoadAndDelete(ruleId); ok {
 		s.refCount.Add(-1)
 		if s.refCount.Load() == 0 {
@@ -182,7 +182,7 @@ func (s *SrcSubTopo) Close(ruleId string) {
 		}
 		for _, op := range s.ops {
 			if so, ok := op.(node.SchemaNode); ok {
-				so.DetachSchema(ruleId)
+				so.DetachSchema(ctx, ruleId)
 			}
 		}
 	}
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index f638c3de..18a704d3 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -98,7 +98,7 @@ func (s *Topo) Cancel() {
 	for _, src := range s.sources {
 		switch rt := src.(type) {
 		case node.MergeableTopo:
-			rt.Close(s.name)
+			rt.Close(s.ctx, s.name)
 		}
 	}
 }


Test patch:
diff --git a/.github/workflows/run_fvt_tests.yaml b/.github/workflows/run_fvt_tests.yaml
index 90741fe1..4c2f27f9 100644
--- a/.github/workflows/run_fvt_tests.yaml
+++ b/.github/workflows/run_fvt_tests.yaml
@@ -255,7 +255,7 @@ jobs:
         sudo ./get_helm.sh
         helm version
     - name: build kuiper for docker
-      run: sudo docker build --no-cache -t lfedge/ekuiper:$(git describe --tags --alway) -f deploy/docker/Dockerfile .
+      run: sudo docker build --no-cache -t lfedge/ekuiper:$(git describe --tags --always | sed 's/^v//g') -f deploy/docker/Dockerfile .
     - name: run emqx on chart
       env:
         KUBECONFIG: "/etc/rancher/k3s/k3s.yaml"
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 495cb222..be634293 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -51,7 +51,7 @@ jobs:
         go test --tags="edgex msgpack script test" --cover -covermode=atomic -coverpkg=./... -coverprofile=coverage.xml $(go list ./... | grep -vE "github.com/lf-edge/ekuiper/internal/topo/topotest/plugin|github.com/lf-edge/ekuiper/extensions")
         make failpoint-disable
     - name: Upload coverage to Codecov
-      uses: codecov/codecov-action@v3
+      uses: codecov/codecov-action@v4
       with:
         files: coverage.xml,topotest-plugin.xml
         token: ${{ secrets.CODECOV_TOKEN }}
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index b94dea38..14e8db1e 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -190,7 +190,7 @@ func TestJSONWithSchema(t *testing.T) {
 				}
 			}
 
-			op.DetachSchema(ctx.GetRuleId())
+			op.DetachSchema(ctx, ctx.GetRuleId())
 			cases = []any{
 				&xsql.RawTuple{Emitter: "test", Rawdata: []byte("{\"a\":1,\"b\":2}"), Timestamp: 111, Metadata: map[string]any{"topic": "demo", "qos": 1}},
 				&xsql.RawTuple{Emitter: "test", Rawdata: []byte("[{\"a\":1,\"b\":2},{\"a\":3,\"b\":4,\"c\":\"hello\"}]"), Timestamp: 111, Metadata: map[string]any{"topic": "demo", "qos": 1}},
diff --git a/internal/topo/subtopo_test.go b/internal/topo/subtopo_test.go
index 49e3e55f..ec31f206 100644
--- a/internal/topo/subtopo_test.go
+++ b/internal/topo/subtopo_test.go
@@ -63,7 +63,8 @@ func TestSubtopoLC(t *testing.T) {
 	}
 	assert.Equal(t, ptopo, subTopo.topo)
 	// Test run
-	subTopo.Open(mockContext.NewMockContext("rule1", "abc"), make(chan error))
+	ctx1 := mockContext.NewMockContext("rule1", "abc")
+	subTopo.Open(ctx1, make(chan error))
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	assert.Equal(t, 1, opNode.schemaCount)
 	// Run another
@@ -73,7 +74,8 @@ func TestSubtopoLC(t *testing.T) {
 	subTopo.StoreSchema("rule2", "shared", map[string]*ast.JsonStreamField{
 		"field2": {Type: "string"},
 	}, false)
-	subTopo2.Open(mockContext.NewMockContext("rule2", "abc"), make(chan error))
+	ctx2 := mockContext.NewMockContext("rule2", "abc")
+	subTopo2.Open(ctx2, make(chan error))
 	assert.Equal(t, int32(2), subTopo.refCount.Load())
 	assert.Equal(t, 2, opNode.schemaCount)
 	// Metrics test
@@ -106,10 +108,10 @@ func TestSubtopoLC(t *testing.T) {
 	assert.Equal(t, []checkpoint.StreamTask{srcNode}, sources)
 	assert.Equal(t, []checkpoint.NonSourceTask{opNode}, ops)
 	// Stop
-	subTopo.Close("rule1")
+	subTopo.Close(ctx1, "rule1")
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	assert.Equal(t, 1, mlen(&subTopoPool))
-	subTopo2.Close("rule2")
+	subTopo2.Close(ctx2, "rule2")
 	assert.Equal(t, int32(0), subTopo.refCount.Load())
 	assert.Equal(t, 0, mlen(&subTopoPool))
 	assert.Equal(t, 2, len(subTopo.schemaReg))
@@ -132,33 +134,36 @@ func TestSubtopoRunError(t *testing.T) {
 	assert.Equal(t, 1, mlen(&subTopoPool))
 	// Test run firstly, successfully
 	assert.Equal(t, false, subTopo.opened.Load())
-	subTopo.Open(mockContext.NewMockContext("rule1", "abc"), make(chan error))
+	ctx1 := mockContext.NewMockContext("rule1", "abc")
+	subTopo.Open(ctx1, make(chan error))
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	assert.Equal(t, true, subTopo.opened.Load())
-	subTopo.Close("rule1")
+	subTopo.Close(ctx1, "rule1")
 	assert.Equal(t, int32(0), subTopo.refCount.Load())
 	assert.Equal(t, 0, mlen(&subTopoPool))
 	time.Sleep(10 * time.Millisecond)
 	assert.Equal(t, false, subTopo.opened.Load())
 	// Test run secondly and thirdly, should fail
 	errCh1 := make(chan error, 1)
-	subTopo.Open(mockContext.NewMockContext("rule1", "abc"), errCh1)
+	ctx1 = mockContext.NewMockContext("rule1", "abc")
+	subTopo.Open(ctx1, errCh1)
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	errCh2 := make(chan error, 1)
 	assert.Equal(t, true, subTopo.opened.Load())
-	subTopo.Open(mockContext.NewMockContext("rule2", "abc"), errCh2)
+	ctx2 := mockContext.NewMockContext("rule2", "abc")
+	subTopo.Open(ctx2, errCh2)
 	assert.Equal(t, int32(2), subTopo.refCount.Load())
 	select {
 	case err := <-errCh1:
 		assert.Equal(t, assert.AnError, err)
-		subTopo.Close("rule1")
+		subTopo.Close(ctx1, "rule1")
 	case <-time.After(1 * time.Second):
 		assert.Fail(t, "Should receive error")
 	}
 	select {
 	case err := <-errCh2:
 		assert.Equal(t, assert.AnError, err)
-		subTopo2.Close("rule2")
+		subTopo2.Close(ctx2, "rule2")
 	case <-time.After(1 * time.Second):
 		assert.Fail(t, "Should receive error")
 	}
@@ -326,6 +331,6 @@ func (m *mockOp) AttachSchema(ctx api.StreamContext, dataSource string, schema m
 	m.schemaCount++
 }
 
-func (m *mockOp) DetachSchema(ruleId string) {
+func (m *mockOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	m.schemaCount--
 }