Repository: megaease/easegress
Pull request #409 (2021-12-14 02:28:55)

Mock filter to support header match
This is to support Mock feature in EaseMesh.

codecov-commenter: # [Codecov](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) Report
> Merging [#409](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (98635ac) into [main](https://codecov.io/gh/megaease/easegress/commit/327b43bfb1a5ccc636a5d6e4f2be68e2830eb908?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (327b43b) will **decrease** coverage by `0.28%`.
> The diff coverage is `87.30%`.

[![Impacted file tree graph](https://codecov.io/gh/megaease/easegress/pull/409/graphs/tree.svg?width=650&height=150&src=pr&token=5Q80B98LPI&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)

```diff
@@            Coverage Diff             @@
##             main     #409      +/-   ##
==========================================
- Coverage   80.37%   80.08%   -0.29%     
==========================================
  Files          62       62              
  Lines        7107     7140      +33     
==========================================
+ Hits         5712     5718       +6     
- Misses       1100     1125      +25     
- Partials      295      297       +2     
```


| [Impacted Files](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | Coverage Δ | |
|---|---|---|
| [pkg/filter/mock/mock.go](https://codecov.io/gh/megaease/easegress/pull/409/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2ZpbHRlci9tb2NrL21vY2suZ28=) | `90.42% <87.30%> (-4.66%)` | :arrow_down: |
| [pkg/object/mqttproxy/session.go](https://codecov.io/gh/megaease/easegress/pull/409/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvc2Vzc2lvbi5nbw==) | `74.39% <0.00%> (-7.93%)` | :arrow_down: |
| [pkg/object/mqttproxy/client.go](https://codecov.io/gh/megaease/easegress/pull/409/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvY2xpZW50Lmdv) | `79.87% <0.00%> (-1.22%)` | :arrow_down: |
| [pkg/cluster/cluster.go](https://codecov.io/gh/megaease/easegress/pull/409/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY2x1c3Rlci5nbw==) | `49.57% <0.00%> (-1.03%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Last update [0ec6ff1...98635ac](https://codecov.io/gh/megaease/easegress/pull/409?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease).


Patch:
diff --git a/doc/cookbook/multi-node-cluster.md b/doc/cookbook/multi-node-cluster.md
index ef4d1f2b..d795b69f 100644
--- a/doc/cookbook/multi-node-cluster.md
+++ b/doc/cookbook/multi-node-cluster.md
@@ -118,12 +118,14 @@ easegress-server \
   --primary-listen-peer-urls http://$HOST1:2380 \
   --state-flag "existing"
 ```
-Here `primary-listen-peer-urls` tell, where to find a *primary* cluster member and `state-flag` with value "existing" means that this cluster was already created.
+Here `primary-listen-peer-urls` tell, where to find a *primary* cluster members and `state-flag` with value "existing" means that this cluster was already created.
 
 We can now see also the 4th instance:
 ```bash
 egctl --server $HOST1:2381 member list | grep " name"
-# prints
+```
+prints to output:
+```bash
     name: machine-1
     name: machine-2
     name: machine-3
@@ -179,7 +181,8 @@ memory-profile-file:
 log-dir: ./log
 debug: false
 cluster:
-  primary-listen-peer-urls: http://$HOST1:2380
+  primary-listen-peer-urls:
+   - http://$HOST1:2380
 ```
 
 ## Configuration tips (optional)
diff --git a/doc/reference/filters.md b/doc/reference/filters.md
index 380cd882..91b3571e 100644
--- a/doc/reference/filters.md
+++ b/doc/reference/filters.md
@@ -64,6 +64,7 @@
     - [proxy.Compression](#proxycompression)
     - [proxy.MTLS](#proxymtls)
     - [mock.Rule](#mockrule)
+    - [mock.MatchRule](#mockmatchrule)
     - [circuitbreaker.Policy](#circuitbreakerpolicy)
     - [ratelimiter.Policy](#ratelimiterpolicy)
     - [timelimiter.URLRule](#timelimiterurlrule)
@@ -285,7 +286,8 @@ Below is an example configuration to mock response for requests to path `/users/
 kind: Mock
 name: mock-example
 rules:
-- path: /users/1
+- match:
+    path: /users/1
   code: 200
   headers:
     Content-Type: application/json
@@ -801,12 +803,21 @@ The relationship between `methods` and `url` is `AND`.
 | Name       | Type              | Description                                                                                                                                         | Required |
 | ---------- | ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
 | code       | int               | HTTP status code of the mocked response                                                                                                             | Yes      |
-| path       | string            | Path match criteria, if request path is the value of this option, then the response of the request is mocked according to this rule                 | No       |
-| pathPrefix | string            | Path prefix match criteria, if request path begins with the value of this option, then the response of the request is mocked according to this rule | No       |
+| match      | [MatchRule](#mock.MatchRule) | Rule to match a request        | Yes      |
 | delay      | string            | Delay duration, for the request processing time mocking                                                                                             | No       |
 | headers    | map[string]string | Headers of the mocked response                                                                                                                      | No       |
 | body       | string            | Body of the mocked response, default is an empty string                                                                                             | No       |
 
+### mock.MatchRule
+
+| Name       | Type              | Description                                                                                                                                         | Required |
+| ---------- | ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
+| path       | string            | Path match criteria, if request path is the value of this option, then the response of the request is mocked according to this rule                 | No       |
+| pathPrefix | string            | Path prefix match criteria, if request path begins with the value of this option, then the response of the request is mocked according to this rule | No       |
+| matchAllHeaders | bool          | Whether to match all headers | No       |
+| headers    | map[string][url.StringMatch](#urlrulestringmatch) | Headers to match, key is a header name, value is the rule to match the header value | No |
+
+
 ### circuitbreaker.Policy
 
 | Name                                  | Type   | Description                                                                                                                                                                                                                                                                                                                                                                                                                              | Required |
diff --git a/example/clean_cluster.sh b/example/clean_cluster.sh
index 53371c7b..ae36360b 100755
--- a/example/clean_cluster.sh
+++ b/example/clean_cluster.sh
@@ -5,7 +5,7 @@ SCRIPTPATH=`pwd -P`
 popd > /dev/null
 SCRIPTFILE=`basename $0`
 
-for MEMBER_PATH in primary-00{1,2,3} secondary-00{4,5}
+for MEMBER_PATH in primary-00{1,2,3} secondary-00{4,5} primary-single
 do
 	${SCRIPTPATH}/${MEMBER_PATH}/stop.sh -f
 	rm -fr ${SCRIPTPATH}/${MEMBER_PATH}/{nohup.out,log,member,data,data_bak,easegress.pid,running_objects.yaml}
diff --git a/example/primary-001/conf/config.yaml b/example/primary-001/conf/config.yaml
index 82ce2d1a..247e5fb1 100644
--- a/example/primary-001/conf/config.yaml
+++ b/example/primary-001/conf/config.yaml
@@ -2,11 +2,18 @@ name: primary-001
 cluster-name: cluster-test
 cluster-role: primary
 cluster:
-cluster-listen-client-urls: [http://127.0.0.1:12379]
-cluster-listen-peer-urls: [http://127.0.0.1:12380]
-cluster-advertise-client-urls: [http://127.0.0.1:12379]
-cluster-initial-advertise-peer-urls: [http://127.0.0.1:12380]
-cluster-join-urls: [http://127.0.0.1:12380,http://127.0.0.1:22380,http://127.0.0.1:32380]
+  listen-peer-urls:
+   - http://127.0.0.1:12380
+  listen-client-urls:
+   - http://127.0.0.1:12379
+  advertise-client-urls:
+   - http://127.0.0.1:12379
+  initial-advertise-peer-urls:
+   - http://127.0.0.1:12380
+  initial-cluster:
+   - primary-001: http://127.0.0.1:12380
+   - primary-002: http://127.0.0.1:22380
+   - primary-003: http://127.0.0.1:32380
 api-addr: 127.0.0.1:12381
 data-dir: ./data
 wal-dir: ""
diff --git a/example/primary-002/conf/config.yaml b/example/primary-002/conf/config.yaml
index 2d947173..33ccb0d9 100644
--- a/example/primary-002/conf/config.yaml
+++ b/example/primary-002/conf/config.yaml
@@ -1,11 +1,19 @@
 name: primary-002
 cluster-name: cluster-test
 cluster-role: primary
-cluster-listen-client-urls: [http://127.0.0.1:22379]
-cluster-listen-peer-urls: [http://127.0.0.1:22380]
-cluster-advertise-client-urls: [http://127.0.0.1:22379]
-cluster-initial-advertise-peer-urls: [http://127.0.0.1:22380]
-cluster-join-urls: [http://127.0.0.1:12380,http://127.0.0.1:22380,http://127.0.0.1:32380]
+cluster:
+  listen-peer-urls:
+   - http://127.0.0.1:22380
+  listen-client-urls:
+   - http://127.0.0.1:22379
+  advertise-client-urls:
+   - http://127.0.0.1:22379
+  initial-advertise-peer-urls:
+   - http://127.0.0.1:22380
+  initial-cluster:
+   - primary-001: http://127.0.0.1:12380
+   - primary-002: http://127.0.0.1:22380
+   - primary-003: http://127.0.0.1:32380
 api-addr: 127.0.0.1:22381
 data-dir: ./data
 wal-dir: ""
diff --git a/example/primary-003/conf/config.yaml b/example/primary-003/conf/config.yaml
index 9948c2a2..9c67b7a7 100644
--- a/example/primary-003/conf/config.yaml
+++ b/example/primary-003/conf/config.yaml
@@ -1,11 +1,19 @@
 name: primary-003
 cluster-name: cluster-test
 cluster-role: primary
-cluster-listen-client-urls: [http://127.0.0.1:32379]
-cluster-listen-peer-urls: [http://127.0.0.1:32380]
-cluster-advertise-client-urls: [http://127.0.0.1:32379]
-cluster-initial-advertise-peer-urls: [http://127.0.0.1:32380]
-cluster-join-urls: [http://127.0.0.1:12380,http://127.0.0.1:22380,http://127.0.0.1:32380]
+cluster:
+  listen-peer-urls:
+   - http://127.0.0.1:32380
+  listen-client-urls:
+   - http://127.0.0.1:32379
+  advertise-client-urls:
+   - http://127.0.0.1:32379
+  initial-advertise-peer-urls:
+   - http://127.0.0.1:32380
+  initial-cluster:
+   - primary-001: http://127.0.0.1:12380
+   - primary-002: http://127.0.0.1:22380
+   - primary-003: http://127.0.0.1:32380
 api-addr: 127.0.0.1:32381
 data-dir: ./data
 wal-dir: ""
diff --git a/example/primary-single/bin b/example/primary-single/bin
new file mode 120000
index 00000000..db397da2
--- /dev/null
+++ b/example/primary-single/bin
@@ -0,0 +1,1 @@
+../../bin
\ No newline at end of file
diff --git a/example/primary-single/conf/config.yaml b/example/primary-single/conf/config.yaml
new file mode 100644
index 00000000..6eaa9343
--- /dev/null
+++ b/example/primary-single/conf/config.yaml
@@ -0,0 +1,10 @@
+name: primary-single
+cluster-name: cluster-test
+cluster-role: primary
+api-addr: 127.0.0.1:12381
+data-dir: ./data
+wal-dir: ""
+cpu-profile-file:
+memory-profile-file:
+log-dir: ./log
+debug: false
diff --git a/example/primary-single/egctl.sh b/example/primary-single/egctl.sh
new file mode 120000
index 00000000..af481a03
--- /dev/null
+++ b/example/primary-single/egctl.sh
@@ -0,0 +1,1 @@
+../sbin/egctl.sh
\ No newline at end of file
diff --git a/example/primary-single/start.sh b/example/primary-single/start.sh
new file mode 120000
index 00000000..792329a4
--- /dev/null
+++ b/example/primary-single/start.sh
@@ -0,0 +1,1 @@
+../sbin/start.sh
\ No newline at end of file
diff --git a/example/primary-single/status.sh b/example/primary-single/status.sh
new file mode 120000
index 00000000..06750a2e
--- /dev/null
+++ b/example/primary-single/status.sh
@@ -0,0 +1,1 @@
+../sbin/status.sh
\ No newline at end of file
diff --git a/example/primary-single/stop.sh b/example/primary-single/stop.sh
new file mode 120000
index 00000000..5046e936
--- /dev/null
+++ b/example/primary-single/stop.sh
@@ -0,0 +1,1 @@
+../sbin/stop.sh
\ No newline at end of file
diff --git a/example/secondary-004/bin b/example/secondary-004/bin
new file mode 120000
index 00000000..db397da2
--- /dev/null
+++ b/example/secondary-004/bin
@@ -0,0 +1,1 @@
+../../bin
\ No newline at end of file
diff --git a/example/secondary-005/bin b/example/secondary-005/bin
new file mode 120000
index 00000000..db397da2
--- /dev/null
+++ b/example/secondary-005/bin
@@ -0,0 +1,1 @@
+../../bin
\ No newline at end of file
diff --git a/example/stop_cluster.sh b/example/stop_cluster.sh
index f9e3afb1..b4f6f0e8 100755
--- a/example/stop_cluster.sh
+++ b/example/stop_cluster.sh
@@ -5,7 +5,7 @@ SCRIPTPATH=`pwd -P`
 popd > /dev/null
 SCRIPTFILE=`basename $0`
 
-for MEMBER_PATH in primary-00{1,2,3} secondary-00{4,5}
+for MEMBER_PATH in primary-00{1,2,3} secondary-00{4,5} primary-single
 do
 	echo "stop ${MEMBER_PATH}"
 	${SCRIPTPATH}/${MEMBER_PATH}/stop.sh -f
diff --git a/go.mod b/go.mod
index 795a21bd..b4a4732e 100644
--- a/go.mod
+++ b/go.mod
@@ -35,7 +35,7 @@ require (
 	github.com/libdns/route53 v1.1.2
 	github.com/libdns/vultr v0.0.0-20211122184636-cd4cb5c12e51
 	github.com/lucas-clemente/quic-go v0.24.0
-	github.com/megaease/easemesh-api v1.3.4
+	github.com/megaease/easemesh-api v1.3.5
 	github.com/megaease/grace v1.0.0
 	github.com/mitchellh/mapstructure v1.4.1
 	github.com/nacos-group/nacos-sdk-go v1.0.8
diff --git a/go.sum b/go.sum
index d1cf7bcb..e8353e5c 100644
--- a/go.sum
+++ b/go.sum
@@ -937,8 +937,8 @@ github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpe
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
-github.com/megaease/easemesh-api v1.3.4 h1:0paB8GGyFofg094Cz4piv9KV/tMVDLNQKbEk0mN3iRI=
-github.com/megaease/easemesh-api v1.3.4/go.mod h1:VJWyuh/airQFJPJ9nfzAjholezslTHYZ70kU4Oq37MU=
+github.com/megaease/easemesh-api v1.3.5 h1:0MV1VVdiVZXqRUSt6rJdzI8+cAAPo6QAK9t6yy3KF+Q=
+github.com/megaease/easemesh-api v1.3.5/go.mod h1:VJWyuh/airQFJPJ9nfzAjholezslTHYZ70kU4Oq37MU=
 github.com/megaease/grace v1.0.0 h1:b44R3j6e/iaN62F4ZUnru9nzL1VaIcxxUZjSPVtTVzI=
 github.com/megaease/grace v1.0.0/go.mod h1:mOR6MVYQ6zGyuz9Y2or/VJ6QWueTL3erxWfIwyCmiIg=
 github.com/microcosm-cc/bluemonday v1.0.1/go.mod h1:hsXNsILzKxV+sX77C5b8FSuKF00vh2OMYv+xgHpAMF4=
diff --git a/pkg/context/httpcontext.go b/pkg/context/httpcontext.go
index 35ad9b01..efdaf6e8 100644
--- a/pkg/context/httpcontext.go
+++ b/pkg/context/httpcontext.go
@@ -57,8 +57,9 @@ type (
 		Cancelled() bool
 		ClientDisconnected() bool
 
-		OnFinish(FinishFunc) // For setting final client statistics, etc.
-		AddTag(tag string)   // For debug, log, etc.
+		OnFinish(FinishFunc)    // For setting final client statistics, etc.
+		AddTag(tag string)      // For debug, log, etc.
+		AddLazyTag(LazyTagFunc) // Return LazyTags as strings.
 
 		StatMetric() *httpstat.Metric
 
@@ -136,12 +137,16 @@ type (
 	// when body is flushing.
 	BodyFlushFunc = func(body []byte, complete bool) (newBody []byte)
 
+	// LazyTagFunc is the type of function to be called back
+	// when converting lazy tags to strings.
+	LazyTagFunc = func() string
+
 	httpContext struct {
 		mutex sync.Mutex
 
 		startTime   time.Time
 		finishFuncs []FinishFunc
-		tags        []string
+		lazyTags    []LazyTagFunc
 		caller      HandlerCaller
 
 		r *httpRequest
@@ -166,12 +171,13 @@ func New(stdw http.ResponseWriter, stdr *http.Request,
 	originalReqCtx := stdr.Context()
 	stdctx, cancelFunc := stdcontext.WithCancel(originalReqCtx)
 	stdr = stdr.WithContext(stdctx)
-
 	startTime := fasttime.Now()
 	span := tracing.NewSpanWithStart(tracer, spanName, startTime)
-	span.SetTag("http.method", stdr.Method)
-	span.SetTag("http.path", stdr.URL.Path)
-	return &httpContext{
+	if !span.IsNoopSpan() {
+		span.SetTag("http.method", stdr.Method)
+		span.SetTag("http.path", stdr.URL.Path)
+	}
+	ctx := &httpContext{
 		startTime:      startTime,
 		span:           span,
 		originalReqCtx: originalReqCtx,
@@ -179,7 +185,15 @@ func New(stdw http.ResponseWriter, stdr *http.Request,
 		cancelFunc:     cancelFunc,
 		r:              newHTTPRequest(stdr),
 		w:              newHTTPResponse(stdw, stdr),
+		lazyTags:       make([]LazyTagFunc, 0, 5),
+		finishFuncs:    make([]FinishFunc, 0, 1),
 	}
+	return ctx
+}
+
+// NewEmptyContext for testing.
+func NewEmptyContext() HTTPContext {
+	return &httpContext{}
 }
 
 func (ctx *httpContext) Protocol() Protocol {
@@ -206,8 +220,25 @@ func (ctx *httpContext) Span() tracing.Span {
 	return ctx.span
 }
 
+// Add new Tag.
 func (ctx *httpContext) AddTag(tag string) {
-	ctx.tags = append(ctx.tags, tag)
+	ctx.lazyTags = append(ctx.lazyTags, func() string {
+		return tag
+	})
+}
+
+// Add new LazyTag.
+func (ctx *httpContext) AddLazyTag(lazyTag LazyTagFunc) {
+	ctx.lazyTags = append(ctx.lazyTags, lazyTag)
+}
+
+// Return all tags in string format.
+func (ctx *httpContext) getTags() []string {
+	tags := make([]string, len(ctx.lazyTags))
+	for i := 0; i < len(ctx.lazyTags); i++ {
+		tags[i] = ctx.lazyTags[i]()
+	}
+	return tags
 }
 
 func (ctx *httpContext) Request() HTTPRequest {
@@ -288,6 +319,7 @@ func (ctx *httpContext) Finish() {
 
 	logger.LazyHTTPAccess(func() string {
 		stdr := ctx.r.std
+		tags := strings.Join(ctx.getTags(), " | ")
 
 		// log format:
 		// [startTime]
@@ -303,7 +335,7 @@ func (ctx *httpContext) Finish() {
 			fasttime.Format(ctx.startTime, fasttime.RFC3339Milli),
 			stdr.RemoteAddr, ctx.r.RealIP(), stdr.Method, stdr.RequestURI, stdr.Proto, ctx.w.code,
 			ctx.metric.Duration, ctx.r.Size(), ctx.w.Size(),
-			strings.Join(ctx.tags, " | "))
+			tags)
 	})
 }
 
diff --git a/pkg/filter/mock/mock.go b/pkg/filter/mock/mock.go
index 781e4c2a..77b3f349 100644
--- a/pkg/filter/mock/mock.go
+++ b/pkg/filter/mock/mock.go
@@ -24,6 +24,7 @@ import (
 	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/urlrule"
 )
 
 const (
@@ -44,8 +45,6 @@ type (
 	Mock struct {
 		filterSpec *httppipeline.FilterSpec
 		spec       *Spec
-
-		body []byte
 	}
 
 	// Spec describes the Mock.
@@ -55,15 +54,22 @@ type (
 
 	// Rule is the mock rule.
 	Rule struct {
-		Path       string            `yaml:"path,omitempty" jsonschema:"omitempty,pattern=^/"`
-		PathPrefix string            `yaml:"pathPrefix,omitempty" jsonschema:"omitempty,pattern=^/"`
-		Code       int               `yaml:"code" jsonschema:"required,format=httpcode"`
-		Headers    map[string]string `yaml:"headers" jsonschema:"omitempty"`
-		Body       string            `yaml:"body" jsonschema:"omitempty"`
-		Delay      string            `yaml:"delay" jsonschema:"omitempty,format=duration"`
+		Match   MatchRule         `yaml:"match" jsonschema:"required"`
+		Code    int               `yaml:"code" jsonschema:"required,format=httpcode"`
+		Headers map[string]string `yaml:"headers" jsonschema:"omitempty"`
+		Body    string            `yaml:"body" jsonschema:"omitempty"`
+		Delay   string            `yaml:"delay" jsonschema:"omitempty,format=duration"`
 
 		delay time.Duration
 	}
+
+	// MatchRule is the rule to match a request
+	MatchRule struct {
+		Path            string                          `yaml:"path,omitempty" jsonschema:"omitempty,pattern=^/"`
+		PathPrefix      string                          `yaml:"pathPrefix,omitempty" jsonschema:"omitempty,pattern=^/"`
+		Headers         map[string]*urlrule.StringMatch `yaml:"headers" jsonschema:"omitempty"`
+		MatchAllHeaders bool                            `yaml:"matchAllHeaders" jsonschema:"omitempty"`
+	}
 )
 
 // Kind returns the kind of Mock.
@@ -108,53 +114,100 @@ func (m *Mock) reload() {
 }
 
 // Handle mocks HTTPContext.
-func (m *Mock) Handle(ctx context.HTTPContext) (result string) {
-	result = m.handle(ctx)
+func (m *Mock) Handle(ctx context.HTTPContext) string {
+	result := ""
+	if rule := m.match(ctx); rule != nil {
+		m.mock(ctx, rule)
+		result = resultMocked
+	}
 	return ctx.CallNextHandler(result)
 }
 
-func (m *Mock) handle(ctx context.HTTPContext) (result string) {
+func (m *Mock) match(ctx context.HTTPContext) *Rule {
 	path := ctx.Request().Path()
-	w := ctx.Response()
+	header := ctx.Request().Header()
 
-	mock := func(rule *Rule) {
-		w.SetStatusCode(rule.Code)
-		for key, value := range rule.Headers {
-			w.Header().Set(key, value)
+	matchPath := func(rule *Rule) bool {
+		if rule.Match.Path == "" && rule.Match.PathPrefix == "" {
+			return true
 		}
-		w.SetBody(strings.NewReader(rule.Body))
-		result = resultMocked
 
-		if rule.delay <= 0 {
-			return
+		if rule.Match.Path == path {
+			return true
 		}
 
-		logger.Debugf("delay for %v ...", rule.delay)
-		select {
-		case <-ctx.Done():
-			logger.Debugf("request cancelled in the middle of delay mocking")
-		case <-time.After(rule.delay):
+		if rule.Match.PathPrefix == "" {
+			return false
 		}
+
+		return strings.HasPrefix(path, rule.Match.PathPrefix)
 	}
 
-	for _, rule := range m.spec.Rules {
-		if rule.Path == "" && rule.PathPrefix == "" {
-			mock(rule)
-			return
+	matchOneHeader := func(key string, rule *urlrule.StringMatch) bool {
+		values := header.GetAll(key)
+		if len(values) == 0 {
+			return rule.Empty
+		}
+		if rule.Empty {
+			return false
 		}
 
-		if rule.Path == path {
-			mock(rule)
-			return
+		for _, v := range values {
+			if rule.Match(v) {
+				return true
+			}
 		}
 
-		if rule.PathPrefix != "" && strings.HasPrefix(path, rule.PathPrefix) {
-			mock(rule)
-			return
+		return false
+	}
+
+	matchHeader := func(rule *Rule) bool {
+		if len(rule.Match.Headers) == 0 {
+			return true
 		}
+
+		for key, r := range rule.Match.Headers {
+			if matchOneHeader(key, r) {
+				if !rule.Match.MatchAllHeaders {
+					return true
+				}
+			} else {
+				if rule.Match.MatchAllHeaders {
+					return false
+				}
+			}
+		}
+
+		return rule.Match.MatchAllHeaders
 	}
 
-	return ""
+	for _, rule := range m.spec.Rules {
+		if matchPath(rule) && matchHeader(rule) {
+			return rule
+		}
+	}
+
+	return nil
+}
+
+func (m *Mock) mock(ctx context.HTTPContext, rule *Rule) {
+	w := ctx.Response()
+	w.SetStatusCode(rule.Code)
+	for key, value := range rule.Headers {
+		w.Header().Set(key, value)
+	}
+	w.SetBody(strings.NewReader(rule.Body))
+
+	if rule.delay <= 0 {
+		return
+	}
+
+	logger.Debugf("delay for %v ...", rule.delay)
+	select {
+	case <-ctx.Done():
+		logger.Debugf("request cancelled in the middle of delay mocking")
+	case <-time.After(rule.delay):
+	}
 }
 
 // Status returns status.
@@ -163,4 +216,5 @@ func (m *Mock) Status() interface{} {
 }
 
 // Close closes Mock.
-func (m *Mock) Close() {}
+func (m *Mock) Close() {
+}
diff --git a/pkg/filter/proxy/pool.go b/pkg/filter/proxy/pool.go
index 53cdbcfb..3127d975 100644
--- a/pkg/filter/proxy/pool.go
+++ b/pkg/filter/proxy/pool.go
@@ -22,8 +22,10 @@ import (
 	"io"
 	"net/http"
 	"strconv"
+	"sync"
 
 	"github.com/opentracing/opentracing-go"
+	gohttpstat "github.com/tcnksm/go-httpstat"
 
 	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
@@ -127,11 +129,32 @@ func (p *pool) status() *PoolStatus {
 	return s
 }
 
+var requestPool = sync.Pool{
+	New: func() interface{} {
+		return &request{}
+	},
+}
+
+var httpstatResultPool = sync.Pool{
+	New: func() interface{} {
+		return &gohttpstat.Result{}
+	},
+}
+
 func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.Client) string {
-	addTag := func(subPrefix, msg string) {
-		tag := stringtool.Cat(p.tagPrefix, "#", subPrefix, ": ", msg)
+	// intMsg is converted to string in AddLazyTag for better performance,
+	// as it is not run when access logs are disabled.
+	addLazyTag := func(subPrefix, msg string, intMsg int) {
 		ctx.Lock()
-		ctx.AddTag(tag)
+		if intMsg > -1 {
+			ctx.AddLazyTag(func() string {
+				return stringtool.Cat(p.tagPrefix, "#", subPrefix, ": ", strconv.Itoa(intMsg))
+			})
+		} else {
+			ctx.AddLazyTag(func() string {
+				return stringtool.Cat(p.tagPrefix, "#", subPrefix, ": ", msg)
+			})
+		}
 		ctx.Unlock()
 	}
 
@@ -143,17 +166,17 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.C
 
 	server, err := p.servers.next(ctx)
 	if err != nil {
-		addTag("serverErr", err.Error())
+		addLazyTag("serverErr", err.Error(), -1)
 		setStatusCode(http.StatusServiceUnavailable)
 		return resultInternalError
 	}
-	addTag("addr", server.URL)
+	addLazyTag("addr", server.URL, -1)
 
-	req, err := p.prepareRequest(ctx, server, reqBody)
+	req, err := p.prepareRequest(ctx, server, reqBody, requestPool, httpstatResultPool)
 	if err != nil {
 		msg := stringtool.Cat("prepare request failed: ", err.Error())
 		logger.Errorf("BUG: %s", msg)
-		addTag("bug", msg)
+		addLazyTag("bug", msg, -1)
 		setStatusCode(http.StatusInternalServerError)
 		return resultInternalError
 	}
@@ -163,8 +186,8 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.C
 		// NOTE: May add option to cancel the tracing if failed here.
 		// ctx.Span().Cancel()
 
-		addTag("doRequestErr", fmt.Sprintf("%v", err))
-		addTag("trace", req.detail())
+		addLazyTag("doRequestErr", fmt.Sprintf("%v", err), -1)
+		addLazyTag("trace", req.detail(), -1)
 		if ctx.ClientDisconnected() {
 			// NOTE: The HTTPContext will set 499 by itself if client is Disconnected.
 			// w.SetStatusCode((499)
@@ -175,7 +198,7 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.C
 		return resultServerError
 	}
 
-	addTag("code", strconv.Itoa(resp.StatusCode))
+	addLazyTag("code", "", resp.StatusCode)
 
 	ctx.Lock()
 	defer ctx.Unlock()
@@ -185,7 +208,7 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.C
 
 	if p.writeResponse {
 		ctx.Response().SetStatusCode(resp.StatusCode)
-		ctx.Response().Header().AddFromStd(resp.Header)
+		ctx.Response().Header().SetRaw(resp.Header)
 		ctx.Response().SetBody(respBody)
 
 		return ""
@@ -203,8 +226,13 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader, client *http.C
 	return ""
 }
 
-func (p *pool) prepareRequest(ctx context.HTTPContext, server *Server, reqBody io.Reader) (req *request, err error) {
-	return p.newRequest(ctx, server, reqBody)
+func (p *pool) prepareRequest(
+	ctx context.HTTPContext,
+	server *Server,
+	reqBody io.Reader,
+	requestPool sync.Pool,
+	httpstatResultPool sync.Pool) (req *request, err error) {
+	return p.newRequest(ctx, server, reqBody, requestPool, httpstatResultPool)
 }
 
 func (p *pool) doRequest(ctx context.HTTPContext, req *request, client *http.Client) (*http.Response, tracing.Span, error) {
@@ -225,6 +253,12 @@ func (p *pool) doRequest(ctx context.HTTPContext, req *request, client *http.Cli
 	return resp, span, nil
 }
 
+var httpstatMetricPool = sync.Pool{
+	New: func() interface{} {
+		return &httpstat.Metric{}
+	},
+}
+
 func (p *pool) statRequestResponse(ctx context.HTTPContext,
 	req *request, resp *http.Response, span tracing.Span) io.Reader {
 
@@ -246,19 +280,25 @@ func (p *pool) statRequestResponse(ctx context.HTTPContext,
 			req.finish()
 			span.Finish()
 		}
+		duration := req.total()
+		ctx.AddLazyTag(func() string {
+			return stringtool.Cat(p.tagPrefix, "#duration: ", duration.String())
+		})
+		// use recycled object
+		metric := httpstatMetricPool.Get().(*httpstat.Metric)
+		metric.StatusCode = resp.StatusCode
+		metric.Duration = duration
+		metric.ReqSize = ctx.Request().Size()
+		metric.RespSize = uint64(responseMetaSize(resp) + count)
 
-		ctx.AddTag(stringtool.Cat(p.tagPrefix, "#duration: ", req.total().String()))
-
-		metric := &httpstat.Metric{
-			StatusCode: resp.StatusCode,
-			Duration:   req.total(),
-			ReqSize:    ctx.Request().Size(),
-			RespSize:   uint64(responseMetaSize(resp) + count),
-		}
 		if !p.writeResponse {
 			metric.RespSize = 0
 		}
 		p.httpStat.Stat(metric)
+		// recycle struct instances
+		httpstatMetricPool.Put(metric)
+		httpstatResultPool.Put(req.statResult)
+		requestPool.Put(req)
 	})
 
 	return callbackBody
diff --git a/pkg/filter/proxy/request.go b/pkg/filter/proxy/request.go
index a2e2b09e..b0ea6e3e 100644
--- a/pkg/filter/proxy/request.go
+++ b/pkg/filter/proxy/request.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	httpstat "github.com/tcnksm/go-httpstat"
@@ -37,8 +38,8 @@ type (
 		std        *http.Request
 		statResult *httpstat.Result
 		createTime time.Time
-		_startTime *time.Time
-		_endTime   *time.Time
+		_startTime time.Time
+		_endTime   time.Time
 	}
 
 	resultState struct {
@@ -46,12 +47,19 @@ type (
 	}
 )
 
-func (p *pool) newRequest(ctx context.HTTPContext, server *Server, reqBody io.Reader) (*request, error) {
-	req := &request{
-		createTime: fasttime.Now(),
-		server:     server,
-		statResult: &httpstat.Result{},
-	}
+func (p *pool) newRequest(
+	ctx context.HTTPContext,
+	server *Server,
+	reqBody io.Reader,
+	requestPool sync.Pool,
+	httpstatResultPool sync.Pool) (*request, error) {
+	statResult := httpstatResultPool.Get().(*httpstat.Result)
+	req := requestPool.Get().(*request)
+	req.createTime = fasttime.Now()
+	req.server = server
+	req.statResult = statResult
+	req._startTime = time.Time{}
+	req._endTime = time.Time{}
 
 	r := ctx.Request()
 
@@ -75,49 +83,48 @@ func (p *pool) newRequest(ctx context.HTTPContext, server *Server, reqBody io.Re
 }
 
 func (r *request) start() {
-	if r._startTime != nil {
+	if !time.Time.IsZero(r._startTime) {
 		logger.Errorf("BUG: started already")
 		return
 	}
 
-	now := fasttime.Now()
-	r._startTime = &now
+	r._startTime = fasttime.Now()
 }
 
 func (r *request) startTime() time.Time {
-	if r._startTime == nil {
+	if time.Time.IsZero(r._startTime) {
 		return r.createTime
 	}
 
-	return *r._startTime
+	return r._startTime
 }
 
 func (r *request) endTime() time.Time {
-	if r._endTime == nil {
+	if time.Time.IsZero(r._endTime) {
 		return fasttime.Now()
 	}
 
-	return *r._endTime
+	return r._endTime
 }
 
 func (r *request) finish() {
-	if r._endTime != nil {
+	if !time.Time.IsZero(r._endTime) {
 		logger.Errorf("BUG: finished already")
 		return
 	}
 
 	now := fasttime.Now()
 	r.statResult.End(now)
-	r._endTime = &now
+	r._endTime = now
 }
 
 func (r *request) total() time.Duration {
-	if r._endTime == nil {
+	if time.Time.IsZero(r._endTime) {
 		logger.Errorf("BUG: call total before finish")
 		return r.statResult.Total(fasttime.Now())
 	}
 
-	return r.statResult.Total(*r._endTime)
+	return r.statResult.Total(r._endTime)
 }
 
 func (r *request) detail() string {
diff --git a/pkg/filter/validator/validator.go b/pkg/filter/validator/validator.go
index 01fbc13e..c59636cf 100644
--- a/pkg/filter/validator/validator.go
+++ b/pkg/filter/validator/validator.go
@@ -132,7 +132,7 @@ func (v *Validator) handle(ctx context.HTTPContext) string {
 	if v.jwt != nil {
 		err := v.jwt.Validate(req)
 		if err != nil {
-			ctx.Response().SetStatusCode(http.StatusForbidden)
+			ctx.Response().SetStatusCode(http.StatusUnauthorized)
 			ctx.AddTag(stringtool.Cat("JWT validator: ", err.Error()))
 			return resultInvalid
 		}
@@ -141,7 +141,7 @@ func (v *Validator) handle(ctx context.HTTPContext) string {
 	if v.signer != nil {
 		err := v.signer.Verify(req.Std())
 		if err != nil {
-			ctx.Response().SetStatusCode(http.StatusForbidden)
+			ctx.Response().SetStatusCode(http.StatusUnauthorized)
 			ctx.AddTag(stringtool.Cat("signature validator: ", err.Error()))
 			return resultInvalid
 		}
@@ -150,7 +150,7 @@ func (v *Validator) handle(ctx context.HTTPContext) string {
 	if v.oauth2 != nil {
 		err := v.oauth2.Validate(req)
 		if err != nil {
-			ctx.Response().SetStatusCode(http.StatusForbidden)
+			ctx.Response().SetStatusCode(http.StatusUnauthorized)
 			ctx.AddTag(stringtool.Cat("oauth2 validator: ", err.Error()))
 			return resultInvalid
 		}
diff --git a/pkg/object/autocertmanager/autocertmanager.go b/pkg/object/autocertmanager/autocertmanager.go
index 50074615..6cb820e4 100644
--- a/pkg/object/autocertmanager/autocertmanager.go
+++ b/pkg/object/autocertmanager/autocertmanager.go
@@ -373,7 +373,10 @@ func (acm *AutoCertManager) getCertificate(chi *tls.ClientHelloInfo, tokenOnly b
 
 	domain := acm.findDomain(name, false)
 	if domain == nil {
-		return nil, fmt.Errorf("host %q is not configured for auto cert", name)
+		// return nil error if the domain is not managed by the AutoCertManager, so that
+		// the Go HTTP package could check the the static certificates configured in the
+		// HTTP server spec.
+		return nil, nil
 	}
 
 	cert := domain.cert()
@@ -414,7 +417,11 @@ func GetCertificate(chi *tls.ClientHelloInfo, tokenOnly bool) (*tls.Certificate,
 	if acm != nil {
 		return acm.getCertificate(chi, tokenOnly)
 	}
-	return nil, fmt.Errorf("auto certificate manager is not started")
+
+	// return a nil error if the AutoCertManager is not started, otherwise:
+	// * static certificates configured in an HTTP server are never used, which is a bug
+	// * the Go HTTP package logs a lot of 'TLS handshake error'
+	return nil, nil
 }
 
 // HandleHTTP01Challenge handles HTTP-01 challenge
diff --git a/pkg/object/httpserver/spec.go b/pkg/object/httpserver/spec.go
index 153d06a4..a1a822a2 100644
--- a/pkg/object/httpserver/spec.go
+++ b/pkg/object/httpserver/spec.go
@@ -115,6 +115,17 @@ func (spec *Spec) Validate() error {
 	return err
 }
 
+func tryDecodeBase64Pem(pem string) []byte {
+	// The pem could in base64 encoding or plain text. It starts with '-' if it is
+	// in plain text, and '-' is not a valid character in standard base64 encoding.
+	// We first try to decode it as base64, and fallback to plain text if failed.
+	d, err := base64.StdEncoding.DecodeString(pem)
+	if err == nil {
+		return d
+	}
+	return []byte(pem)
+}
+
 func (spec *Spec) tlsConfig() (*tls.Config, error) {
 	var certificates []tls.Certificate
 
@@ -135,8 +146,8 @@ func (spec *Spec) tlsConfig() (*tls.Config, error) {
 			return nil, fmt.Errorf("certs %s hasn't secret corresponded to it", k)
 		}
 
-		certPem, _ := base64.StdEncoding.DecodeString(v)
-		keyPem, _ := base64.StdEncoding.DecodeString(secret)
+		certPem := tryDecodeBase64Pem(v)
+		keyPem := tryDecodeBase64Pem(secret)
 		cert, err := tls.X509KeyPair(certPem, keyPem)
 		if err != nil {
 			return nil, fmt.Errorf("generate x509 key pair for %s failed: %s ", k, err)
diff --git a/pkg/object/meshcontroller/api/api_customresource.go b/pkg/object/meshcontroller/api/api_customresource.go
index 755f6914..c04c96d3 100644
--- a/pkg/object/meshcontroller/api/api_customresource.go
+++ b/pkg/object/meshcontroller/api/api_customresource.go
@@ -56,12 +56,11 @@ func (a *API) listCustomResourceKinds(w http.ResponseWriter, r *http.Request) {
 		pbKinds = append(pbKinds, kind)
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err := json.NewEncoder(w).Encode(pbKinds)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", kinds, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) getCustomResourceKind(w http.ResponseWriter, r *http.Request) {
@@ -83,12 +82,11 @@ func (a *API) getCustomResourceKind(w http.ResponseWriter, r *http.Request) {
 		panic(fmt.Errorf("convert spec %#v to pb failed: %v", kind, err))
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err = json.NewEncoder(w).Encode(pbKind)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", pbKind, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) saveCustomResourceKind(w http.ResponseWriter, r *http.Request, update bool) error {
@@ -175,12 +173,11 @@ func (a *API) listAllCustomResources(w http.ResponseWriter, r *http.Request) {
 		return resources[i].Name() < resources[j].Name()
 	})
 
+	w.Header().Set("Content-Type", "application/json")
 	err := json.NewEncoder(w).Encode(resources)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", resources, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) listCustomResources(w http.ResponseWriter, r *http.Request) {
@@ -200,12 +197,11 @@ func (a *API) listCustomResources(w http.ResponseWriter, r *http.Request) {
 		return resources[i].Name() < resources[j].Name()
 	})
 
+	w.Header().Set("Content-Type", "application/json")
 	err = json.NewEncoder(w).Encode(resources)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", resources, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) getCustomResource(w http.ResponseWriter, r *http.Request) {
@@ -231,12 +227,11 @@ func (a *API) getCustomResource(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err = json.NewEncoder(w).Encode(resource)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", resource, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) saveCustomResource(w http.ResponseWriter, r *http.Request, update bool) error {
diff --git a/pkg/object/meshcontroller/api/api_trafficcontrol.go b/pkg/object/meshcontroller/api/api_trafficcontrol.go
index 4c7ec409..12ff8254 100644
--- a/pkg/object/meshcontroller/api/api_trafficcontrol.go
+++ b/pkg/object/meshcontroller/api/api_trafficcontrol.go
@@ -46,12 +46,11 @@ func (a *API) listHTTPRouteGroups(w http.ResponseWriter, r *http.Request) {
 		pbGroups = append(pbGroups, group)
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err := json.NewEncoder(w).Encode(pbGroups)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", groups, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) getHTTPRouteGroup(w http.ResponseWriter, r *http.Request) {
@@ -73,12 +72,11 @@ func (a *API) getHTTPRouteGroup(w http.ResponseWriter, r *http.Request) {
 		panic(fmt.Errorf("convert spec %#v to pb failed: %v", group, err))
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err = json.NewEncoder(w).Encode(pbGroup)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", pbGroup, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) saveHTTPRouteGroup(w http.ResponseWriter, r *http.Request, update bool) error {
@@ -160,12 +158,11 @@ func (a *API) listTrafficTargets(w http.ResponseWriter, r *http.Request) {
 		pbTrafficTargets = append(pbTrafficTargets, tt)
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err := json.NewEncoder(w).Encode(pbTrafficTargets)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", tts, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) getTrafficTarget(w http.ResponseWriter, r *http.Request) {
@@ -187,12 +184,11 @@ func (a *API) getTrafficTarget(w http.ResponseWriter, r *http.Request) {
 		panic(fmt.Errorf("convert spec %#v to pb failed: %v", tt, err))
 	}
 
+	w.Header().Set("Content-Type", "application/json")
 	err = json.NewEncoder(w).Encode(pbTrafficTarget)
 	if err != nil {
 		panic(fmt.Errorf("marshal %#v to json failed: %v", pbTrafficTarget, err))
 	}
-
-	w.Header().Set("Content-Type", "application/json")
 }
 
 func (a *API) saveTrafficTarget(w http.ResponseWriter, r *http.Request, update bool) error {
diff --git a/pkg/object/meshcontroller/ingresscontroller/ingresscontroller.go b/pkg/object/meshcontroller/ingresscontroller/ingresscontroller.go
index 7b4b7934..e7eec7fe 100644
--- a/pkg/object/meshcontroller/ingresscontroller/ingresscontroller.go
+++ b/pkg/object/meshcontroller/ingresscontroller/ingresscontroller.go
@@ -210,7 +210,7 @@ func (ic *IngressController) _reloadIngress() {
 				serviceSpec := &spec.Service{
 					Name: path.Backend,
 				}
-				path.Backend = serviceSpec.IngressPipelineName()
+				path.Backend = serviceSpec.IngressControllerPipelineName()
 			}
 
 			ingressRules = append(ingressRules, rule)
diff --git a/pkg/object/meshcontroller/master/master.go b/pkg/object/meshcontroller/master/master.go
index 54fe19d5..3be39c30 100644
--- a/pkg/object/meshcontroller/master/master.go
+++ b/pkg/object/meshcontroller/master/master.go
@@ -18,9 +18,7 @@
 package master
 
 import (
-	"fmt"
 	"runtime/debug"
-	"strings"
 	"time"
 
 	"gopkg.in/yaml.v2"
@@ -36,18 +34,17 @@ import (
 )
 
 const (
-	defaultCleanInterval       time.Duration = 10 * time.Minute
 	defaultDeadRecordExistTime time.Duration = 20 * time.Minute
 )
 
 type (
 	// Master is the master role of Easegress for mesh control plane.
 	Master struct {
-		super               *supervisor.Supervisor
-		superSpec           *supervisor.Spec
-		spec                *spec.Admin
-		maxHeartbeatTimeout time.Duration
-		certMananger        *certmanager.CertManager
+		super             *supervisor.Supervisor
+		superSpec         *supervisor.Spec
+		spec              *spec.Admin
+		heartbeatInterval time.Duration
+		certMananger      *certmanager.CertManager
 
 		registrySyncer *registrySyncer
 		store          storage.Storage
@@ -56,21 +53,10 @@ type (
 		done chan struct{}
 	}
 
-	serviceInstances []*spec.ServiceInstanceSpec
-
 	// Status is the status of mesh master.
 	Status struct{}
 )
 
-func (s serviceInstances) String() string {
-	instances := make([]string, len(s))
-	for i, instance := range s {
-		instances[i] = fmt.Sprintf("%s/%s", instance.ServiceName, instance.InstanceID)
-	}
-
-	return fmt.Sprintf("[%s]", strings.Join(instances, ","))
-}
-
 // New creates a mesh master.
 func New(superSpec *supervisor.Spec) *Master {
 	store := storage.New(superSpec.Name(), superSpec.Super().Cluster())
@@ -89,17 +75,19 @@ func New(superSpec *supervisor.Spec) *Master {
 
 	heartbeat, err := time.ParseDuration(m.spec.HeartbeatInterval)
 	if err != nil {
-		logger.Errorf("BUG: parse heartbeat interval %s to duration failed: %v",
-			m.spec.HeartbeatInterval, err)
+		format := "failed to parse heartbeat interval '%s', fallback to default(5s)"
+		logger.Errorf(format, m.spec.HeartbeatInterval)
+		heartbeat = 5 * time.Second
 	}
-	m.maxHeartbeatTimeout = heartbeat * 2
+	m.heartbeatInterval = heartbeat
 
-	m.run()
+	m.initMTLS()
+	go m.run()
 
 	return m
 }
 
-func (m *Master) securityRoutine() error {
+func (m *Master) initMTLS() error {
 	if !m.spec.EnablemTLS() {
 		return nil
 	}
@@ -120,150 +108,108 @@ func (m *Master) securityRoutine() error {
 }
 
 func (m *Master) run() {
-	watchInterval, err := time.ParseDuration(m.spec.HeartbeatInterval)
-	if err != nil {
-		logger.Errorf("BUG: parse duration %s failed: %v",
-			m.spec.HeartbeatInterval, err)
-		return
-	}
-
-	if err := m.securityRoutine(); err != nil {
-		logger.Errorf("start security routine failed, err: %v", err)
-		return
-	}
-
-	go m.checkHeartbeat(watchInterval)
-	go m.clean()
-
-}
-
-// only handle master routines when its the cluster leader.
-func (m *Master) needHandle() bool {
-	return m.superSpec.Super().Cluster().IsLeader()
-}
+	ticker := time.NewTicker(m.heartbeatInterval)
 
-func (m *Master) checkHeartbeat(watchInterval time.Duration) {
 	for {
 		select {
 		case <-m.done:
+			ticker.Stop()
 			return
-		case <-time.After(watchInterval):
+
+		case <-ticker.C:
 			if m.needHandle() {
-				func() {
-					defer func() {
-						if err := recover(); err != nil {
-							logger.Errorf("failed to check instance heartbeat %v, stack trace: \n%s\n",
-								err, debug.Stack())
-						}
-					}()
-					m.checkInstancesHeartbeat()
-				}()
+				m.checkServiceInstances()
 			}
 		}
 	}
 }
 
-func (m *Master) clean() {
-	for {
-		select {
-		case <-m.done:
-			return
-		case <-time.After(defaultCleanInterval):
-			if m.needHandle() {
-				func() {
-					defer func() {
-						if err := recover(); err != nil {
-							logger.Errorf("failed to clean dead instances: %v, stack trace: \n%s\n",
-								err, debug.Stack())
-						}
-					}()
-					m.cleanDeadInstances()
-				}()
-			}
-		}
-	}
+// only handle master routines when its the cluster leader.
+func (m *Master) needHandle() bool {
+	return m.superSpec.Super().Cluster().IsLeader()
 }
 
-func (m *Master) scanInstances() (failedInstances []*spec.ServiceInstanceSpec,
-	rebornInstances []*spec.ServiceInstanceSpec, deadInstances []*spec.ServiceInstanceSpec) {
+func (m *Master) checkServiceInstances() {
+	defer func() {
+		if err := recover(); err != nil {
+			format := "failed to check instances %v, stack trace: \n%s\n"
+			logger.Errorf(format, err, debug.Stack())
+		}
+	}()
 
 	statuses := m.service.ListAllServiceInstanceStatuses()
 	specs := m.service.ListAllServiceInstanceSpecs()
 
-	now := time.Now()
 	for _, _spec := range specs {
 		if !m.isMeshRegistryName(_spec.RegistryName) {
 			continue
 		}
 
+		// TODO: improve search performance
 		var status *spec.ServiceInstanceStatus
 		for _, s := range statuses {
 			if s.ServiceName == _spec.ServiceName && s.InstanceID == _spec.InstanceID {
 				status = s
 			}
 		}
-		if status != nil {
-			lastHeartbeatTime, err := time.Parse(time.RFC3339, status.LastHeartbeatTime)
-			if err != nil {
-				logger.Errorf("BUG: parse last heartbeat time %s failed: %v", status.LastHeartbeatTime, err)
-				continue
-			}
-			gap := now.Sub(lastHeartbeatTime)
-			if gap > m.maxHeartbeatTimeout {
-				if _spec.Status != spec.ServiceStatusOutOfService {
-					failedInstances = append(failedInstances, _spec)
-				}
-
-				if gap > defaultDeadRecordExistTime {
-					deadInstances = append(deadInstances, _spec)
-				}
-			} else {
-				if _spec.Status == spec.ServiceStatusOutOfService {
-					rebornInstances = append(rebornInstances, _spec)
-				}
-			}
-		} else {
-			failedInstances = append(failedInstances, _spec)
-			deadInstances = append(deadInstances, _spec)
+
+		if status == nil {
+			format := "status of %s/%s not found, need to delete"
+			logger.Warnf(format, _spec.ServiceName, _spec.InstanceID)
+			m.deleteInstance(_spec)
+			continue
 		}
-	}
 
-	if len(failedInstances) > 0 {
-		logger.Warnf("failed instances: %s", serviceInstances(failedInstances))
+		m.checkLastHeartbeatTime(_spec, status.LastHeartbeatTime)
 	}
-	if len(deadInstances) > 0 {
-		logger.Warnf("dead instances: %s", serviceInstances(deadInstances))
+}
+
+func (m *Master) checkLastHeartbeatTime(_spec *spec.ServiceInstanceSpec, lastHeartbeatTime string) {
+	t, err := time.Parse(time.RFC3339, lastHeartbeatTime)
+	if err != nil {
+		logger.Errorf("BUG: parse last heartbeat time %s failed: %v", lastHeartbeatTime, err)
+		return
 	}
-	if len(rebornInstances) > 0 {
-		logger.Infof("reborn instances: %s", serviceInstances(rebornInstances))
+	gap := time.Since(t)
+
+	// This instance record's time gap is beyond our tolerance, needs to be clean immediately.
+	// For freeing storage space
+	if gap > defaultDeadRecordExistTime {
+		format := "%s/%s expired for %s, need to be deleted"
+		logger.Warnf(format, _spec.ServiceName, _spec.InstanceID, gap.String())
+		m.deleteInstance(_spec)
+		return
 	}
 
-	return
-}
+	// Bring it down if no heartbeat in 2 heartbeat interval
+	if gap > m.heartbeatInterval*2 {
+		if _spec.Status != spec.ServiceStatusOutOfService {
+			logger.Warnf("%s/%s expired for %s", _spec.ServiceName, _spec.InstanceID, gap.String())
+			m.updateInstanceStatus(_spec, spec.ServiceStatusOutOfService)
+		}
+		return
+	}
 
-func (m *Master) checkInstancesHeartbeat() {
-	failedInstances, rebornInstances, _ := m.scanInstances()
-	m.handleFailedInstances(failedInstances)
-	m.handleRebornInstances(rebornInstances)
+	if _spec.Status == spec.ServiceStatusOutOfService {
+		logger.Infof("%s/%s heartbeat recovered, make it UP", _spec.ServiceName, _spec.InstanceID)
+		m.updateInstanceStatus(_spec, spec.ServiceStatusUp)
+	}
 }
 
-func (m *Master) cleanDeadInstances() {
-	_, _, deadInstances := m.scanInstances()
-	for _, _spec := range deadInstances {
-		specKey := layout.ServiceInstanceSpecKey(_spec.ServiceName, _spec.InstanceID)
-		err := m.store.Delete(specKey)
-		if err != nil {
-			api.ClusterPanic(err)
-		} else {
-			logger.Infof("clean instance spec: %s", specKey)
-		}
+func (m *Master) deleteInstance(_spec *spec.ServiceInstanceSpec) {
+	specKey := layout.ServiceInstanceSpecKey(_spec.ServiceName, _spec.InstanceID)
+	err := m.store.Delete(specKey)
+	if err != nil {
+		api.ClusterPanic(err)
+	} else {
+		logger.Infof("clean instance spec: %s", specKey)
+	}
 
-		statusKey := layout.ServiceInstanceStatusKey(_spec.ServiceName, _spec.InstanceID)
-		if err = m.store.Delete(statusKey); err != nil {
-			api.ClusterPanic(err)
-		} else {
-			logger.Infof("clean instance status: %s", statusKey)
-		}
+	statusKey := layout.ServiceInstanceStatusKey(_spec.ServiceName, _spec.InstanceID)
+	if err = m.store.Delete(statusKey); err != nil {
+		api.ClusterPanic(err)
+	} else {
+		logger.Infof("clean instance status: %s", statusKey)
 	}
 }
 
@@ -277,29 +223,19 @@ func (m *Master) isMeshRegistryName(registryName string) bool {
 	}
 }
 
-func (m *Master) handleRebornInstances(rebornInstances []*spec.ServiceInstanceSpec) {
-	m.updateInstanceStatus(rebornInstances, spec.ServiceStatusUp)
-}
-
-func (m *Master) handleFailedInstances(failedInstances []*spec.ServiceInstanceSpec) {
-	m.updateInstanceStatus(failedInstances, spec.ServiceStatusOutOfService)
-}
-
-func (m *Master) updateInstanceStatus(instances []*spec.ServiceInstanceSpec, status string) {
-	for _, _spec := range instances {
-		_spec.Status = status
+func (m *Master) updateInstanceStatus(_spec *spec.ServiceInstanceSpec, status string) {
+	_spec.Status = status
 
-		buff, err := yaml.Marshal(_spec)
-		if err != nil {
-			logger.Errorf("BUG: marshal %#v to yaml failed: %v", _spec, err)
-			continue
-		}
+	buff, err := yaml.Marshal(_spec)
+	if err != nil {
+		logger.Errorf("BUG: marshal %#v to yaml failed: %v", _spec, err)
+		return
+	}
 
-		key := layout.ServiceInstanceSpecKey(_spec.ServiceName, _spec.InstanceID)
-		err = m.store.Put(key, string(buff))
-		if err != nil {
-			api.ClusterPanic(err)
-		}
+	key := layout.ServiceInstanceSpecKey(_spec.ServiceName, _spec.InstanceID)
+	err = m.store.Put(key, string(buff))
+	if err != nil {
+		api.ClusterPanic(err)
 	}
 }
 
diff --git a/pkg/object/meshcontroller/spec/spec.go b/pkg/object/meshcontroller/spec/spec.go
index 9495e319..18c9d0c6 100644
--- a/pkg/object/meshcontroller/spec/spec.go
+++ b/pkg/object/meshcontroller/spec/spec.go
@@ -202,6 +202,12 @@ type (
 		ServiceHeaders map[string][]string `yaml:"serviceHeaders" jsonschema:"omitempty"`
 	}
 
+	// GlobalTransmission is the spec of global transmission data.
+	// All endpoints of mesh should pass them.
+	GlobalTransmission struct {
+		Headers []string `yaml:"headers" jsonschema:"omitempty"`
+	}
+
 	// LoadBalance is the spec of service load balance.
 	LoadBalance = proxy.LoadBalance
 
@@ -886,7 +892,7 @@ rules:`
 func (s *Service) IngressControllerPipelineSpec(instanceSpecs []*ServiceInstanceSpec,
 	canaries []*ServiceCanary, cert, rootCert *Certificate) (*supervisor.Spec, error) {
 
-	pipelineSpecBuilder := newPipelineSpecBuilder(s.IngressPipelineName())
+	pipelineSpecBuilder := newPipelineSpecBuilder(s.IngressControllerPipelineName())
 
 	pipelineSpecBuilder.appendMeshAdaptor(canaries)
 	pipelineSpecBuilder.appendProxyWithCanary(instanceSpecs, canaries, s.LoadBalance, cert, rootCert)
@@ -984,11 +990,16 @@ func (s *Service) IngressHandlerName() string {
 	return fmt.Sprintf("mesh-ingress-handler-%s", s.Name)
 }
 
-// IngressPipelineName returns the ingress pipeline name
-func (s *Service) IngressPipelineName() string {
+// SidecarIngressPipelineName returns the ingress pipeline name
+func (s *Service) SidecarIngressPipelineName() string {
 	return fmt.Sprintf("mesh-ingress-pipeline-%s", s.Name)
 }
 
+// IngressControllerPipelineName returns the ingress pipeline name
+func (s *Service) IngressControllerPipelineName() string {
+	return fmt.Sprintf("mesh-ingresscontroller-pipeline-%s", s.Name)
+}
+
 // BackendName returns backend service name
 func (s *Service) BackendName() string {
 	return s.Name
@@ -1034,7 +1045,7 @@ func (s *Service) SidecarIngressPipelineSpec(applicationPort uint32) (*superviso
 		},
 	}
 
-	pipelineSpecBuilder := newPipelineSpecBuilder(s.IngressPipelineName())
+	pipelineSpecBuilder := newPipelineSpecBuilder(s.SidecarIngressPipelineName())
 
 	if s.Resilience != nil {
 		pipelineSpecBuilder.appendRateLimiter(s.Resilience.RateLimiter)
@@ -1066,15 +1077,16 @@ func (s *Service) SidecarEgressPipelineSpec(instanceSpecs []*ServiceInstanceSpec
 
 	if !s.Runnable() {
 		pipelineSpecBuilder.appendMock(s.Mock.Rules)
-	} else {
-		if s.Resilience != nil {
-			pipelineSpecBuilder.appendTimeLimiter(s.Resilience.TimeLimiter)
-			pipelineSpecBuilder.appendRetryer(s.Resilience.Retryer)
-			pipelineSpecBuilder.appendCircuitBreaker(s.Resilience.CircuitBreaker)
-		}
+	}
 
-		pipelineSpecBuilder.appendProxyWithCanary(instanceSpecs, canaries, s.LoadBalance, appCert, rootCert)
+	if s.Resilience != nil {
+		pipelineSpecBuilder.appendTimeLimiter(s.Resilience.TimeLimiter)
+		pipelineSpecBuilder.appendRetryer(s.Resilience.Retryer)
+		pipelineSpecBuilder.appendCircuitBreaker(s.Resilience.CircuitBreaker)
 	}
+
+	pipelineSpecBuilder.appendProxyWithCanary(instanceSpecs, canaries, s.LoadBalance, appCert, rootCert)
+
 	yamlConfig := pipelineSpecBuilder.yamlConfig()
 	superSpec, err := supervisor.NewSpec(yamlConfig)
 	if err != nil {
diff --git a/pkg/object/meshcontroller/worker/ingress.go b/pkg/object/meshcontroller/worker/ingress.go
index 35360028..a08f410d 100644
--- a/pkg/object/meshcontroller/worker/ingress.go
+++ b/pkg/object/meshcontroller/worker/ingress.go
@@ -99,7 +99,7 @@ func (ings *IngressServer) _ready() bool {
 		Name: ings.serviceName,
 	}
 
-	_, pipelineReady := ings.pipelines[serviceSpec.IngressPipelineName()]
+	_, pipelineReady := ings.pipelines[serviceSpec.SidecarIngressPipelineName()]
 
 	return pipelineReady && (ings.httpServer != nil)
 }
@@ -111,7 +111,7 @@ func (ings *IngressServer) InitIngress(service *spec.Service, port uint32) error
 
 	ings.applicationPort = port
 
-	if _, ok := ings.pipelines[service.IngressPipelineName()]; !ok {
+	if _, ok := ings.pipelines[service.SidecarIngressPipelineName()]; !ok {
 		superSpec, err := service.SidecarIngressPipelineSpec(port)
 		if err != nil {
 			return err
@@ -120,7 +120,7 @@ func (ings *IngressServer) InitIngress(service *spec.Service, port uint32) error
 		if err != nil {
 			return fmt.Errorf("create http pipeline %s failed: %v", superSpec.Name(), err)
 		}
-		ings.pipelines[service.IngressPipelineName()] = entity
+		ings.pipelines[service.SidecarIngressPipelineName()] = entity
 	}
 
 	admSpec := ings.superSpec.ObjectSpec().(*spec.Admin)
diff --git a/pkg/object/meshcontroller/worker/observability.go b/pkg/object/meshcontroller/worker/observability.go
index ef123a3f..ba9ffa43 100644
--- a/pkg/object/meshcontroller/worker/observability.go
+++ b/pkg/object/meshcontroller/worker/observability.go
@@ -57,9 +57,9 @@ func (server *ObservabilityManager) UpdateService(newService *spec.Service, vers
 	return nil
 }
 
-// UpdateCanary updates canary.
-func (server *ObservabilityManager) UpdateCanary(globalHeaders *spec.GlobalCanaryHeaders, version int64) error {
-	err := server.agentClient.UpdateCanary(globalHeaders, version)
+// UpdateGlobalTransmission updates global transmission.
+func (server *ObservabilityManager) UpdateGlobalTransmission(transmission *spec.GlobalTransmission) error {
+	err := server.agentClient.UpdateGlobalTransmission(transmission)
 	if err != nil {
 		return fmt.Errorf("Update Canary Spec: %v ", err)
 	}
diff --git a/pkg/object/meshcontroller/worker/worker.go b/pkg/object/meshcontroller/worker/worker.go
index b302d5e6..8dbabe7d 100644
--- a/pkg/object/meshcontroller/worker/worker.go
+++ b/pkg/object/meshcontroller/worker/worker.go
@@ -23,6 +23,7 @@ import (
 	"net/url"
 	"os"
 	"runtime/debug"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -293,20 +294,32 @@ func (worker *Worker) pushSpecToJavaAgent() {
 			logger.Errorf("update service %s failed: %v", serviceSpec.Name, err)
 		}
 
-		globalCanaryHeaders, info := worker.service.GetGlobalCanaryHeadersWithInfo()
-		if globalCanaryHeaders != nil {
-			err := worker.observabilityManager.UpdateCanary(globalCanaryHeaders, info.Version)
-			if err != nil {
-				logger.Errorf("update canary failed: %v", err)
+		canaries := worker.service.ListServiceCanaries()
+		transmission := &spec.GlobalTransmission{}
+
+		headersMap := map[string]struct{}{spec.ServiceCanaryHeaderKey: {}}
+		for _, canary := range canaries {
+			for key := range canary.TrafficRules.Headers {
+				headersMap[key] = struct{}{}
 			}
 		}
+		for key := range headersMap {
+			transmission.Headers = append(transmission.Headers, key)
+		}
+
+		sort.Strings(transmission.Headers)
+
+		err = worker.observabilityManager.UpdateGlobalTransmission(transmission)
+		if err != nil {
+			logger.Errorf("update canary failed: %v", err)
+		}
 	}
 
 	for {
 		select {
 		case <-worker.done:
 			return
-		case <-time.After(1 * time.Minute):
+		case <-time.After(30 * time.Second):
 			routine()
 		}
 	}
diff --git a/pkg/object/mqttproxy/broker.go b/pkg/object/mqttproxy/broker.go
index 9a2ff9aa..131314a2 100644
--- a/pkg/object/mqttproxy/broker.go
+++ b/pkg/object/mqttproxy/broker.go
@@ -552,6 +552,7 @@ func (b *Broker) httpGetAllSessionHandler(w http.ResponseWriter, r *http.Request
 	}
 
 	allSession, err := b.sessMgr.store.getPrefix(sessionStoreKey(""), false)
+	logger.SpanDebugf(span, "httpGetAllSessionHandler current total %v sessions, query %v, topic %v", len(allSession), []int{page, pageSize}, topic)
 	if err != nil {
 		logger.SpanErrorf(span, "get all sessions with prefix %v failed, %v", sessionStoreKey(""), err)
 		api.HandleAPIError(w, r, http.StatusInternalServerError, fmt.Errorf("get all sessions failed, %v", err))
diff --git a/pkg/option/option.go b/pkg/option/option.go
index 40c3fc8d..7a8ef5c4 100644
--- a/pkg/option/option.go
+++ b/pkg/option/option.go
@@ -112,16 +112,16 @@ func addClusterVars(opt *Options) {
 	// Cluster connection configuration style 1
 	opt.flags.StringSliceVar(&opt.ClusterListenClientURLs, "cluster-listen-client-urls", []string{"http://localhost:2379"}, "List of URLs to listen on for cluster client traffic.")
 	opt.flags.StringSliceVar(&opt.ClusterListenPeerURLs, "cluster-listen-peer-urls", []string{"http://localhost:2380"}, "List of URLs to listen on for cluster peer traffic.")
-	opt.flags.StringSliceVar(&opt.ClusterAdvertiseClientURLs, "cluster-advertise-client-urls", []string{"http://localhost:2379"}, "List of this member’s client URLs to advertise to the rest of the cluster.")
-	opt.flags.StringSliceVar(&opt.ClusterInitialAdvertisePeerURLs, "cluster-initial-advertise-peer-urls", []string{"http://localhost:2380"}, "List of this member’s peer URLs to advertise to the rest of the cluster.")
+	opt.flags.StringSliceVar(&opt.ClusterAdvertiseClientURLs, "cluster-advertise-client-urls", []string{"http://localhost:2379"}, "List of this member's client URLs to advertise to the rest of the cluster.")
+	opt.flags.StringSliceVar(&opt.ClusterInitialAdvertisePeerURLs, "cluster-initial-advertise-peer-urls", []string{"http://localhost:2380"}, "List of this member's peer URLs to advertise to the rest of the cluster.")
 	opt.flags.StringSliceVar(&opt.ClusterJoinURLs, "cluster-join-urls", nil,
 		"List of URLs to join, when the first url is the same with any one of cluster-initial-advertise-peer-urls, it means to join itself, and this config will be treated empty. When used, leave initial-cluster empty.")
 
 	// Cluster connection configuration style 2
 	opt.flags.StringSliceVar(&opt.Cluster.ListenClientURLs, "listen-client-urls", []string{"http://localhost:2379"}, "List of URLs to listen on for cluster client traffic.")
 	opt.flags.StringSliceVar(&opt.Cluster.ListenPeerURLs, "listen-peer-urls", []string{"http://localhost:2380"}, "List of URLs to listen on for cluster peer traffic.")
-	opt.flags.StringSliceVar(&opt.Cluster.AdvertiseClientURLs, "advertise-client-urls", []string{"http://localhost:2379"}, "List of this member’s client URLs to advertise to the rest of the cluster.")
-	opt.flags.StringSliceVar(&opt.Cluster.InitialAdvertisePeerURLs, "initial-advertise-peer-urls", []string{"http://localhost:2380"}, "List of this member’s peer URLs to advertise to the rest of the cluster.")
+	opt.flags.StringSliceVar(&opt.Cluster.AdvertiseClientURLs, "advertise-client-urls", []string{"http://localhost:2379"}, "List of this member's client URLs to advertise to the rest of the cluster.")
+	opt.flags.StringSliceVar(&opt.Cluster.InitialAdvertisePeerURLs, "initial-advertise-peer-urls", []string{"http://localhost:2380"}, "List of this member's peer URLs to advertise to the rest of the cluster.")
 	opt.flags.StringToStringVarP(&opt.Cluster.InitialCluster, "initial-cluster", "", nil,
 		"List of (member name, URL) pairs that will form the cluster. E.g. primary-1=http://localhost:2380. When used, leave cluster-join-urls empty.")
 	opt.flags.StringVar(&opt.Cluster.StateFlag, "state-flag", "new", "Cluster state (new, existing)")
@@ -232,9 +232,12 @@ func (opt *Options) Parse() (string, error) {
 		opt.viper.Set(key, val)
 	}
 
-	opt.viper.Unmarshal(opt, func(c *mapstructure.DecoderConfig) {
+	err = opt.viper.Unmarshal(opt, func(c *mapstructure.DecoderConfig) {
 		c.TagName = "yaml"
 	})
+	if err != nil {
+		return "", fmt.Errorf("yaml file unmarshal failed, please make sure you provide valid yaml file, %v", err)
+	}
 
 	opt.renameLegacyClusterRoles()
 	err = opt.validate()
diff --git a/pkg/tracing/span.go b/pkg/tracing/span.go
index 78d4f164..39aab78b 100644
--- a/pkg/tracing/span.go
+++ b/pkg/tracing/span.go
@@ -65,6 +65,8 @@ type (
 
 		// SetTag sets tag key and value.
 		SetTag(key string, value string)
+		// IsNoopSpan returns true if span is NoopSpan.
+		IsNoopSpan() bool
 	}
 
 	span struct {
@@ -75,13 +77,25 @@ type (
 	}
 )
 
+// NoopSpan does nothing.
+var NoopSpan = &span{
+	tracer: NoopTracing,
+	span:   NoopTracing.StartSpan(""), // will return opentracing.defaultNoopSpan
+}
+
 // NewSpan creates a span.
 func NewSpan(tracer *Tracing, name string) Span {
+	if tracer.IsNoopTracer() {
+		return NoopSpan
+	}
 	return newSpanWithStart(tracer, name, fasttime.Now())
 }
 
 // NewSpanWithStart creates a span with specify start time.
 func NewSpanWithStart(tracer *Tracing, name string, startAt time.Time) Span {
+	if tracer.IsNoopTracer() {
+		return NoopSpan
+	}
 	return newSpanWithStart(tracer, name, startAt)
 }
 
@@ -96,6 +110,10 @@ func newSpanWithStart(tracer *Tracing, name string, startAt time.Time) Span {
 	}
 }
 
+func (s *span) IsNoopSpan() bool {
+	return s == NoopSpan
+}
+
 func (s *span) Tracer() opentracing.Tracer {
 	return s.tracer
 }
@@ -116,10 +134,16 @@ func (s *span) Cancel() {
 }
 
 func (s *span) NewChild(name string) Span {
+	if s.IsNoopSpan() {
+		return s
+	}
 	return s.newChildWithStart(name, fasttime.Now())
 }
 
 func (s *span) NewChildWithStart(name string, startAt time.Time) Span {
+	if s.IsNoopSpan() {
+		return s
+	}
 	return s.newChildWithStart(name, startAt)
 }
 
diff --git a/pkg/tracing/tracing.go b/pkg/tracing/tracing.go
index a32234ea..6a3a64ba 100644
--- a/pkg/tracing/tracing.go
+++ b/pkg/tracing/tracing.go
@@ -67,6 +67,11 @@ func New(spec *Spec) (*Tracing, error) {
 	}, nil
 }
 
+// IsNoopTracer checks whether tracer is noop tracer.
+func (t *Tracing) IsNoopTracer() bool {
+	return t == NoopTracing
+}
+
 // Close closes Tracing.
 func (t *Tracing) Close() error {
 	if t.closer != nil {
diff --git a/pkg/util/callbackreader/callbackreader.go b/pkg/util/callbackreader/callbackreader.go
index 0b3a1591..9e853420 100644
--- a/pkg/util/callbackreader/callbackreader.go
+++ b/pkg/util/callbackreader/callbackreader.go
@@ -43,7 +43,10 @@ type (
 // New creates CallbackReader.
 func New(r io.Reader) *CallbackReader {
 	return &CallbackReader{
-		reader: r,
+		// pre-allocate slices for performance
+		beforeFuncs: make([]BeforeFunc, 0, 1),
+		afterFuncs:  make([]AfterFunc, 0, 1),
+		reader:      r,
 	}
 }
 
diff --git a/pkg/util/httpheader/httpheader.go b/pkg/util/httpheader/httpheader.go
index 435cb432..0efbe314 100644
--- a/pkg/util/httpheader/httpheader.go
+++ b/pkg/util/httpheader/httpheader.go
@@ -74,6 +74,11 @@ func (h *HTTPHeader) Std() http.Header {
 	return h.h
 }
 
+// SetRaw set the http.Header object to header.
+func (h *HTTPHeader) SetRaw(header http.Header) {
+	h.h = header
+}
+
 // Add adds the key value pair.
 func (h *HTTPHeader) Add(key, value string) {
 	h.h.Add(key, value)
diff --git a/pkg/util/jmxtool/agent_controller.go b/pkg/util/jmxtool/agent_controller.go
index 553d458f..8f77fd7f 100644
--- a/pkg/util/jmxtool/agent_controller.go
+++ b/pkg/util/jmxtool/agent_controller.go
@@ -31,14 +31,14 @@ import (
 )
 
 const (
-	canaryConfigURL  = "/config-canary"
-	serviceConfigURL = "/config-service"
+	globalTransmissionConfigURL = "/config-global-transmission"
+	serviceConfigURL            = "/config-service"
 )
 
 // AgentInterface is the interface operate the agent client
 type AgentInterface interface {
 	UpdateService(newService *spec.Service, version int64) error
-	UpdateCanary(globalHeaders *spec.GlobalCanaryHeaders, version int64) error
+	UpdateGlobalTransmission(transmission *spec.GlobalTransmission) error
 }
 
 // AgentClient stores the information of agent client
@@ -82,29 +82,25 @@ func (agent *AgentClient) UpdateService(newService *spec.Service, version int64)
 	return err
 }
 
-// UpdateCanary updates canary.
-func (agent *AgentClient) UpdateCanary(globalHeaders *spec.GlobalCanaryHeaders, version int64) error {
-	buff, err := yaml.Marshal(globalHeaders)
+// UpdateGlobalTransmission updates GlobalTransmission.
+func (agent *AgentClient) UpdateGlobalTransmission(transmission *spec.GlobalTransmission) error {
+	buff, err := yaml.Marshal(transmission)
 	if err != nil {
-		return fmt.Errorf("marshal %#v to yaml failed: %v", globalHeaders, err)
+		return fmt.Errorf("marshal %#v to yaml failed: %v", transmission, err)
 	}
+
 	jsonBytes, err := yamljsontool.YAMLToJSON(buff)
 	if err != nil {
 		return fmt.Errorf("convert yaml %s to json failed: %v", buff, err)
 	}
-	kvMap, err := JSONToKVMap(string(jsonBytes))
-	kvMap["version"] = strconv.FormatInt(version, 10)
 
-	bytes, err := json.Marshal(kvMap)
-	if err != nil {
-		return fmt.Errorf("marshal %s to json failed: %v", kvMap, err)
-	}
-
-	url := agent.URL + canaryConfigURL
-	bodyString, err := handleRequest(http.MethodPut, url, bytes)
+	url := agent.URL + globalTransmissionConfigURL
+	bodyString, err := handleRequest(http.MethodPut, url, jsonBytes)
 	if err != nil {
 		return fmt.Errorf("handleRequest error: %v", err)
 	}
-	logger.Infof("Update Canary, URL: %s,request: %s, result: %v", url, string(bytes), string(bodyString))
+
+	logger.Infof("update global transmission %s req: %s resp: %s", url, jsonBytes, bodyString)
+
 	return err
 }
diff --git a/pkg/util/urlclusteranalyzer/urlclusteranalyzer.go b/pkg/util/urlclusteranalyzer/urlclusteranalyzer.go
index 375a1967..cc670e44 100644
--- a/pkg/util/urlclusteranalyzer/urlclusteranalyzer.go
+++ b/pkg/util/urlclusteranalyzer/urlclusteranalyzer.go
@@ -20,6 +20,8 @@ package urlclusteranalyzer
 import (
 	"strings"
 	"sync"
+
+	lru "github.com/hashicorp/golang-lru"
 )
 
 const (
@@ -31,6 +33,7 @@ const (
 type URLClusterAnalyzer struct {
 	slots []*field `yaml:"slots"`
 	mutex *sync.Mutex
+	cache *lru.Cache
 }
 
 type field struct {
@@ -55,9 +58,11 @@ func newField(name string) *field {
 
 // New creates a URLClusterAnalyzer.
 func New() *URLClusterAnalyzer {
+	cache, _ := lru.New(4098)
 	u := &URLClusterAnalyzer{
 		mutex: &sync.Mutex{},
 		slots: make([]*field, maxLayers),
+		cache: cache,
 	}
 
 	for i := 0; i < maxLayers; i++ {
@@ -74,6 +79,9 @@ func (u *URLClusterAnalyzer) GetPattern(urlPath string) string {
 	if urlPath == "" {
 		return ""
 	}
+	if val, ok := u.cache.Get(urlPath); ok {
+		return val.(string)
+	}
 
 	var values []string
 	if urlPath[0] == '/' {
@@ -132,5 +140,7 @@ LOOP:
 		currField = newF
 	}
 
+	u.cache.Add(urlPath, currPattern)
+
 	return currPattern
 }


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 38ab30e8..5eb4d93b 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -6,6 +6,8 @@ on:
       - main
     paths:
       - "**/*.go"
+      - "**/*.yaml"
+      - "**/*.sh"
       - "go.mod"
       - "go.sum"
       - ".github/workflows/test.yml"
diff --git a/build/test/httpserver_test.sh b/build/test/httpserver_test.sh
index 1fecf110..85d0a02b 100755
--- a/build/test/httpserver_test.sh
+++ b/build/test/httpserver_test.sh
@@ -26,10 +26,10 @@ set -e
 SCRIPTPATH="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
 pushd $SCRIPTPATH"/../../example" > /dev/null
 EXAMPLEDIR="$SCRIPTPATH"/../../example
-PRIMARY01DIR=$EXAMPLEDIR"/primary-001"
+PRIMARYDIR=$EXAMPLEDIR"/primary-single"
 
 # target file related define.
-server="primary-001/bin/easegress-server"
+server="primary-single/bin/easegress-server"
 backend="$EXAMPLEDIR/backend-service/echo/echo.go"
 httpsvr_port=10080
 echo_port=9095
@@ -40,7 +40,7 @@ COLOR_NONE='\033[0m'
 COLOR_INFO='\033[0;36m'
 COLOR_ERROR='\033[1;31m'
 
-# clean cleans primary-001's cluster data and the `go run` process.
+# clean cleans primary-single's data and cluster data and the `go run` process.
 function clean()
 {
     # basic cleaning routine
@@ -65,8 +65,8 @@ function clean()
 # clean the cluster resource first.
 clean
 
-# start primary01 for testing.
-bash $PRIMARY01DIR/start.sh
+# start primary-single for testing.
+bash $PRIMARYDIR/start.sh
 try_time=0
 # wait Easegress to be ready, it will retry three times.
 # reference: https://unix.stackexchange.com/questions/5277/how-do-i-tell-a-script-to-wait-for-a-process-to-start-accepting-requests-on-a-po
@@ -80,7 +80,7 @@ do
     fi
 done
 
-# check the primary01 running status
+# check the primary-single running status
 pid=`ps -eo pid,args | grep "$server" | grep -v grep | awk '{print $1}'`
 if [ "$pid" = "" ]; then
     echo -e "\n${COLOR_ERROR}start test server $server failed${COLOR_NONE}"
@@ -101,7 +101,7 @@ https: false
 rules:
   - paths:
     - pathPrefix: /pipeline
-      backend: pipeline-demo' | $PRIMARY01DIR/egctl.sh object create
+      backend: pipeline-demo' | $PRIMARYDIR/egctl.sh object create
 
 #  create Pipeline
 echo '
@@ -116,7 +116,7 @@ filters:
       servers:
       - url: http://127.0.0.1:9095
       loadBalance:
-        policy: roundRobin' | $PRIMARY01DIR/egctl.sh object create
+        policy: roundRobin' | $PRIMARYDIR/egctl.sh object create
 
 while ! nc -z localhost $httpsvr_port </dev/null
 do
diff --git a/pkg/context/contexttest/context.go b/pkg/context/contexttest/context.go
index c00f1405..0b31ed36 100644
--- a/pkg/context/contexttest/context.go
+++ b/pkg/context/contexttest/context.go
@@ -45,6 +45,7 @@ type MockedHTTPContext struct {
 	MockedClientDisconnected func() bool
 	MockedOnFinish           func(func())
 	MockedAddTag             func(tag string)
+	MockedAddLazyTag         func(func() string)
 	MockedStatMetric         func() *httpstat.Metric
 	MockedFinish             func()
 	MockedTemplate           func() texttemplate.TemplateEngine
@@ -164,6 +165,13 @@ func (c *MockedHTTPContext) AddTag(tag string) {
 	}
 }
 
+// AddLazyTag mocks the AddLazyTag function of HTTPContext
+func (c *MockedHTTPContext) AddLazyTag(lazyTag func() string) {
+	if c.MockedAddLazyTag != nil {
+		c.MockedAddLazyTag(lazyTag)
+	}
+}
+
 // StatMetric mocks the StatMetric function of HTTPContext
 func (c *MockedHTTPContext) StatMetric() *httpstat.Metric {
 	if c.MockedStatMetric != nil {
diff --git a/pkg/filter/mock/mock_test.go b/pkg/filter/mock/mock_test.go
index afac3f2a..bc7110d9 100644
--- a/pkg/filter/mock/mock_test.go
+++ b/pkg/filter/mock/mock_test.go
@@ -42,17 +42,47 @@ func TestMock(t *testing.T) {
 kind: Mock
 name: mock
 rules:
-- pathPrefix: /login/
+- match:
+    pathPrefix: /login/
   code: 202
   body: 'mocked body'
   headers:
     X-Test: test1
-- path: /sales
+- match:
+    path: /sales
   code: 203
   body: 'mocked body'
   headers:
     X-Test: test2
   delay: 1ms
+- match:
+    path: /pets
+    headers:
+      X-Mock:
+        exact: mock
+  code: 205
+  body: 'mocked body'
+  headers:
+    X-Test: test2
+- match:
+    path: /customers
+    headers:
+      X-Mock:
+        empty: true
+  code: 206
+  body: 'mocked body'
+  headers:
+    X-Test: test2
+- match:
+    path: /vets
+    matchAllHeader: true
+    headers:
+      X-Mock:
+        exact: mock
+  code: 207
+  body: 'mocked body'
+  headers:
+    X-Test: test2
 - code: 204
   body: 'mocked body 2'
   headers:
@@ -114,6 +144,79 @@ rules:
 		t.Error("header 'X-Test' should be 'test2'")
 	}
 
+	resp = httptest.NewRecorder()
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/pets"
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		h := http.Header{}
+		h.Add("X-Mock", "mock")
+		return httpheader.New(h)
+	}
+	m.Handle(ctx)
+	if resp.Code != 205 {
+		t.Error("status code is not 205")
+	}
+	if resp.Body.String() != "mocked body" {
+		t.Error("body should be 'mocked body'")
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		h := http.Header{}
+		h.Add("X-Mock", "mock1")
+		return httpheader.New(h)
+	}
+	resp = httptest.NewRecorder()
+	m.Handle(ctx)
+	if resp.Code != 204 {
+		t.Errorf("status code is %d, not 204", resp.Code)
+	}
+
+	resp = httptest.NewRecorder()
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/customers"
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		h := http.Header{
+			"X-Mock": []string{},
+		}
+		return httpheader.New(h)
+	}
+	m.Handle(ctx)
+	if resp.Code != 206 {
+		t.Errorf("status code is %d, not 206", resp.Code)
+	}
+	if body := resp.Body.String(); body != "mocked body" {
+		t.Errorf("body is %q should be 'mocked body'", body)
+	}
+
+	resp = httptest.NewRecorder()
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/vets"
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		h := http.Header{}
+		h.Add("X-Mock", "mock")
+		return httpheader.New(h)
+	}
+	m.Handle(ctx)
+	if resp.Code != 207 {
+		t.Error("status code is not 207")
+	}
+	if resp.Body.String() != "mocked body" {
+		t.Error("body should be 'mocked body'")
+	}
+
+	resp = httptest.NewRecorder()
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		h := http.Header{}
+		h.Add("X-Mock", "mock1")
+		return httpheader.New(h)
+	}
+	m.Handle(ctx)
+	if resp.Code != 204 {
+		t.Errorf("status code is %d, not 204", resp.Code)
+	}
+
 	if m.Status() != nil {
 		t.Error("behavior changed, please update this case")
 	}
diff --git a/pkg/filter/proxy/request_test.go b/pkg/filter/proxy/request_test.go
index 69f4e2a4..17fb1c20 100644
--- a/pkg/filter/proxy/request_test.go
+++ b/pkg/filter/proxy/request_test.go
@@ -24,7 +24,10 @@ import (
 	"testing"
 	"time"
 
+	httpstat "github.com/tcnksm/go-httpstat"
+
 	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/util/fasttime"
 	"github.com/megaease/easegress/pkg/util/httpheader"
 )
 
@@ -52,7 +55,8 @@ func TestRequest(t *testing.T) {
 
 	p := pool{}
 	sr := strings.NewReader("this is the raw body")
-	req, _ := p.newRequest(ctx, &server, sr)
+	req, _ := p.newRequest(ctx, &server, sr, requestPool, httpstatResultPool)
+	defer requestPool.Put(req) // recycle request
 
 	req.start()
 	tm := req.startTime()
@@ -93,3 +97,35 @@ func TestResultState(t *testing.T) {
 		t.Error("implementation changed, this case should be updated")
 	}
 }
+
+func TestRequestStatus(t *testing.T) {
+	statResult := httpstatResultPool.Get().(*httpstat.Result)
+	req := requestPool.Get().(*request)
+	req.createTime = fasttime.Now()
+	req._startTime = time.Time{}
+	req._endTime = time.Time{}
+	req.statResult = statResult
+
+	if !req.createTime.Equal(req.startTime()) {
+		t.Error("starttime should be createtime before start()")
+	}
+	time.Sleep(time.Millisecond)
+	req.start()
+
+	if req.createTime.Equal(req.startTime()) {
+		t.Error("starttime should not be createtime after start()")
+	}
+
+	if req.endTime().Equal(req._endTime) {
+		t.Error("endtime should be now before finish()")
+	}
+	time.Sleep(time.Millisecond)
+	req.finish()
+
+	if !req.endTime().Equal(req._endTime) {
+		t.Error("endtime should be _endtime after finish()")
+	}
+
+	httpstatResultPool.Put(req.statResult)
+	requestPool.Put(req)
+}
diff --git a/pkg/object/meshcontroller/spec/spec_test.go b/pkg/object/meshcontroller/spec/spec_test.go
index 7bec7fba..7430a779 100644
--- a/pkg/object/meshcontroller/spec/spec_test.go
+++ b/pkg/object/meshcontroller/spec/spec_test.go
@@ -349,9 +349,11 @@ func TestSidecarEgressPipelineSpecWithMock(t *testing.T) {
 			Enabled: true,
 			Rules: []*mock.Rule{
 				{
-					Path:       "/abc",
-					PathPrefix: "/",
-					Code:       200,
+					Match: mock.MatchRule{
+						Path:       "/abc",
+						PathPrefix: "/",
+					},
+					Code: 200,
 					Headers: map[string]string{
 						"mock-by-eg": "yes",
 					},
@@ -378,9 +380,11 @@ func TestMockPBConvert(t *testing.T) {
 		Enabled: true,
 		Rules: []*v1alpha1.MockRule{
 			{
-				Path:       "/",
-				PathPrefix: "/abc",
-				Code:       200,
+				Match: &v1alpha1.MockMatchRule{
+					Path:       "/",
+					PathPrefix: "/abc",
+				},
+				Code: 200,
 				Headers: map[string]string{
 					"mock-by-eg": "yes",
 				},
diff --git a/pkg/object/mqttproxy/mqtt_test.go b/pkg/object/mqttproxy/mqtt_test.go
index ca7ef049..4bd0f6d5 100644
--- a/pkg/object/mqttproxy/mqtt_test.go
+++ b/pkg/object/mqttproxy/mqtt_test.go
@@ -50,7 +50,6 @@ func (t *testMQ) get() *packets.PublishPacket {
 
 func init() {
 	logger.InitNop()
-	// logger.InitMock()
 	pipeline.Register(&pipeline.MockMQTTFilter{})
 }
 
@@ -178,6 +177,37 @@ func TestSubUnsub(t *testing.T) {
 	broker.close()
 }
 
+func checkSessionStore(broker *Broker, cid, topic string) error {
+	checkFn := func() error {
+		sessStr, err := broker.sessMgr.store.get(sessionStoreKey(cid))
+		if err != nil {
+			return err
+		}
+		if topic == "" {
+			return nil
+		}
+		sess := Session{
+			info: &SessionInfo{},
+		}
+		sess.decode(*sessStr)
+
+		for t := range sess.info.Topics {
+			if topic == t {
+				return nil
+			}
+		}
+		return fmt.Errorf("topic %v not in session %v", topic, cid)
+	}
+
+	for i := 0; i < 20; i++ {
+		if checkFn() == nil {
+			return nil
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	return fmt.Errorf("session %v with topic %v not been stored", cid, topic)
+}
+
 func TestCleanSession(t *testing.T) {
 	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
 	broker := getBroker("test", "test", b64passwd, 1883)
@@ -185,9 +215,16 @@ func TestCleanSession(t *testing.T) {
 	// client that set cleanSession
 	cid := "cleanSessionClient"
 	client := getMQTTClient(t, cid, "test", "test", true)
+	if err := checkSessionStore(broker, cid, ""); err != nil {
+		t.Fatal(err)
+	}
 	if token := client.Subscribe("test/cleanSession/0", 0, nil); token.Wait() && token.Error() != nil {
 		t.Errorf("subscribe qos0 error %s", token.Error())
 	}
+	if err := checkSessionStore(broker, cid, "test/cleanSession/0"); err != nil {
+		t.Fatal(err)
+	}
+
 	client.Disconnect(200)
 
 	c := broker.getClient(cid)
@@ -1846,13 +1883,32 @@ func TestHTTPGetAllSession(t *testing.T) {
 	clients := []paho.Client{}
 	clientNum := 10
 	for i := 0; i < clientNum; i++ {
-		client := getMQTTClient(t, strconv.Itoa(i), "test", "test", true)
+		cid := strconv.Itoa(i)
+		client := getMQTTClient(t, cid, "test", "test", true)
+		if err := checkSessionStore(broker, cid, ""); err != nil {
+			t.Fatal(err)
+		}
 		if token := client.Subscribe("topic", 1, nil); token.Wait() && token.Error() != nil {
 			t.Errorf("subscribe qos0 error %s", token.Error())
 		}
+		if err := checkSessionStore(broker, cid, "topic"); err != nil {
+			t.Fatal(err)
+		}
 		clients = append(clients, client)
 	}
 
+	// we use goroutine to store session, make sure all sessions have been stored before we go forward.
+	for i := 0; i < 20; i++ {
+		sessions, _ := broker.sessMgr.store.getPrefix(sessionStoreKey(""), true)
+		if len(sessions) == clientNum {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+		if i == 19 && len(sessions) != clientNum {
+			t.Fatalf("not all sessions have been stored %v", sessions)
+		}
+	}
+
 	// start server
 	srv := newServer(":8888")
 	srv.addHandlerFunc("/session/query", broker.httpGetAllSessionHandler)
@@ -1895,6 +1951,10 @@ func TestHTTPGetAllSession(t *testing.T) {
 			json.NewDecoder(resp.Body).Decode(sessions)
 			if len(sessions.Sessions) != test.ansLen {
 				t.Errorf("get wrong session number wanted %v, got %v", test.ansLen, len(sessions.Sessions))
+				sessions, _ := broker.sessMgr.store.getPrefix(sessionStoreKey(""), true)
+				broker.Lock()
+				t.Errorf("broker clients %v, sessions %v", broker.clients, sessions)
+				broker.Unlock()
 			}
 		}
 		resp.Body.Close()
diff --git a/pkg/util/jmxtool/agent_controller_test.go b/pkg/util/jmxtool/agent_controller_test.go
index 813ea9de..12761b99 100644
--- a/pkg/util/jmxtool/agent_controller_test.go
+++ b/pkg/util/jmxtool/agent_controller_test.go
@@ -45,7 +45,7 @@ func createHTTPServer(finished chan bool, notFoundFlag bool) error {
 			m.HandleFunc(serviceConfigURL, func(w http.ResponseWriter, r *http.Request) {
 				fmt.Fprintf(w, "Hello, %q", html.EscapeString(r.URL.Path))
 			})
-			m.HandleFunc(canaryConfigURL, func(w http.ResponseWriter, r *http.Request) {
+			m.HandleFunc(globalTransmissionConfigURL, func(w http.ResponseWriter, r *http.Request) {
 				fmt.Fprintf(w, "Hello, %q", html.EscapeString(r.URL.Path))
 			})
 		}
@@ -109,11 +109,11 @@ func TestAgentClientSuccess(t *testing.T) {
 		t.Errorf("agent update service failed: %v\n", err)
 	}
 
-	// UpdateCanary
-	header := &spec.GlobalCanaryHeaders{
-		ServiceHeaders: map[string][]string{},
+	// UpdateGlobalTransmission
+	transmission := &spec.GlobalTransmission{
+		Headers: []string{},
 	}
-	err = agent.UpdateCanary(header, 1)
+	err = agent.UpdateGlobalTransmission(transmission)
 	if err != nil {
 		t.Errorf("agent update canary failed: %v\n", err)
 	}
@@ -134,10 +134,10 @@ func TestAgentClientFail(t *testing.T) {
 	if err == nil {
 		t.Errorf("agent should fail\n")
 	}
-	header := &spec.GlobalCanaryHeaders{
-		ServiceHeaders: map[string][]string{},
+	transmission := &spec.GlobalTransmission{
+		Headers: []string{},
 	}
-	err = agent.UpdateCanary(header, 1)
+	err = agent.UpdateGlobalTransmission(transmission)
 	if err == nil {
 		t.Errorf("agent should fail\n")
 	}
@@ -154,7 +154,7 @@ func TestAgentClientFail(t *testing.T) {
 	if err == nil {
 		t.Errorf("agent should fail\n")
 	}
-	err = agent.UpdateCanary(header, 1)
+	err = agent.UpdateGlobalTransmission(transmission)
 	if err == nil {
 		t.Errorf("agent should fail\n")
 	}