Repository: megaease/easegress
Pull request #439 (2021-12-24 00:07:04)

Reduce etcd server memory usage
- set snapshot count to lower value to reduce etcd ram usage https://etcd.io/docs/v3.5/tuning/#snapshot-tuning
- fix https://github.com/megaease/easegress/issues/420 ; do not print debug client logs if `--debug` flag is not provided for server

codecov-commenter: # [Codecov](https://codecov.io/gh/megaease/easegress/pull/439?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) Report
> Merging [#439](https://codecov.io/gh/megaease/easegress/pull/439?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (cbf203f) into [main](https://codecov.io/gh/megaease/easegress/commit/fb7d28c6d8b3748b67667976749d57467be29fc1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (fb7d28c) will **decrease** coverage by `0.21%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/megaease/easegress/pull/439/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/439?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)

```diff
@@            Coverage Diff             @@
##             main     #439      +/-   ##
==========================================
- Coverage   80.44%   80.22%   -0.22%     
==========================================
  Files          70       70              
  Lines        8094     8096       +2     
==========================================
- Hits         6511     6495      -16     
- Misses       1225     1243      +18     
  Partials      358      358              
```


| [Impacted Files](https://codecov.io/gh/megaease/easegress/pull/439?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | Coverage Δ | |
|---|---|---|
| [pkg/cluster/config.go](https://codecov.io/gh/megaease/easegress/pull/439/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY29uZmlnLmdv) | `89.07% <100.00%> (+0.18%)` | :arrow_up: |
| [pkg/object/mqttproxy/session.go](https://codecov.io/gh/megaease/easegress/pull/439/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvc2Vzc2lvbi5nbw==) | `74.23% <0.00%> (-7.98%)` | :arrow_down: |
| [pkg/object/mqttproxy/client.go](https://codecov.io/gh/megaease/easegress/pull/439/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvY2xpZW50Lmdv) | `77.67% <0.00%> (-2.33%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/megaease/easegress/pull/439?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/439?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Last update [94c87ad...cbf203f](https://codecov.io/gh/megaease/easegress/pull/439?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).

suchen-sci: @samutamm please update `httpserver_test.sh` to solve conflicts...
samutamm: > @samutamm please update `httpserver_test.sh` to solve conflicts...

@suchen-sci conflicts resolved

Patch:
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/doc/reference/ingresscontroller.md b/doc/reference/ingresscontroller.md
index 9cbe5a5d..28af7855 100644
--- a/doc/reference/ingresscontroller.md
+++ b/doc/reference/ingresscontroller.md
@@ -6,13 +6,10 @@
     - [Controller spec](#controller-spec)
   - [Getting Started](#getting-started)
     - [Role Based Access Control configuration](#role-based-access-control-configuration)
-    - [Access to disk](#access-to-disk)
-    - [Create Headless service and set ports](#create-headless-service-and-set-ports)
     - [Configurations to ConfigMap](#configurations-to-configmap)
     - [Deploy Easegress IngressController](#deploy-easegress-ingresscontroller)
     - [Create backend service & Kubernetes ingress](#create-backend-service--kubernetes-ingress)
-  - [Multi-node IngressController](#multi-node-ingresscontroller)
-  - [Add secondary Easegress instances](#add-secondary-easegress-instances)
+  - [Multi-instance IngressController](#multi-instance-ingresscontroller)
 
 The IngressController is an implementation of [Kubernetes ingress controller](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/), it watches Kubernetes Ingress, Service, Endpoints, and Secrets then translates them to Easegress HTTP server and pipelines.
 
@@ -89,123 +86,23 @@ roleRef:
 
 Note the name of the ServiceAccount we just created is `easegress-ingress-controller`, it will be used later.
 
-### Access to disk
-
-Easegress instances needs to access disk, to support failure of the pods. In this example we use local volume, but you could also mount any cloud providers volumes ([here](https://kubernetes.io/docs/concepts/storage/volumes/) are some of the options).
-
-To mount local directory at host machine, we create PersistentVolume and StorageClass. Replace `<YOUR-HOSTNAME-HERE>` by hostname of the machine, where Easegress can use the disk.
-
-```yaml
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: easegress-pv-1
-spec:
-  capacity:
-    storage: 4Gi
-  volumeMode: Filesystem
-  accessModes:
-  - ReadWriteOnce
-  persistentVolumeReclaimPolicy: Delete
-  storageClassName: easegress-storage
-  hostPath:
-    path: /opt/easegress
-  nodeAffinity:
-    required:
-      nodeSelectorTerms:
-      - matchExpressions:
-        - key: kubernetes.io/hostname
-          operator: In
-          values:
-          - <YOUR-HOSTNAME-HERE>
----
-
-apiVersion: storage.k8s.io/v1
-kind: StorageClass
-metadata:
-  name: easegress-storage
-provisioner: kubernetes.io/no-provisioner
-reclaimPolicy: Delete
-volumeBindingMode: WaitForFirstConsumer
-```
-
-### Create Headless service and set ports
-
-Create ClusterIP [Headless](https://kubernetes.io/docs/concepts/services-networking/service/#headless-services) service to facilitate Easegress server(s) communication.
-
-```yaml
----
-
-apiVersion: v1
-kind: Service
-metadata:
-  name: easegress-hs
-  namespace: default
-spec:
-  clusterIP: None
-  ports:
-  - name: admin-port
-    port: 2381
-    protocol: TCP
-    targetPort: 2381
-  - name: peer-port
-    port: 2380
-    protocol: TCP
-    targetPort: 2380
-  - name: client-port
-    port: 2379
-    protocol: TCP
-    targetPort: 2379
-  selector:
-    app: easegress-ingress
-  type: ClusterIP
-
----
-
-apiVersion: v1
-kind: Service
-metadata:
-  name: easegress-public
-  namespace: default
-spec:
-  ports:
-  - name: admin-port
-    nodePort: 31255 # random port
-    port: 2381
-    protocol: TCP
-    targetPort: 2381
-  selector:
-    app: easegress-ingress
-  type: NodePort
-```
-
 ### Configurations to ConfigMap
 
-Let's use ConfigMap to store Easegress server configuration and Easegress ingress configuration. We will use this ConfigMap later in StatefulSet.
+Let's use ConfigMap to store Easegress server configuration and Easegress ingress configuration. This ConfigMap is used later in Deployment.
 
 ```yaml
 apiVersion: v1
 kind: ConfigMap
 metadata:
-  name: easegress-cluster-cm
+  name: easegress-cm
   namespace: default
 data:
-  eg-primary.yaml: |
+  easegress-server.yaml: |
+    name: ingress-easegress
     cluster-name: easegress-ingress-controller
     cluster-role: primary
-    cluster:
-      listen-client-urls:
-      - http://0.0.0.0:2379
-      listen-peer-urls:
-      - http://0.0.0.0:2380
-      initial-cluster:
-      - easegress-0: http://easegress-0.easegress-hs.default:2380
     api-addr: 0.0.0.0:2381
     data-dir: /opt/easegress/data
-    wal-dir: ""
-    cpu-profile-file: ""
-    memory-profile-file: ""
     log-dir: /opt/easegress/log
     debug: false
   controller.yaml: |
@@ -223,15 +120,15 @@ data:
       maxConnections: 10240
 ```
 
-The `eg-primary.yaml` creates Easegress cluster with one member: `http://easegress-0.easegress-hs.default:2380`. Once we create a StatefulSet called `easegress`, using serviceName `easegress-hs` and inside namespace `default`, this URL will match Kubernetes DNS record. You can read more about stable network ids [here](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#stable-network-id).
+The `easegress-serverl` creates Easegress instance named *easegress-ingress-controller* and `controller.yaml` defines IngressController object for Easegress.
 
 ### Deploy Easegress IngressController
 
-To deploy the IngressController, we will create a StatefulSet and a Service as below:
+To deploy the IngressController, we will create a Deployment and a Service as below:
 
 ```yaml
 apiVersion: apps/v1
-kind: StatefulSet
+kind: Deployment
 metadata:
   labels:
     app: easegress-ingress
@@ -242,7 +139,6 @@ spec:
   selector:
     matchLabels:
       app: easegress-ingress
-  serviceName: easegress-hs
   template:
     metadata:
       labels:
@@ -253,13 +149,10 @@ spec:
       - args:
         - -c
         - |-
-          echo name: $EG_NAME > /opt/eg-config/config.yaml &&
-          cat /opt/eg-config/eg-primary.yaml >> /opt/eg-config/config.yaml &&
           /opt/easegress/bin/easegress-server \
-            -f /opt/eg-config/config.yaml \
-            --advertise-client-urls http://$(EG_NAME).easegress-hs.default:2379 \
-            --initial-advertise-peer-urls http://$(EG_NAME).easegress-hs.default:2380 \
-            --initial-object-config-files /opt/eg-config/controller.yaml
+            -f /opt/eg-config/easegress-server.yaml \
+            --initial-object-config-files /opt/eg-config/controller.yaml \
+            --initial-cluster $(EG_NAME)=http://localhost:2380
         command:
         - /bin/sh
         env:
@@ -270,17 +163,7 @@ spec:
               fieldPath: metadata.name
         image: megaease/easegress:latest
         imagePullPolicy: IfNotPresent
-        name: easegress-ingress
-        ports:
-        - containerPort: 2381
-          name: admin-port
-          protocol: TCP
-        - containerPort: 2380
-          name: peer-port
-          protocol: TCP
-        - containerPort: 2379
-          name: client-port
-          protocol: TCP
+        name: easegress-primary
         resources:
           limits:
             cpu: 1200m
@@ -289,61 +172,53 @@ spec:
             cpu: 100m
             memory: 256Mi
         volumeMounts:
-        - mountPath: /opt/eg-config/eg-primary.yaml
-          name: easegress-cluster-cm
-          subPath: eg-primary.yaml
+        - mountPath: /opt/eg-config/easegress-server.yaml
+          name: easegress-cm
+          subPath: easegress-server.yaml
         - mountPath: /opt/eg-config/controller.yaml
-          name: easegress-cluster-cm
+          name: easegress-cm
           subPath: controller.yaml
-        - mountPath: /opt/eg-data/
-          name: easegress-pv
+        - mountPath: /opt/easegress/data
+          name: ingress-data-volume
+        - mountPath: /opt/easegress/log
+          name: ingress-data-volume
       restartPolicy: Always
       volumes:
+      - emptyDir: {}
+        name: ingress-data-volume
       - configMap:
           defaultMode: 420
           items:
-          - key: eg-primary.yaml
-            path: eg-primary.yaml
+          - key: easegress-server.yaml
+            path: easegress-server.yaml
           - key: controller.yaml
             path: controller.yaml
-          name: easegress-cluster-cm
-        name: easegress-cluster-cm
-  updateStrategy:
-    type: RollingUpdate
-  volumeClaimTemplates:
-  - apiVersion: v1
-    kind: PersistentVolumeClaim
-    metadata:
-      name: easegress-pv
-    spec:
-      accessModes:
-      - ReadWriteOnce
-      resources:
-        requests:
-          storage: 3Gi
-      storageClassName: easegress-storage
-      volumeMode: Filesystem
+          name: easegress-cm
+        name: easegress-cm
+```
 
----
-kind: Service
+The IngressController is created via the command line argument `initial-object-config-files` of `easegress-server`. Notice how Easegress logs and data are stored to *emptyDir* called `ingress-data-volume` inside the pod. IngressController is stateless so we can restart new pods without preserving previous state.
+
+Last but not least let's create service for forwarding the Ingress traffic to Easegress.
+
+```yaml
 apiVersion: v1
+kind: Service
 metadata:
-  name: easegress-ingress-service
+  name: easegress-public
   namespace: default
 spec:
+  ports:
+  - name: web
+    protocol: TCP
+    port: 8080
+    nodePort: 30080
   selector:
     app: easegress-ingress
-  ports:
-    - name: web
-      protocol: TCP
-      port: 8080
-      nodePort: 80
   type: NodePort
 ```
 
-The IngressController is created via the command line argument `initial-object-config-files` of `easegress-server`. And the exported port `web` is to receive external HTTP requests and forward them to the HTTP server in Easegress.
-
-This StatefulSet creates one replica (at runtime easegress-0), but we will see later in this tutorial, what changes are needed to be able to have 3,5 or 7 replicas.
+The port `web` is to receive external HTTP requests from port 30080 and forward them to the HTTP server in Easegress.
 
 ### Create backend service & Kubernetes ingress
 
@@ -432,12 +307,12 @@ spec:
 Once all pods are up and running, we can leverage the command below to access both versions of the `hello` application:
 
 ```bash
-$ curl http://{NODE_IP}/ -HHost:www.megaease.com
+$ curl http://{NODE_IP}:30080/ -HHost:www.megaease.com
 Hello, world!
 Version: 2.0.0
 Hostname: hello-deployment-6cbf765985-r6242
 
-$ curl http://{NODE_IP}/ -HHost:www.example.com
+$ curl http://{NODE_IP}:30080/ -HHost:www.example.com
 Hello, world!
 Version: 1.0.0
 Hostname: hello-deployment-6cbf765985-r6242
@@ -445,310 +320,19 @@ Hostname: hello-deployment-6cbf765985-r6242
 
 And we can see Easegress IngressController has forwarded requests to the correct application version according to Kubernetes ingress.
 
-## Multi-node IngressController
-
-Until now we have only created IngressController with one instance running. To support high-availability scenarios, let's go through the configurations for creating a cluster of 3 *primary* Easegress instances.
-
-Before applying the changes, let's remove the old resources first:
-
-```yaml
-kubectl delete statefulset easegress
-kubectl delete pv easegress-pv-0
-kubectl delete configmap easegress-cluster-cm
-```
-
-Also run `rm -rf /opt/easegress` on machine that `easegress-pv-0` was attached to.
-
-Here are briefly the modifications we need in order to support 3 member cluster:
-
-- in part [Access to disk](#access-to-disk): instead of creating only one *PersistentVolume* `easegress-pv-1`, create three of them `easegress-pv-1`, `easegress-pv-2`, `easegress-pv-3`
-- in part [Configurations to ConfigMap](#configurations-to-configmap) add all three member urls to cluster configuration
-- in part [Deploy Easegress IngressController](#deploy-easegress-ingresscontroller), change the *StatefulfulSet* replica count to 3
-
-So the *PersistentVolume* creation would look something like this:
-
-```yaml
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: easegress-pv-1
-spec:
-  capacity:
-    storage: 4Gi
-  volumeMode: Filesystem
-  accessModes:
-  - ReadWriteOnce
-  persistentVolumeReclaimPolicy: Delete
-  storageClassName: easegress-storage
-  hostPath:
-    path: /opt/easegress
-  nodeAffinity:
-    required:
-      nodeSelectorTerms:
-      - matchExpressions:
-        - key: kubernetes.io/hostname
-          operator: In
-          values:
-          - <YOUR-HOSTNAME-1-HERE>
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: easegress-pv-2
-spec:
-  capacity:
-    storage: 4Gi
-  volumeMode: Filesystem
-  accessModes:
-  - ReadWriteOnce
-  persistentVolumeReclaimPolicy: Delete
-  storageClassName: easegress-storage
-  hostPath:
-    path: /opt/easegress
-  nodeAffinity:
-    required:
-      nodeSelectorTerms:
-      - matchExpressions:
-        - key: kubernetes.io/hostname
-          operator: In
-          values:
-          - <YOUR-HOSTNAME-2-HERE>
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: easegress-pv-3
-spec:
-  capacity:
-    storage: 4Gi
-  volumeMode: Filesystem
-  accessModes:
-  - ReadWriteOnce
-  persistentVolumeReclaimPolicy: Delete
-  storageClassName: easegress-storage
-  hostPath:
-    path: /opt/easegress
-  nodeAffinity:
-    required:
-      nodeSelectorTerms:
-      - matchExpressions:
-        - key: kubernetes.io/hostname
-          operator: In
-          values:
-          - <YOUR-HOSTNAME-3-HERE>
-
----
-apiVersion: storage.k8s.io/v1
-kind: StorageClass
-metadata:
-  name: easegress-storage
-provisioner: kubernetes.io/no-provisioner
-reclaimPolicy: Delete
-volumeBindingMode: WaitForFirstConsumer
-```
+## Multi-instance IngressController
 
-Now here's how we add all three cluster member URLs to *ConfigMap*:
-
-```yaml
-kind: ConfigMap
-metadata:
-  name: easegress-cluster-cm
-  namespace: default
-data:
-  eg-primary.yaml: |
-    cluster-name: easegress-ingress-controller
-    cluster-role: primary
-    cluster:
-      listen-client-urls:
-      - http://0.0.0.0:2379
-      listen-peer-urls:
-      - http://0.0.0.0:2380
-      initial-cluster:
-      - easegress-0: http://easegress-0.easegress-hs.default:2380
-      - easegress-1: http://easegress-1.easegress-hs.default:2380
-      - easegress-2: http://easegress-2.easegress-hs.default:2380
-    api-addr: 0.0.0.0:2381
-    data-dir: /opt/easegress/data
-    wal-dir: ""
-    cpu-profile-file: ""
-    memory-profile-file: ""
-    log-dir: /opt/easegress/log
-    debug: false
-  controller.yaml: |
-    ...
-```
-
-And last but not least the *StatefulSet* with three replicas:
+In previous chapters we created IngressController with one instance running. To support high-availability scenarios, you can increase the number of replicas in the Deployment:
 
 ```yaml
 apiVersion: apps/v1
-kind: StatefulSet
+kind: Deployment
 metadata:
   labels:
     app: easegress-ingress
   name: easegress
   namespace: default
 spec:
-  replicas: 3 # changed from 1 to 3
-  selector:
-    matchLabels:
-      app: easegress-ingress
+  replicas: 2 # number of IngressController instances running
   ...
 ```
-
-Now you have configuration yaml files for bootstrapping three member Easegress Ingress Controller cluster. You can test ingress like before `curl http://{NODE_IP}/ -HHost:www.example.com`. To verify that the cluster is healthy, query admin port with `egctl`:
-
-```bash
-egctl --server {NODE_IP}:31255 member list |grep " name"
-```
-
-which should output:
-
-```bash
-#     name: easegress-0
-#     name: easegress-1
-#     name: easegress-2
-```
-
-You can read more multi-node Easegress this cookbook [chapter](./doc/../cookbook/multi_node_cluster.md).
-
-## Add secondary Easegress instances
-
-Previously in this tutorial we went through the necessary kubernetes files for single instance Easegress Ingress Controller and three member Easegress Ingress Controller cluster. In both cases we deployed *primary* Easegress instances. Sometimes we want to scale the number of instances up or down. This can be achieved using *secondary* Easegress instances, that unlike *primary* instances, can be added to existing cluster or removed anytime. *Secondary* instances do not require [stable network id](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#stable-network-id) so we can use Deployment instead of StatfulSet.
-
-Let's create an Kubernetes Deployment that adds two *secondary* Easegress instances to cluster.
-
-Here's Easegress configuration for *secondary* instance.
-
-```yaml
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: easegress-cluster-cm
-  namespace: default
-data:
-  eg-primary.yaml: |
-    ...
-  eg-secondary.yaml: |
-    cluster-name: easegress-ingress-controller
-    cluster-role: secondary
-    cluster:
-      primary-listen-peer-urls:
-      - http://easegress-0.easegress-hs.default:2380
-    api-addr: 0.0.0.0:2381
-    data-dir: /opt/easegress/data
-    wal-dir: ""
-    cpu-profile-file: ""
-    memory-profile-file: ""
-    log-dir: /opt/easegress/log
-    debug: false
-  controller.yaml: |
-    ...
-```
-
-The URL `http://easegress-0.easegress-hs.default:2380` defined in `cluster.primary-listen-peer-urls` is an address of one cluster member, which *secondary* instance can use to connect to the cluster.
-
-Deployment could look like this:
-
-```yaml
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  labels:
-    app: easegress-ingress
-  name: easegress-secondary
-  namespace: default
-spec:
-  replicas: 2
-  selector:
-    matchLabels:
-      app: easegress-ingress
-  template:
-    metadata:
-      labels:
-        app: easegress-ingress
-    spec:
-      serviceAccountName: easegress-ingress-controller
-      containers:
-      - args:
-        - -c
-        - |-
-          echo name: $EG_NAME > /opt/eg-config/config.yaml &&
-          cat /opt/eg-config/eg-secondary.yaml >> /opt/eg-config/config.yaml &&
-          /opt/easegress/bin/easegress-server \
-            -f /opt/eg-config/config.yaml \
-            --initial-object-config-files /opt/eg-config/controller.yaml
-        command:
-        - /bin/sh
-        env:
-        - name: EG_NAME
-          valueFrom:
-            fieldRef:
-              apiVersion: v1
-              fieldPath: metadata.name
-        image: megaease/easegress:latest
-        imagePullPolicy: IfNotPresent
-        name: easegress-secondary
-        ports:
-        - containerPort: 2381
-          name: admin-port
-          protocol: TCP
-        - containerPort: 2380
-          name: peer-port
-          protocol: TCP
-        - containerPort: 2379
-          name: client-port
-          protocol: TCP
-        resources:
-          limits:
-            cpu: 1200m
-            memory: 2Gi
-          requests:
-            cpu: 100m
-            memory: 256Mi
-        volumeMounts:
-        - mountPath: /opt/eg-config/eg-secondary.yaml
-          name: easegress-cluster-cm
-          subPath: eg-secondary.yaml
-        - mountPath: /opt/eg-config/controller.yaml
-          name: easegress-cluster-cm
-          subPath: controller.yaml
-        - mountPath: /opt/eg-data/
-          name: ingress-data-volume
-      restartPolicy: Always
-      volumes:
-      - emptyDir: {}
-        name: ingress-data-volume
-      - configMap:
-          defaultMode: 420
-          items:
-          - key: eg-secondary.yaml
-            path: eg-secondary.yaml
-          - key: controller.yaml
-            path: controller.yaml
-          name: easegress-cluster-cm
-        name: easegress-cluster-cm
----
-```
-
-Applying the deployment to the cluster creates two pods with prefix `easegress-secondary-`. We can again log into one of pods and see new *secondary* members. Let's check again the cluster members using `egctl`:
-
-```bash
-egctl --server {NODE_IP}:31255 member list |grep " name"
-```
-
-which should give something like following:
-
-```bash
-    # name: easegress-0
-    # name: easegress-1
-    # name: easegress-2
-    # name: easegress-secondary-758695b96c-2rzqt
-    # name: easegress-secondary-758695b96c-2vcpz
-```
-
-Note that secondary member names might something different, as we did not use headless ClusterIP for them.
-
-We now have 5 Easegress Ingress Controller instances running, handling the ingress traffic of our Kubernetes cluster.
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/helm-charts/ingress-controller/Chart.yaml b/helm-charts/ingress-controller/Chart.yaml
index 5033ba5b..77ed53c3 100644
--- a/helm-charts/ingress-controller/Chart.yaml
+++ b/helm-charts/ingress-controller/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
 name: easegress-ingress-controller
 type: application
-version: 1.0.0
+version: 1.0.1
 
diff --git a/helm-charts/ingress-controller/README.md b/helm-charts/ingress-controller/README.md
index 88bc3b33..bd968e84 100644
--- a/helm-charts/ingress-controller/README.md
+++ b/helm-charts/ingress-controller/README.md
@@ -9,14 +9,6 @@ Helm charts for installing Easegress ingress controller on Kubernetes
 kubectl create ns ingress-easegress
 ```
 
-### Prepare persistent volume (optional)
-
-If you are going to use persistent volumes, run following shell command on each persistent volume node:
-```bash
-sudo mkdir /opt/easegress
-sudo chmod 700 /opt/easegress
-```
-
 ## Usage
 ```shell
 # install with default values
@@ -29,40 +21,23 @@ helm install ingress-easegress -n ingress-easegress ./helm-charts/ingress-contro
   --set ingressClass.name=test-eg \
   --set controller.name=test-eg \
   --set 'controller.namespaces={ingress-easegress, default}'
-
-# install cluster of 3 primary and 2 secondary Easegress ingress instances
-helm install ingress-easegress -n ingress-easegress ./helm-charts/ingress-controller \
-  --set cluster.primaryReplicas=3 \
-  --set cluster.secondaryReplicas=2
-
-# install using persistentVolume on node with hostname "hostname-xyz"
-# to support recovery when pod crashes
-helm install ingress-easegress -n ingress-easegress ./helm-charts/ingress-controller \
-  --set cluster.volumeType=persistentVolume \
-  --set 'cluster.nodeHostnames={hostname-xyz}'
 ```
 
+You can reference [this example](/doc/reference/ingresscontroller.md#create-backend-service--kubernetes-ingress) for creating Kubernetes Ingress rules.
+
 ## Uninstall
 
 ```shell
 helm uninstall ingress-easegress -n ingress-easegress
-
-#sometimes helm does not delete pvc and pv. Delete manually each pvc.
-kubectl delete pvc easegress-pv-ingress-easegress-0 -n ingress-easegress
-# same for easegress-pv-ingress-easegress-i...n
 ```
 
 ## Parameters
 
-The following table lists the configurable parameters of the MegaEase Easegress Helm installation.
+The following table lists the configurable parameters of the Ingress Controller Helm installation.
 
 | Key | Type | Default | Description |
 |-----|------|---------|-------------|
-| service.nodePort | int | `30080` | nodePort for easegress service. |
-| service.adminPort | int | `31255` | nodePort for egctl access. |
-| cluster.primaryReplicas | int | `1` | number of easegress service that persists cluster data to disk |
-| cluster.volumeType | string | `emptyDir` | `emptyDir`: use pods internal filesystem that is not persisted when pod crashes. `persistentVolume`, create as many persistenVolumes and persistentVolumeClaims as there are nodeHostnames.
-| cluster.nodeHostnames | list | `[]` | nodeHostnames are hostnames of VMs/Kubernetes nodes. Only used when `volumeType: persistentVolume`. Note that this require nodes to be static. |
-| secondaryReplicas | int | `0` | number of easegress service that not persists cluster data to disk. |
+| service.nodePort | int | `30080` | nodePort for Easegress Ingress Controller. |
+| replicas | int | `1` | number of Easegress Ingress Controllers |
 
-> By default, k8s use range 30000-32767 for NodePort. Make sure you choose right port number.
\ No newline at end of file
+> By default, k8s use range 30000-32767 for NodePort. Make sure you choose right port number.
diff --git a/helm-charts/ingress-controller/templates/ConfigMap.yaml b/helm-charts/ingress-controller/templates/ConfigMap.yaml
index ae6b252e..c5b7f162 100644
--- a/helm-charts/ingress-controller/templates/ConfigMap.yaml
+++ b/helm-charts/ingress-controller/templates/ConfigMap.yaml
@@ -1,24 +1,11 @@
 apiVersion: v1
 data:
-  eg-primary.yaml: |
+  easegress-server.yaml: |
+    name: {{ .Release.Name }}
     cluster-name: easegress-ingress-controller
     cluster-role: primary
-    cluster:
-      listen-client-urls:
-      - http://0.0.0.0:2379
-      listen-peer-urls:
-      - http://0.0.0.0:2380
-      initial-cluster:
-      {{- $releasename := printf "%s" .Release.Name -}}
-      {{- $releasenamespace := printf "%s" .Release.Namespace -}}
-      {{- range $i, $e := until (.Values.cluster.primaryReplicas | int) }}
-      - {{ $releasename }}-{{$i}}: http://{{ $releasename }}-{{$i}}.easegress-hs.{{ $releasenamespace }}:2380
-      {{- end }}
     api-addr: 0.0.0.0:2381
     data-dir: /opt/easegress/data
-    wal-dir: ""
-    cpu-profile-file: ""
-    memory-profile-file: ""
     log-dir: /opt/easegress/log
     debug: false
   controller.yaml: |
@@ -37,19 +24,6 @@ data:
       keepAlive: true
       keepAliveTimeout: 60s
       maxConnections: 10240
-  eg-secondary.yaml: |
-    cluster-name: easegress-ingress-controller
-    cluster-role: secondary
-    cluster:
-      primary-listen-peer-urls:
-      - http://{{ .Release.Name }}-0.easegress-hs.{{ .Release.Namespace }}:2380
-    api-addr: 0.0.0.0:2381
-    data-dir: /opt/easegress/data
-    wal-dir: ""
-    cpu-profile-file: ""
-    memory-profile-file: ""
-    log-dir: /opt/easegress/log
-    debug: false
 kind: ConfigMap
 metadata:
   name: {{ .Release.Name }}
diff --git a/helm-charts/ingress-controller/templates/Deployment.yaml b/helm-charts/ingress-controller/templates/Deployment.yaml
index eafae4d1..f2993379 100644
--- a/helm-charts/ingress-controller/templates/Deployment.yaml
+++ b/helm-charts/ingress-controller/templates/Deployment.yaml
@@ -1,5 +1,4 @@
 
-{{ if gt ( .Values.cluster.secondaryReplicas | int ) 0 }}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -8,7 +7,7 @@ metadata:
   name: {{ .Release.Name }}
   namespace: {{ .Release.Namespace }}
 spec:
-  replicas: {{ .Values.cluster.secondaryReplicas }}
+  replicas: {{ .Values.replicas }}
   selector:
     matchLabels:
       app: {{ .Release.Name }}
@@ -22,11 +21,10 @@ spec:
       - args:
         - -c
         - |-
-          echo name: $EG_NAME > /opt/eg-config/config.yaml &&
-          cat /opt/eg-config/eg-secondary.yaml >> /opt/eg-config/config.yaml &&
           /opt/easegress/bin/easegress-server \
-            -f /opt/eg-config/config.yaml \
-            --initial-object-config-files /opt/eg-config/controller.yaml
+            -f /opt/eg-config/easegress-server.yaml \
+            --initial-object-config-files /opt/eg-config/controller.yaml \
+            --initial-cluster $(EG_NAME)=http://localhost:2380
         command:
         - /bin/sh
         env:
@@ -37,17 +35,7 @@ spec:
               fieldPath: metadata.name
         image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
         imagePullPolicy: IfNotPresent
-        name: easegress-secondary
-        ports:
-        - containerPort: 2381
-          name: admin-port
-          protocol: TCP
-        - containerPort: 2380
-          name: peer-port
-          protocol: TCP
-        - containerPort: 2379
-          name: client-port
-          protocol: TCP
+        name: ingress-easegress
         resources:
           limits:
             cpu: 1200m
@@ -56,9 +44,9 @@ spec:
             cpu: 100m
             memory: 256Mi
         volumeMounts:
-        - mountPath: /opt/eg-config/eg-secondary.yaml
+        - mountPath: /opt/eg-config/easegress-server.yaml
           name: {{ .Release.Name }}
-          subPath: eg-secondary.yaml
+          subPath: easegress-server.yaml
         - mountPath: /opt/eg-config/controller.yaml
           name: {{ .Release.Name }}
           subPath: controller.yaml
@@ -71,10 +59,9 @@ spec:
       - configMap:
           defaultMode: 420
           items:
-          - key: eg-secondary.yaml
-            path: eg-secondary.yaml
+          - key: easegress-server.yaml
+            path: easegress-server.yaml
           - key: controller.yaml
             path: controller.yaml
           name: {{ .Release.Name }}
         name: {{ .Release.Name }}
-{{ end }}
diff --git a/helm-charts/ingress-controller/templates/PersistentVolume.yaml b/helm-charts/ingress-controller/templates/PersistentVolume.yaml
deleted file mode 100644
index 75e78452..00000000
--- a/helm-charts/ingress-controller/templates/PersistentVolume.yaml
+++ /dev/null
@@ -1,36 +0,0 @@
-{{- if eq .Values.cluster.volumeType "persistentVolume" }}
-{{- range $i, $hostname := .Values.cluster.nodeHostnames }}
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: easegress-pv-{{$i}}
-spec:
-  capacity:
-    storage: 4Gi
-  volumeMode: Filesystem
-  accessModes:
-  - ReadWriteOnce
-  persistentVolumeReclaimPolicy: Delete
-  storageClassName: easegress-storage
-  local:
-    path: /opt/easegress
-  nodeAffinity:
-    required:
-      nodeSelectorTerms:
-      - matchExpressions:
-        - key: kubernetes.io/hostname
-          operator: In
-          values:
-          - {{$hostname}}
----
-{{ end }}
-
-apiVersion: storage.k8s.io/v1
-kind: StorageClass
-metadata:
-  name: easegress-storage
-provisioner: kubernetes.io/no-provisioner
-reclaimPolicy: Delete
-volumeBindingMode: WaitForFirstConsumer
-
-{{- end }}
diff --git a/helm-charts/ingress-controller/templates/Service.yaml b/helm-charts/ingress-controller/templates/Service.yaml
index 28a45a7a..b2936955 100644
--- a/helm-charts/ingress-controller/templates/Service.yaml
+++ b/helm-charts/ingress-controller/templates/Service.yaml
@@ -1,46 +1,3 @@
-apiVersion: v1
-kind: Service
-metadata:
-  name: easegress-hs
-  namespace: {{ .Release.Namespace }}
-spec:
-  clusterIP: None
-  ports:
-  - name: admin-port
-    port: 2381
-    protocol: TCP
-    targetPort: 2381
-  - name: peer-port
-    port: 2380
-    protocol: TCP
-    targetPort: 2380
-  - name: client-port
-    port: 2379
-    protocol: TCP
-    targetPort: 2379
-  selector:
-    app: {{ .Release.Name }}
-  type: ClusterIP
-
----
-
-apiVersion: v1
-kind: Service
-metadata:
-  name: easegress-public
-  namespace: {{ .Release.Namespace }}
-spec:
-  ports:
-  - name: admin-port
-    port: 2381
-    protocol: TCP
-    targetPort: 2381
-    nodePort: {{ .Values.service.adminPort }}
-  selector:
-    app: {{ .Release.Name }}
-  type: NodePort
-
----
 kind: Service
 apiVersion: v1
 metadata:
diff --git a/helm-charts/ingress-controller/templates/StatefulSet.yaml b/helm-charts/ingress-controller/templates/StatefulSet.yaml
deleted file mode 100644
index 43ff482e..00000000
--- a/helm-charts/ingress-controller/templates/StatefulSet.yaml
+++ /dev/null
@@ -1,99 +0,0 @@
-apiVersion: apps/v1
-kind: StatefulSet
-metadata:
-  labels:
-    name: {{ .Release.Name }}
-  name: {{ .Release.Name }}
-  namespace: {{ .Release.Namespace }}
-spec:
-  replicas: {{ .Values.cluster.primaryReplicas }}
-  selector:
-    matchLabels:
-      app: {{ .Release.Name }}
-  serviceName: easegress-hs
-  template:
-    metadata:
-      labels:
-        app: {{ .Release.Name }}
-    spec:
-      serviceAccountName: {{ .Release.Name }}
-      containers:
-      - args:
-        - -c
-        - |-
-          echo name: $EG_NAME > /opt/eg-config/config.yaml &&
-          cat /opt/eg-config/eg-primary.yaml >> /opt/eg-config/config.yaml && 
-          /opt/easegress/bin/easegress-server \
-            -f /opt/eg-config/config.yaml \
-            --advertise-client-urls http://$(EG_NAME).easegress-hs.{{ .Release.Namespace }}:2379 \
-            --initial-advertise-peer-urls http://$(EG_NAME).easegress-hs.{{ .Release.Namespace }}:2380 \
-            --initial-object-config-files /opt/eg-config/controller.yaml
-        command:
-        - /bin/sh
-        env:
-        - name: EG_NAME
-          valueFrom:
-            fieldRef:
-              apiVersion: v1
-              fieldPath: metadata.name
-        image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
-        imagePullPolicy: IfNotPresent
-        name: easegress-ingress
-        ports:
-        - containerPort: 2381
-          name: admin-port
-          protocol: TCP
-        - containerPort: 2380
-          name: peer-port
-          protocol: TCP
-        - containerPort: 2379
-          name: client-port
-          protocol: TCP
-        resources:
-          limits:
-            cpu: 1200m
-            memory: 2Gi
-          requests:
-            cpu: 100m
-            memory: 256Mi
-        volumeMounts:
-        - mountPath: /opt/eg-config/eg-primary.yaml
-          name: {{ .Release.Name }}
-          subPath: eg-primary.yaml
-        - mountPath: /opt/eg-config/controller.yaml
-          name: {{ .Release.Name }}
-          subPath: controller.yaml
-        - mountPath: /opt/easegress/data
-          name: {{ if eq .Values.cluster.volumeType "persistentVolume" }}easegress-pv {{ else }} ingress-data-volume {{ end }}
-        - mountPath: /opt/easegress/log
-          name: {{ if eq .Values.cluster.volumeType "persistentVolume" }}easegress-pv {{ else }} easegress-data-volume {{ end }}
-      restartPolicy: Always
-      volumes:
-      - emptyDir: {}
-        name: ingress-data-volume
-      - configMap:
-          defaultMode: 420
-          items:
-          - key: eg-primary.yaml
-            path: eg-primary.yaml
-          - key: controller.yaml
-            path: controller.yaml
-          name: {{ .Release.Name }}
-        name: {{ .Release.Name }}
-  updateStrategy:
-    type: RollingUpdate
-  {{- if eq .Values.cluster.volumeType "persistentVolume" }}
-  volumeClaimTemplates:
-  - apiVersion: v1
-    kind: PersistentVolumeClaim
-    metadata:
-      name: easegress-pv
-    spec:
-      accessModes:
-      - ReadWriteOnce
-      resources:
-        requests:
-          storage: 3Gi
-      storageClassName: easegress-storage
-      volumeMode: Filesystem
-  {{- end }}
diff --git a/helm-charts/ingress-controller/values.yaml b/helm-charts/ingress-controller/values.yaml
index 4c4cf9c4..c36a3a72 100644
--- a/helm-charts/ingress-controller/values.yaml
+++ b/helm-charts/ingress-controller/values.yaml
@@ -1,8 +1,6 @@
 service:
   # nodePort for easegress service
   nodePort: 30080
-  # nodeport for egctl access
-  adminPort: 31255
 
 # image for easegress deployment
 image:
@@ -19,18 +17,5 @@ controller:
   namespaces:
   - default
 
-# Easegress cluster configurations.
-# When volumeType is
-# - emptyDir: use pods internal filesystem that is not persisted when pod crashes.
-# - persistentVolume, create as many persistenVolumes and persistentVolumeClaims as there are nodeHostnames.
-cluster:
-  primaryReplicas: 1 # or 3,5,7
-  volumeType: emptyDir # or persistentVolume
-  # nodeHostnames are hostnames of VMs/Kubernetes nodes. Only used when `volumeType: persistentVolume`.
-  # Note that this require nodes to be static.
-  nodeHostnames:
-  # - hostname-1
-  # - hostname-2
-  # - hostname-3
-
-  secondaryReplicas: 0
+# number of Easegress IngressController replicas
+replicas: 1
diff --git a/pkg/api/cluster.go b/pkg/api/cluster.go
index 0aa3158a..5b7ffa4c 100644
--- a/pkg/api/cluster.go
+++ b/pkg/api/cluster.go
@@ -24,6 +24,10 @@ import (
 
 	yaml "gopkg.in/yaml.v2"
 
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/object/httpserver"
+	"github.com/megaease/easegress/pkg/object/rawconfigtrafficcontroller"
+	"github.com/megaease/easegress/pkg/object/trafficcontroller"
 	"github.com/megaease/easegress/pkg/supervisor"
 )
 
@@ -164,3 +168,53 @@ func (s *Server) _listStatusObjects() map[string]map[string]interface{} {
 
 	return status
 }
+
+func getSubStatusFromTrafficControllerStatus(status *trafficcontroller.Status, spec *supervisor.Spec) string {
+	for _, ns := range status.Specs {
+		if ns.Namespace != rawconfigtrafficcontroller.DefaultNamespace {
+			continue
+		}
+		if spec.Kind() == httpserver.Kind {
+			if val, ok := ns.HTTPServers[spec.Name()]; ok {
+				b, err := yaml.Marshal(val.Status)
+				if err != nil {
+					ClusterPanic(fmt.Errorf("unmarshal %v to yaml failed: %v", val.Status, err))
+				}
+				return string(b)
+			}
+			return ""
+		} else if spec.Kind() == httppipeline.Kind {
+			if val, ok := ns.HTTPPipelines[spec.Name()]; ok {
+				b, err := yaml.Marshal(val.Status)
+				if err != nil {
+					ClusterPanic(fmt.Errorf("unmarshal %v to yaml failed: %v", val.Status, err))
+				}
+				return string(b)
+			}
+			return ""
+		}
+	}
+	return ""
+}
+
+func (s *Server) _getStatusObjectFromTrafficController(name string, spec *supervisor.Spec) map[string]string {
+	prefix := s.cluster.Layout().StatusObjectPrefix(trafficcontroller.Kind)
+	kvs, err := s.cluster.GetPrefix(prefix)
+	if err != nil {
+		ClusterPanic(err)
+	}
+	status := &trafficcontroller.Status{}
+	ans := make(map[string]string)
+	for k, v := range kvs {
+		// different member
+		memberName := strings.TrimPrefix(k, prefix)
+
+		err = yaml.Unmarshal([]byte(v), status)
+		if err != nil {
+			ClusterPanic(fmt.Errorf("unmarshal %s to yaml failed: %v", v, err))
+		}
+		nsStatus := getSubStatusFromTrafficControllerStatus(status, spec)
+		ans[memberName] = nsStatus
+	}
+	return ans
+}
diff --git a/pkg/api/object.go b/pkg/api/object.go
index c9572b89..014a1241 100644
--- a/pkg/api/object.go
+++ b/pkg/api/object.go
@@ -26,6 +26,8 @@ import (
 	"github.com/go-chi/chi/v5"
 	yaml "gopkg.in/yaml.v2"
 
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/object/httpserver"
 	"github.com/megaease/easegress/pkg/supervisor"
 )
 
@@ -224,8 +226,13 @@ func (s *Server) getStatusObject(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// NOTE: Maybe inconsistent, the object was deleted already here.
-	status := s._getStatusObject(name)
+	var status map[string]string
+	if spec.Kind() == httpserver.Kind || spec.Kind() == httppipeline.Kind {
+		status = s._getStatusObjectFromTrafficController(name, spec)
+	} else {
+		// NOTE: Maybe inconsistent, the object was deleted already here.
+		status = s._getStatusObject(name)
+	}
 
 	buff, err := yaml.Marshal(status)
 	if err != nil {
diff --git a/pkg/cluster/config.go b/pkg/cluster/config.go
index bde108ae..366ba346 100644
--- a/pkg/cluster/config.go
+++ b/pkg/cluster/config.go
@@ -40,6 +40,10 @@ const (
 
 	maxTxnOps       = 10240
 	maxRequestBytes = 10 * 1024 * 1024 // 10MB
+
+	// Threshold for number of changes etcd stores in memory before creating a new snapshot.
+	// Reference: https://etcd.io/docs/v3.5/tuning/#snapshot-tuning
+	snapshotCount = 5000
 )
 
 var (
@@ -90,6 +94,7 @@ func CreateStaticClusterEtcdConfig(opt *option.Options) (*embed.Config, error) {
 	ec.QuotaBackendBytes = quotaBackendBytes
 	ec.MaxTxnOps = maxTxnOps
 	ec.MaxRequestBytes = maxRequestBytes
+	ec.SnapshotCount = snapshotCount
 	ec.Logger = "zap"
 	ec.LogOutputs = []string{common.NormalizeZapLogPath(filepath.Join(opt.AbsLogDir, logFilename))}
 
@@ -147,6 +152,7 @@ func CreateEtcdConfig(opt *option.Options, members *members) (*embed.Config, err
 	ec.QuotaBackendBytes = quotaBackendBytes
 	ec.MaxTxnOps = maxTxnOps
 	ec.MaxRequestBytes = maxRequestBytes
+	ec.SnapshotCount = snapshotCount
 	ec.Logger = "zap"
 	ec.LogOutputs = []string{common.NormalizeZapLogPath(filepath.Join(opt.AbsLogDir, logFilename))}
 
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/context/httpresponse.go b/pkg/context/httpresponse.go
index 30ebfc1d..78a544f6 100644
--- a/pkg/context/httpresponse.go
+++ b/pkg/context/httpresponse.go
@@ -155,6 +155,11 @@ func (w *httpResponse) FlushedBodyBytes() uint64 {
 }
 
 func (w *httpResponse) finish() {
+	// copy backend http response header
+	// NOTE: this copy should call before WriteHeader
+	w.header.VisitAll(func(k, v string) {
+		w.std.Header().Add(k, v)
+	})
 	// NOTE: WriteHeader must be called at most one time.
 	w.std.WriteHeader(w.StatusCode())
 	w.flushBody()
diff --git a/pkg/context/mqttcontext.go b/pkg/context/mqttcontext.go
index e38186d1..e0c49146 100644
--- a/pkg/context/mqttcontext.go
+++ b/pkg/context/mqttcontext.go
@@ -38,9 +38,11 @@ type (
 		Finish()
 
 		PacketType() MQTTPacketType
-		ConnectPacket() *packets.ConnectPacket       // read only
-		PublishPacket() *packets.PublishPacket       // read only
-		DisconnectPacket() *packets.DisconnectPacket // read only
+		ConnectPacket() *packets.ConnectPacket         // read only
+		PublishPacket() *packets.PublishPacket         // read only
+		DisconnectPacket() *packets.DisconnectPacket   // read only
+		SubscribePacket() *packets.SubscribePacket     // read only
+		UnsubscribePacket() *packets.UnsubscribePacket // read only
 
 		SetDrop()         // set drop value to true
 		Drop() bool       // if true, this mqtt packet will be dropped
@@ -72,14 +74,12 @@ type (
 		ctx        stdcontext.Context
 		cancelFunc stdcontext.CancelFunc
 
-		startTime        time.Time
-		endTime          time.Time
-		client           MQTTClient
-		backend          MQTTBackend
-		connectPacket    *packets.ConnectPacket
-		publishPacket    *packets.PublishPacket
-		disconnectPacket *packets.DisconnectPacket
-		packetType       MQTTPacketType
+		startTime  time.Time
+		endTime    time.Time
+		client     MQTTClient
+		backend    MQTTBackend
+		packet     packets.ControlPacket
+		packetType MQTTPacketType
 
 		err        error
 		drop       int32
@@ -103,6 +103,12 @@ const (
 	// MQTTDisconnect is mqtt packet type of disconnect
 	MQTTDisconnect = 3
 
+	// MQTTSubscribe is mqtt packet type of subscribe
+	MQTTSubscribe = 4
+
+	// MQTTUnsubscribe is mqtt packet type of unsubscribe
+	MQTTUnsubscribe = 5
+
 	// MQTTOther is all other mqtt packet type
 	MQTTOther MQTTPacketType = 99
 )
@@ -121,19 +127,21 @@ func NewMQTTContext(ctx stdcontext.Context, backend MQTTBackend, client MQTTClie
 		backend:    backend,
 	}
 
-	switch packet := packet.(type) {
+	switch packet.(type) {
 	case *packets.ConnectPacket:
-		mqttCtx.connectPacket = packet
 		mqttCtx.packetType = MQTTConnect
 	case *packets.PublishPacket:
-		mqttCtx.publishPacket = packet
 		mqttCtx.packetType = MQTTPublish
 	case *packets.DisconnectPacket:
-		mqttCtx.disconnectPacket = packet
 		mqttCtx.packetType = MQTTDisconnect
+	case *packets.SubscribePacket:
+		mqttCtx.packetType = MQTTSubscribe
+	case *packets.UnsubscribePacket:
+		mqttCtx.packetType = MQTTUnsubscribe
 	default:
 		mqttCtx.packetType = MQTTOther
 	}
+	mqttCtx.packet = packet
 	return mqttCtx
 }
 
@@ -215,15 +223,23 @@ func (ctx *mqttContext) PacketType() MQTTPacketType {
 }
 
 func (ctx *mqttContext) ConnectPacket() *packets.ConnectPacket {
-	return ctx.connectPacket
+	return ctx.packet.(*packets.ConnectPacket)
 }
 
 func (ctx *mqttContext) PublishPacket() *packets.PublishPacket {
-	return ctx.publishPacket
+	return ctx.packet.(*packets.PublishPacket)
 }
 
 func (ctx *mqttContext) DisconnectPacket() *packets.DisconnectPacket {
-	return ctx.disconnectPacket
+	return ctx.packet.(*packets.DisconnectPacket)
+}
+
+func (ctx *mqttContext) SubscribePacket() *packets.SubscribePacket {
+	return ctx.packet.(*packets.SubscribePacket)
+}
+
+func (ctx *mqttContext) UnsubscribePacket() *packets.UnsubscribePacket {
+	return ctx.packet.(*packets.UnsubscribePacket)
 }
 
 func (ctx *mqttContext) SetDrop() {
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/logger/logger.go b/pkg/logger/logger.go
index 02e508c6..d140a827 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -92,7 +92,11 @@ func EtcdClientLoggerConfig(opt *option.Options, filename string) *zap.Config {
 	encoderConfig := defaultEncoderConfig()
 
 	level := zap.NewAtomicLevel()
-	level.SetLevel(zapcore.DebugLevel)
+	if opt.Debug {
+		level.SetLevel(zapcore.DebugLevel)
+	} else {
+		level.SetLevel(zapcore.InfoLevel)
+	}
 
 	outputPaths := []string{common.NormalizeZapLogPath(filepath.Join(opt.AbsLogDir, filename))}
 
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/client.go b/pkg/object/mqttproxy/client.go
index ce7999da..98142396 100644
--- a/pkg/object/mqttproxy/client.go
+++ b/pkg/object/mqttproxy/client.go
@@ -21,6 +21,7 @@ import (
 	stdcontext "context"
 	"errors"
 	"net"
+	"reflect"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -45,6 +46,25 @@ const (
 	QoS2 byte = 2
 )
 
+type processFn func(*Client, packets.ControlPacket)
+type processFnWithErr func(*Client, packets.ControlPacket) error
+
+var processPacketMap = map[string]processFnWithErr{
+	"*packets.ConnectPacket":     errorWrapper("double connect"),
+	"*packets.ConnackPacket":     errorWrapper("client should not send connack"),
+	"*packets.PubrecPacket":      errorWrapper("qos2 not support now"),
+	"*packets.PubrelPacket":      errorWrapper("qos2 not support now"),
+	"*packets.PubcompPacket":     errorWrapper("qos2 not support now"),
+	"*packets.SubackPacket":      errorWrapper("broker not subscribe"),
+	"*packets.UnsubackPacket":    errorWrapper("broker not unsubscribe"),
+	"*packets.PingrespPacket":    errorWrapper("broker not ping"),
+	"*packets.PublishPacket":     pipelineWrapper(processPublish),
+	"*packets.SubscribePacket":   pipelineWrapper(processSubscribe),
+	"*packets.UnsubscribePacket": pipelineWrapper(processUnsubscribe),
+	"*packets.PingreqPacket":     nilErrWrapper(processPingreq),
+	"*packets.PubackPacket":      nilErrWrapper(processPuback),
+}
+
 type (
 	// ClientInfo is basic information for client
 	ClientInfo struct {
@@ -173,37 +193,12 @@ func (c *Client) readLoop() {
 }
 
 func (c *Client) processPacket(packet packets.ControlPacket) error {
-	var err error
-	switch p := packet.(type) {
-	case *packets.ConnectPacket:
-		err = errors.New("double connect")
-	case *packets.ConnackPacket:
-		err = errors.New("client send connack")
-	case *packets.PublishPacket:
-		c.processPublish(p)
-	case *packets.PubackPacket:
-		c.processPuback(p)
-	case *packets.PubrecPacket, *packets.PubrelPacket, *packets.PubcompPacket:
-		err = errors.New("qos2 not support now")
-	case *packets.SubscribePacket:
-		c.processSubscribe(p)
-	case *packets.SubackPacket:
-		err = errors.New("broker not subscribe")
-	case *packets.UnsubscribePacket:
-		c.processUnsubscribe(p)
-	case *packets.UnsubackPacket:
-		err = errors.New("broker not unsubscribe")
-	case *packets.PingreqPacket:
-		c.processPingreq(p)
-	case *packets.PingrespPacket:
-		err = errors.New("broker not ping")
-	default:
-		err = errors.New("unknown packet")
+	packetType := reflect.TypeOf(packet).String()
+	fn, ok := processPacketMap[packetType]
+	if !ok {
+		return errors.New("unknown packet")
 	}
-	if err != nil {
-		logger.SpanDebugf(nil, "client %v process packet failed, %v", c.info.cid, err)
-	}
-	return err
+	return fn(c, packet)
 }
 
 func (c *Client) checkPublishLimit(publish *packets.PublishPacket) bool {
@@ -211,86 +206,29 @@ func (c *Client) checkPublishLimit(publish *packets.PublishPacket) bool {
 	return c.publishLimit.acquirePermission(size)
 }
 
-func (c *Client) processPublish(publish *packets.PublishPacket) {
-	logger.SpanDebugf(nil, "client %s process publish %v", c.info.cid, publish.TopicName)
-
-	if !c.checkPublishLimit(publish) {
-		logger.SpanErrorf(nil, "client %v publish limiter drop packet %v", c.info.cid, publish.TopicName)
-		return
-	}
-	if c.broker.pipeline != "" {
-		pipe, err := pipeline.GetPipeline(c.broker.pipeline, context.MQTT)
-		if err != nil {
-			logger.SpanErrorf(nil, "get pipeline %v failed, %v", c.broker.pipeline, err)
-		} else {
-			ctx := context.NewMQTTContext(stdcontext.Background(), c.broker.backend, c, publish)
-			pipe.HandleMQTT(ctx)
-			if ctx.Disconnect() {
-				logger.SpanDebugf(nil, "client %v set disconnect during process publish", c.info.cid)
-				c.close()
-				return
-			}
-			if ctx.Drop() {
-				logger.SpanDebugf(nil, "client %v drop packet %v during process publish", c.info.cid, publish.TopicName)
-				return
-			}
-		}
+// runPipeline will run MQTT pipeline by using packet.
+// it will return an error if MQTT pipline set MQTTContext to Disconnect or Drop.
+func (c *Client) runPipeline(packet packets.ControlPacket) error {
+	if c.broker.pipeline == "" {
+		return nil
 	}
 
-	err := c.broker.backend.publish(publish)
+	pipe, err := pipeline.GetPipeline(c.broker.pipeline, context.MQTT)
 	if err != nil {
-		logger.SpanErrorf(nil, "client %v publish %v failed: %v", c.info.cid, publish.TopicName, err)
+		logger.SpanErrorf(nil, "get pipeline %v failed, %v", c.broker.pipeline, err)
+		return nil
 	}
-	switch publish.Qos {
-	case QoS0:
-		// do nothing
-	case QoS1:
-		puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
-		puback.MessageID = publish.MessageID
-		c.writePacket(puback)
-	case QoS2:
-		// not support yet
-	}
-}
-
-func (c *Client) processPuback(puback *packets.PubackPacket) {
-	c.session.puback(puback)
-}
 
-func (c *Client) processSubscribe(packet *packets.SubscribePacket) {
-	logger.SpanDebugf(nil, "client %s subscribe %v with qos %v", c.info.cid, packet.Topics, packet.Qoss)
-	err := c.broker.topicMgr.subscribe(packet.Topics, packet.Qoss, c.info.cid)
-	if err != nil {
-		logger.SpanErrorf(nil, "client %v subscribe %v failed: %v", c.info.cid, packet.Topics, err)
-		return
+	ctx := context.NewMQTTContext(stdcontext.Background(), c.broker.backend, c, packet)
+	pipe.HandleMQTT(ctx)
+	if ctx.Disconnect() {
+		c.close()
+		return errors.New("pipeline set disconnect")
 	}
-	c.session.subscribe(packet.Topics, packet.Qoss)
-
-	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
-	suback.MessageID = packet.MessageID
-	suback.ReturnCodes = make([]byte, len(packet.Topics))
-	for i := range packet.Topics {
-		suback.ReturnCodes[i] = packet.Qos
-	}
-	c.writePacket(suback)
-}
-
-func (c *Client) processUnsubscribe(packet *packets.UnsubscribePacket) {
-	logger.SpanDebugf(nil, "client %s processUnsubscribe %v", c.info.cid, packet.Topics)
-	err := c.broker.topicMgr.unsubscribe(packet.Topics, c.info.cid)
-	if err != nil {
-		logger.SpanErrorf(nil, "client %v unsubscribe %v failed: %v", c.info.cid, packet.Topics, err)
+	if ctx.Drop() {
+		return errors.New("pipeline set drop")
 	}
-	c.session.unsubscribe(packet.Topics)
-
-	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
-	unsuback.MessageID = packet.MessageID
-	c.writePacket(unsuback)
-}
-
-func (c *Client) processPingreq(packet *packets.PingreqPacket) {
-	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
-	c.writePacket(resp)
+	return nil
 }
 
 func (c *Client) writePacket(packet packets.ControlPacket) {
@@ -351,3 +289,99 @@ func (c *Client) closeAndDelSession() {
 
 	c.close()
 }
+
+func errorWrapper(errMsg string) processFnWithErr {
+	return func(c *Client, p packets.ControlPacket) error {
+		return errors.New(errMsg)
+	}
+}
+
+func nilErrWrapper(fn processFn) processFnWithErr {
+	return func(c *Client, p packets.ControlPacket) error {
+		fn(c, p)
+		return nil
+	}
+}
+
+func pipelineWrapper(fn processFn) processFnWithErr {
+	return func(c *Client, p packets.ControlPacket) error {
+		err := c.runPipeline(p)
+		if err != nil {
+			logger.SpanDebugf(nil, "client process pipeline failed, %v", c.info.cid, err)
+			return nil
+		}
+		fn(c, p)
+		return nil
+	}
+}
+
+func processPublish(c *Client, packet packets.ControlPacket) {
+	publish := packet.(*packets.PublishPacket)
+	logger.SpanDebugf(nil, "client %s process publish %v", c.info.cid, publish.TopicName)
+
+	if !c.checkPublishLimit(publish) {
+		logger.SpanErrorf(nil, "client %v publish limiter drop packet %v", c.info.cid, publish.TopicName)
+		return
+	}
+
+	err := c.broker.backend.publish(publish)
+	if err != nil {
+		logger.SpanErrorf(nil, "client %v publish %v failed: %v", c.info.cid, publish.TopicName, err)
+	}
+	switch publish.Qos {
+	case QoS0:
+		// do nothing
+	case QoS1:
+		puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
+		puback.MessageID = publish.MessageID
+		c.writePacket(puback)
+	case QoS2:
+		// not support yet
+	}
+}
+
+func processPuback(c *Client, packet packets.ControlPacket) {
+	puback := packet.(*packets.PubackPacket)
+	c.session.puback(puback)
+}
+
+func processSubscribe(c *Client, p packets.ControlPacket) {
+	packet := p.(*packets.SubscribePacket)
+	logger.SpanDebugf(nil, "client %s subscribe %v with qos %v", c.info.cid, packet.Topics, packet.Qoss)
+
+	err := c.broker.topicMgr.subscribe(packet.Topics, packet.Qoss, c.info.cid)
+	if err != nil {
+		logger.SpanErrorf(nil, "client %v subscribe %v failed: %v", c.info.cid, packet.Topics, err)
+		return
+	}
+	c.session.subscribe(packet.Topics, packet.Qoss)
+
+	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
+	suback.MessageID = packet.MessageID
+	suback.ReturnCodes = make([]byte, len(packet.Topics))
+	for i := range packet.Topics {
+		suback.ReturnCodes[i] = packet.Qos
+	}
+	c.writePacket(suback)
+}
+
+func processUnsubscribe(c *Client, p packets.ControlPacket) {
+	packet := p.(*packets.UnsubscribePacket)
+
+	logger.SpanDebugf(nil, "client %s processUnsubscribe %v", c.info.cid, packet.Topics)
+
+	err := c.broker.topicMgr.unsubscribe(packet.Topics, c.info.cid)
+	if err != nil {
+		logger.SpanErrorf(nil, "client %v unsubscribe %v failed: %v", c.info.cid, packet.Topics, err)
+	}
+	c.session.unsubscribe(packet.Topics)
+
+	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
+	unsuback.MessageID = packet.MessageID
+	c.writePacket(unsuback)
+}
+
+func processPingreq(c *Client, packet packets.ControlPacket) {
+	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
+	c.writePacket(resp)
+}
diff --git a/pkg/object/pipeline/mock.go b/pkg/object/pipeline/mock.go
index a432702a..dbddc8a2 100644
--- a/pkg/object/pipeline/mock.go
+++ b/pkg/object/pipeline/mock.go
@@ -47,9 +47,11 @@ type MockMQTTFilter struct {
 	mockFilter
 	mu sync.Mutex
 
-	spec       *MockMQTTSpec
-	clients    map[string]int
-	disconnect map[string]struct{}
+	spec        *MockMQTTSpec
+	clients     map[string]int
+	disconnect  map[string]struct{}
+	subscribe   map[string][]string
+	unsubscribe map[string][]string
 }
 
 // MockMQTTSpec is spec of MockMQTTFilter
@@ -68,6 +70,8 @@ type MockMQTTSpec struct {
 type MockMQTTStatus struct {
 	ClientCount      map[string]int
 	ClientDisconnect map[string]struct{}
+	Subscribe        map[string][]string
+	Unsubscribe      map[string][]string
 }
 
 var _ MQTTFilter = (*MockMQTTFilter)(nil)
@@ -87,28 +91,33 @@ func (m *MockMQTTFilter) Init(filterSpec *FilterSpec) {
 	m.spec = filterSpec.FilterSpec().(*MockMQTTSpec)
 	m.clients = make(map[string]int)
 	m.disconnect = make(map[string]struct{})
+	m.subscribe = make(map[string][]string)
+	m.unsubscribe = make(map[string][]string)
 }
 
 // HandleMQTT handle MQTTContext
 func (m *MockMQTTFilter) HandleMQTT(ctx context.MQTTContext) *context.MQTTResult {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-
 	m.clients[ctx.Client().ClientID()]++
-	if ctx.PacketType() == context.MQTTConnect {
+
+	switch ctx.PacketType() {
+	case context.MQTTConnect:
+		ctx.Client().Store(m.spec.ConnectKey, struct{}{})
 		if ctx.ConnectPacket().Username != m.spec.UserName || string(ctx.ConnectPacket().Password) != m.spec.Password {
 			ctx.SetDisconnect()
 		}
-	}
-	if ctx.PacketType() == context.MQTTDisconnect {
+	case context.MQTTDisconnect:
 		m.disconnect[ctx.Client().ClientID()] = struct{}{}
+	case context.MQTTSubscribe:
+		m.subscribe[ctx.Client().ClientID()] = ctx.SubscribePacket().Topics
+	case context.MQTTUnsubscribe:
+		m.unsubscribe[ctx.Client().ClientID()] = ctx.UnsubscribePacket().Topics
 	}
+
 	for _, k := range m.spec.KeysToStore {
 		ctx.Client().Store(k, struct{}{})
 	}
-	if ctx.PacketType() == context.MQTTConnect {
-		ctx.Client().Store(m.spec.ConnectKey, struct{}{})
-	}
 	if m.spec.PublishBackendClientID {
 		backend := ctx.Backend()
 		backend.Publish(ctx.Client().ClientID(), nil, nil)
@@ -116,31 +125,36 @@ func (m *MockMQTTFilter) HandleMQTT(ctx context.MQTTContext) *context.MQTTResult
 	return nil
 }
 
-func (m *MockMQTTFilter) clientCount() map[string]int {
+// Status return status of MockMQTTFilter
+func (m *MockMQTTFilter) Status() interface{} {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	ans := make(map[string]int)
+
+	clientCount := make(map[string]int)
 	for k, v := range m.clients {
-		ans[k] = v
+		clientCount[k] = v
 	}
-	return ans
-}
-
-func (m *MockMQTTFilter) clientDisconnect() map[string]struct{} {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	ans := make(map[string]struct{})
-	for k, v := range m.disconnect {
-		ans[k] = v
+	disconnect := make(map[string]struct{})
+	for k := range m.disconnect {
+		disconnect[k] = struct{}{}
+	}
+	subscribe := make(map[string][]string)
+	for k, v := range m.subscribe {
+		vv := make([]string, len(v))
+		copy(vv, v)
+		subscribe[k] = v
+	}
+	unsubscribe := make(map[string][]string)
+	for k, v := range m.unsubscribe {
+		vv := make([]string, len(v))
+		copy(vv, v)
+		unsubscribe[k] = vv
 	}
-	return ans
-}
-
-// Status return status of MockMQTTFilter
-func (m *MockMQTTFilter) Status() interface{} {
 	return MockMQTTStatus{
-		ClientCount:      m.clientCount(),
-		ClientDisconnect: m.clientDisconnect(),
+		ClientCount:      clientCount,
+		ClientDisconnect: disconnect,
+		Subscribe:        subscribe,
+		Unsubscribe:      unsubscribe,
 	}
 }
 
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/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 4bd0f6d5..a74df63f 100644
--- a/pkg/object/mqttproxy/mqtt_test.go
+++ b/pkg/object/mqttproxy/mqtt_test.go
@@ -1613,6 +1613,13 @@ filters:
 		text := "text"
 		token := client.Publish(topic, 1, false, text)
 		token.Wait()
+
+		token = client.Subscribe(strconv.Itoa(i), 1, nil)
+		token.Wait()
+
+		token = client.Unsubscribe(strconv.Itoa(i))
+		token.Wait()
+
 		p := broker.backend.(*testMQ).get()
 		if p.TopicName != topic || string(p.Payload) != text {
 			t.Errorf("get wrong publish")
@@ -1644,6 +1651,12 @@ filters:
 	if len(filterStatus.ClientDisconnect) != clientNum {
 		t.Errorf("filter get wrong result %v for client disconnect", filterStatus.ClientDisconnect)
 	}
+	if len(filterStatus.Subscribe) != clientNum {
+		t.Errorf("filter get wrong result %v for client subscribe", filterStatus.Subscribe)
+	}
+	if len(filterStatus.Unsubscribe) != clientNum {
+		t.Errorf("filter get wrong result %v for client unsubscribe", filterStatus.Unsubscribe)
+	}
 }
 
 func TestAuthByPipeline(t *testing.T) {
diff --git a/pkg/object/pipeline/pipeline_test.go b/pkg/object/pipeline/pipeline_test.go
index 0cd02ec7..951b90d5 100644
--- a/pkg/object/pipeline/pipeline_test.go
+++ b/pkg/object/pipeline/pipeline_test.go
@@ -147,12 +147,26 @@ filters:
 			assert.Equal(true, ok)
 			_, ok = ctx.Client().Load("filter")
 			assert.Equal(true, ok)
+
+			subscribe := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
+			subscribe.Topics = []string{strconv.Itoa(i)}
+			ctx = context.NewMQTTContext(stdcontext.Background(), backend, c, subscribe)
+			p.HandleMQTT(ctx)
+
+			unsubscribe := packets.NewControlPacket(packets.Unsubscribe).(*packets.UnsubscribePacket)
+			subscribe.Topics = []string{strconv.Itoa(i)}
+			ctx = context.NewMQTTContext(stdcontext.Background(), backend, c, unsubscribe)
+			p.HandleMQTT(ctx)
+
 			wg.Done()
 		}(i)
 	}
 	wg.Wait()
 	f := p.getRunningFilter("mqtt-filter").filter.(*MockMQTTFilter)
-	assert.Equal(len(f.Status().(MockMQTTStatus).ClientCount), 1000, "wrong client count")
+	status := f.Status().(MockMQTTStatus)
+	assert.Equal(len(status.ClientCount), 1000, "wrong client count")
+	assert.Equal(1000, len(status.Subscribe))
+	assert.Equal(1000, len(status.Unsubscribe))
 
 	assert.Equal(1000, len(backend.Messages))
 
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")
 	}