Repository: megaease/easegress
Pull request #399 (2021-12-06 07:22:30)

Merge the new pipeline
# merge new pipeline

## main changes

1. add new `BusinessController` `Pipeline`, its structure is based on original `HTTPPipeline`. For now, `Pipeline` only support MQTT protocol. `HTTPPipeline` still works for HTTP protocol.
2. the new `Pipeline` part contain changes in `pkg/context` (add `MQTTContext`), a new MQTT filter called `ConnectControl` in `pkg/filter/connectcontrol` and add pipeline for MQTTProxy in `pkg/object/mqttproxy`. The new pipeline and MQTT filter contain tests. See test files for more information.
3. Update for `MQTTProxy` during add pipeline for MQTT protocol:
    - run pipeline for MQTT `connect`, `disconnect` and `publish` packets.
    - add shareable kv map for MQTT client, for example, store kv values when `connect` and use these values when `disconnect`.
    - allow MQTT filter to send message to backend.
    - add max-allowed connection, max-allowed publish rate to MQTT proxy.
    - add zipkin b3 logging to MQTTProxy log.
    - use etcd watcher to allow delete running MQTT client. (reconnect and update state when watcher disconnect).

## file changes (main changes)

cluster:

1. `pkg/cluster_interface.go`: add `GetWithOp` and `WatchWithOp`, allow pass etcd client options to speed up get and watch, for example, get prefix and only return keys not values, or not watch put event etc.
2. `pkg/cluster/op.go`: impl `GetWithOp` based on previous `Get`.
3. `pkg/cluster/watcher.go`: impl `WatchWithOp` based on previous `Watch`.

context:

1. `pkg/context/context.go`: add new context interface for HTTP, MQTT, TCP.
2. `pkg/context/mqttcontext.go`: impl `MQTTContext`.

filter:

1. `pkg/filter/connectcontrol/connectcontrol.go`: MQTT filter for connect control.

logger:

1. `pkg/logger/export.go`: add `Span` wrapper for default logger to print logs with `zipkin` b3 tracing.
2. `pkg/logger/logger.go`: add `InitMock` for logging, when write test, use this to init logger will print logs to stdout, mainly for debug when write tests.

mqttproxy `pkg/object/mqttproxy`

1. `backend.go`: add `Publish` method for filters to directly send message to backend.
2. `broker.go`: allow do authentication by using pipeline (`AuthByPipeline` part). `watchDelete` and `reconnectWatcher` used for watch etcd delete certain session and disconnect their client (highly related to impl in `mqttproxy/storage.go`). `checkConnectPermission` used for rate limiter (highly related to impl in `mqttproxy/ratelimiter.go`). In `handleConn`, add pipeline and rate limit. In `httpGetAllSessionHandler` and `httpDeleteSessionHandler` used to get all session and delete certain session.
3. `client.go` mainly add pipeline when close.
4. `mqttproxy.go` panic when broker is nil to avoid running panic.
5. `ratelimiter.go` impl `Limiter` based on `util/ratelimiter`.
6. `storage.go` add `watchDelete` method to watch etcd delete values in certain prefix.

pipeline `pkg/object/pipeline`

1. `mock.go` add a mock filter for test. Main logic in `HandleMQTT` function.
2. other files in `pipeline` mainly based on previous `HTTPPipeline` but add a spec and method for `Protocol`.


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

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

```diff
@@            Coverage Diff             @@
##             main     #399      +/-   ##
==========================================
- Coverage   80.82%   80.11%   -0.71%     
==========================================
  Files          60       69       +9     
  Lines        6966     7922     +956     
==========================================
+ Hits         5630     6347     +717     
- Misses       1045     1229     +184     
- Partials      291      346      +55     
```


| [Impacted Files](https://codecov.io/gh/megaease/easegress/pull/399?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | Coverage Δ | |
|---|---|---|
| [pkg/object/mqttproxy/mqttproxy.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvbXF0dHByb3h5Lmdv) | `22.22% <0.00%> (+7.03%)` | :arrow_up: |
| [pkg/object/mqttproxy/spec.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvc3BlYy5nbw==) | `100.00% <ø> (ø)` | |
| [pkg/object/mqttproxy/session\_manager.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvc2Vzc2lvbl9tYW5hZ2VyLmdv) | `86.36% <33.33%> (ø)` | |
| [pkg/object/mqttproxy/session.go](https://codecov.io/gh/megaease/easegress/pull/399/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% <45.45%> (-8.09%)` | :arrow_down: |
| [pkg/object/mqttproxy/storage.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvc3RvcmFnZS5nbw==) | `73.23% <55.26%> (-21.64%)` | :arrow_down: |
| [pkg/object/mqttproxy/client.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvY2xpZW50Lmdv) | `78.60% <68.57%> (-2.50%)` | :arrow_down: |
| [pkg/object/mqttproxy/broker.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvYnJva2VyLmdv) | `76.00% <70.11%> (-3.68%)` | :arrow_down: |
| [pkg/cluster/op.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvb3AuZ28=) | `64.96% <71.42%> (+1.17%)` | :arrow_up: |
| [pkg/object/pipeline/spec.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9waXBlbGluZS9zcGVjLmdv) | `71.42% <71.42%> (ø)` | |
| [pkg/cluster/cluster\_interface.go](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY2x1c3Rlcl9pbnRlcmZhY2UuZ28=) | `76.92% <76.92%> (ø)` | |
| ... and [18 more](https://codecov.io/gh/megaease/easegress/pull/399/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/megaease/easegress/pull/399?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/399?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Last update [efd3d6a...6465ac7](https://codecov.io/gh/megaease/easegress/pull/399?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease).


Patch:
diff --git a/doc/reference/controllers.md b/doc/reference/controllers.md
index 051d76c4..0422a7ba 100644
--- a/doc/reference/controllers.md
+++ b/doc/reference/controllers.md
@@ -32,7 +32,7 @@
     - [nacos.ServerSpec](#nacosserverspec)
     - [autocertmanager.DomainSpec](#autocertmanagerdomainspec)
 
-As the [architecture diagram](./imgs/architecture.png) shows, the controller is the core entity to control kinds of working. There are two kinds of controllers overall:
+As the [architecture diagram](../imgs/architecture.png) shows, the controller is the core entity to control kinds of working. There are two kinds of controllers overall:
 
 - System Controller: It is created one and only one instance in every Easegress node, which can't be deleted. They mainly aim to control essential system-level stuff.
 - Business Controller: It could be created, updated, deleted by admin operation. They control various resources such as mesh traffic, service discovery, faas, and so on.
diff --git a/helm-charts/easegress/README.md b/helm-charts/easegress/README.md
index c245af63..16fd8ba3 100644
--- a/helm-charts/easegress/README.md
+++ b/helm-charts/easegress/README.md
@@ -10,6 +10,15 @@ kubectl create ns easegress
 # update common helm charts
 helm dependency update ./helm-charts/easegress
 ```
+
+### Prepare persistent volume (optional)
+
+If you are going to use persistent volumes, run following shell command on each persistent volume node:
+```shell
+sudo mkdir /opt/easegress
+sudo chmod 700 /opt/easegress
+```
+
 ## Usage
 ```shell
 
@@ -39,3 +48,13 @@ Add filters and objects to Easegress:
 egctl --server {NODE_IP}:31255 object create -f pipeline.yaml
 ```
 where NODE_IP is the IP address a node running Easegress pod and `pipeline.yaml` Easegress object definition.
+
+## Uninstall
+
+```shell
+helm uninstall easegress -n easegress
+
+# sometimes helm does not delete pvc and pv. Delete manually each pvc.
+kubectl delete pvc easegress-pv-easegress-0 -n easegress
+# same for easegress-pv-easegress-i...n
+```
diff --git a/helm-charts/easegress/templates/Deployment.yaml b/helm-charts/easegress/templates/Deployment.yaml
index 0b6cc343..7523517b 100644
--- a/helm-charts/easegress/templates/Deployment.yaml
+++ b/helm-charts/easegress/templates/Deployment.yaml
@@ -57,7 +57,7 @@ spec:
         - mountPath: /opt/eg-config/eg-secondary.yaml
           name: {{ .Release.Name }}
           subPath: eg-secondary.yaml
-        - mountPath: /opt/eg-data/
+        - mountPath: /opt/easegress/data
           name: easegress-data-volume
       restartPolicy: Always
       volumes:
diff --git a/helm-charts/easegress/templates/PersistentVolume.yaml b/helm-charts/easegress/templates/PersistentVolume.yaml
index fb773f52..9a18d74b 100644
--- a/helm-charts/easegress/templates/PersistentVolume.yaml
+++ b/helm-charts/easegress/templates/PersistentVolume.yaml
@@ -12,7 +12,7 @@ spec:
   - ReadWriteOnce
   persistentVolumeReclaimPolicy: Delete
   storageClassName: easegress-storage
-  hostPath:
+  local:
     path: /opt/easegress
   nodeAffinity:
     required:
diff --git a/helm-charts/easegress/templates/StatefulSet.yaml b/helm-charts/easegress/templates/StatefulSet.yaml
index e2f43e50..124e1433 100644
--- a/helm-charts/easegress/templates/StatefulSet.yaml
+++ b/helm-charts/easegress/templates/StatefulSet.yaml
@@ -59,7 +59,9 @@ spec:
         - mountPath: /opt/eg-config/eg-primary.yaml
           name: {{ .Release.Name }}
           subPath: eg-primary.yaml
-        - mountPath: /opt/eg-data/
+        - mountPath: /opt/easegress/data
+          name: {{ if eq .Values.cluster.volumeType "persistentVolume" }}easegress-pv {{ else }} easegress-data-volume {{ end }}
+        - mountPath: /opt/easegress/log
           name: {{ if eq .Values.cluster.volumeType "persistentVolume" }}easegress-pv {{ else }} easegress-data-volume {{ end }}
       restartPolicy: Always
       volumes:
diff --git a/helm-charts/ingress-controller/README.md b/helm-charts/ingress-controller/README.md
index c1f1de6d..37b0e7d6 100644
--- a/helm-charts/ingress-controller/README.md
+++ b/helm-charts/ingress-controller/README.md
@@ -11,6 +11,15 @@ kubectl create ns ingress-easegress
 # update common helm charts
 helm dependency update ./helm-charts/ingress-controller
 ```
+
+### 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
@@ -35,3 +44,13 @@ helm install ingress-easegress -n ingress-easegress ./helm-charts/ingress-contro
   --set cluster.volumeType=persistentVolume \
   --set 'cluster.nodeHostnames={hostname-xyz}'
 ```
+
+## 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
+```
diff --git a/helm-charts/ingress-controller/templates/Deployment.yaml b/helm-charts/ingress-controller/templates/Deployment.yaml
index d9dfd4ea..eafae4d1 100644
--- a/helm-charts/ingress-controller/templates/Deployment.yaml
+++ b/helm-charts/ingress-controller/templates/Deployment.yaml
@@ -62,7 +62,7 @@ spec:
         - mountPath: /opt/eg-config/controller.yaml
           name: {{ .Release.Name }}
           subPath: controller.yaml
-        - mountPath: /opt/eg-data/
+        - mountPath: /opt/easegress/data
           name: ingress-data-volume
       restartPolicy: Always
       volumes:
diff --git a/helm-charts/ingress-controller/templates/PersistentVolume.yaml b/helm-charts/ingress-controller/templates/PersistentVolume.yaml
index 54610380..eb2516a7 100644
--- a/helm-charts/ingress-controller/templates/PersistentVolume.yaml
+++ b/helm-charts/ingress-controller/templates/PersistentVolume.yaml
@@ -12,7 +12,7 @@ spec:
   - ReadWriteOnce
   persistentVolumeReclaimPolicy: Delete
   storageClassName: easegress-storage
-  hostPath:
+  local:
     path: /opt/easegress
   nodeAffinity:
     required:
diff --git a/helm-charts/ingress-controller/templates/StatefulSet.yaml b/helm-charts/ingress-controller/templates/StatefulSet.yaml
index 0fa726a6..43ff482e 100644
--- a/helm-charts/ingress-controller/templates/StatefulSet.yaml
+++ b/helm-charts/ingress-controller/templates/StatefulSet.yaml
@@ -63,8 +63,10 @@ spec:
         - mountPath: /opt/eg-config/controller.yaml
           name: {{ .Release.Name }}
           subPath: controller.yaml
-        - mountPath: /opt/eg-data/
+        - 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: {}
diff --git a/pkg/cluster/cluster_interface.go b/pkg/cluster/cluster_interface.go
index f6d1cafc..ac9656d7 100644
--- a/pkg/cluster/cluster_interface.go
+++ b/pkg/cluster/cluster_interface.go
@@ -21,6 +21,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/megaease/easegress/pkg/logger"
+
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
@@ -37,6 +39,7 @@ type (
 		GetPrefix(prefix string) (map[string]string, error)
 		GetRaw(key string) (*mvccpb.KeyValue, error)
 		GetRawPrefix(prefix string) (map[string]*mvccpb.KeyValue, error)
+		GetWithOp(key string, ops ...ClientOp) (map[string]string, error)
 
 		Put(key, value string) error
 		PutUnderLease(key, value string) error
@@ -64,12 +67,46 @@ type (
 		PurgeMember(member string) error
 	}
 
+	// ClientOp is client operation option type for etcd client used in cluster and watcher
+	ClientOp string
+
 	// Watcher wraps etcd watcher.
 	Watcher interface {
 		Watch(key string) (<-chan *string, error)
 		WatchPrefix(prefix string) (<-chan map[string]*string, error)
 		WatchRaw(key string) (<-chan *clientv3.Event, error)
 		WatchRawPrefix(prefix string) (<-chan map[string]*clientv3.Event, error)
+		WatchWithOp(key string, ops ...ClientOp) (<-chan map[string]*string, error)
 		Close()
 	}
 )
+
+const (
+	// OpPrefix will watch all event with certain prefix
+	OpPrefix ClientOp = "prefix"
+
+	// OpNotWatchPut will not watch put event
+	OpNotWatchPut ClientOp = "put"
+
+	// OpNotWatchDelete will not watch delete event
+	OpNotWatchDelete ClientOp = "delete"
+
+	// OpKeysOnly will get etcd and only return keys, for example, get all prefix without values
+	OpKeysOnly ClientOp = "keysOnly"
+)
+
+func getOpOption(op ClientOp) clientv3.OpOption {
+	switch op {
+	case OpPrefix:
+		return clientv3.WithPrefix()
+	case OpNotWatchPut:
+		return clientv3.WithFilterPut()
+	case OpNotWatchDelete:
+		return clientv3.WithFilterDelete()
+	case OpKeysOnly:
+		return clientv3.WithKeysOnly()
+	default:
+		logger.Errorf("unsupported client operation: %v", op)
+		return nil
+	}
+}
diff --git a/pkg/cluster/op.go b/pkg/cluster/op.go
index ebf3eef4..aa7ad21c 100644
--- a/pkg/cluster/op.go
+++ b/pkg/cluster/op.go
@@ -175,6 +175,30 @@ func (c *cluster) GetRawPrefix(prefix string) (map[string]*mvccpb.KeyValue, erro
 	return kvs, nil
 }
 
+func (c *cluster) GetWithOp(key string, op ...ClientOp) (map[string]string, error) {
+	kvs := make(map[string]string)
+
+	client, err := c.getClient()
+	if err != nil {
+		return kvs, err
+	}
+
+	newOps := []clientv3.OpOption{}
+	for _, o := range op {
+		if opOption := getOpOption(o); opOption != nil {
+			newOps = append(newOps, opOption)
+		}
+	}
+	resp, err := client.Get(c.requestContext(), key, newOps...)
+	if err != nil {
+		return kvs, err
+	}
+	for _, kv := range resp.Kvs {
+		kvs[string(kv.Key)] = string(kv.Value)
+	}
+	return kvs, nil
+}
+
 func (c *cluster) STM(apply func(concurrency.STM) error) error {
 	client, err := c.getClient()
 	if err != nil {
diff --git a/pkg/cluster/watcher.go b/pkg/cluster/watcher.go
index bdd8d6e0..f5322b83 100644
--- a/pkg/cluster/watcher.go
+++ b/pkg/cluster/watcher.go
@@ -220,6 +220,58 @@ func (w *watcher) WatchRawPrefix(prefix string) (<-chan map[string]*clientv3.Eve
 	return prefixChan, nil
 }
 
+func (w *watcher) WatchWithOp(key string, ops ...ClientOp) (<-chan map[string]*string, error) {
+	newOps := []clientv3.OpOption{}
+	for _, o := range ops {
+		if opOption := getOpOption(o); opOption != nil {
+			newOps = append(newOps, opOption)
+		}
+	}
+
+	// NOTE: Can't use Context with timeout here.
+	ctx, cancel := context.WithCancel(context.Background())
+	watchResp := w.w.Watch(ctx, key, newOps...)
+	prefixChan := make(chan map[string]*string, 10)
+
+	go func() {
+		defer cancel()
+		defer close(prefixChan)
+
+		for {
+			select {
+			case <-w.done:
+				return
+			case resp := <-watchResp:
+				if resp.Canceled {
+					logger.Errorf("watch %s with ops %v canceled: %v", key, ops, resp.Err())
+					return
+				}
+				if resp.IsProgressNotify() {
+					continue
+				}
+				for _, event := range resp.Events {
+					switch event.Type {
+					case mvccpb.PUT:
+						value := string(event.Kv.Value)
+						prefixChan <- map[string]*string{
+							string(event.Kv.Key): &value,
+						}
+					case mvccpb.DELETE:
+						prefixChan <- map[string]*string{
+							string(event.Kv.Key): nil,
+						}
+					default:
+						logger.Errorf("BUG: key %s with ops %v received unknown event type %v",
+							key, ops, event.Type)
+					}
+				}
+			}
+		}
+	}()
+
+	return prefixChan, nil
+}
+
 func (w *watcher) Close() {
 	close(w.done)
 
diff --git a/pkg/context/context.go b/pkg/context/context.go
new file mode 100644
index 00000000..b484a896
--- /dev/null
+++ b/pkg/context/context.go
@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package context
+
+import (
+	stdcontext "context"
+)
+
+type (
+	// Protocol is type of protocol that context support
+	Protocol string
+
+	// Context is general context for HTTPContext, MQTTContext, TCPContext
+	Context interface {
+		stdcontext.Context
+		Protocol() Protocol
+	}
+)
+
+const (
+	// HTTP is HTTP protocol
+	HTTP Protocol = "HTTP"
+
+	// MQTT is MQTT protocol
+	MQTT Protocol = "MQTT"
+
+	// TCP is TCP protocol
+	TCP Protocol = "TCP"
+
+	// UDP is UDP protocol
+	UDP Protocol = "UDP"
+)
diff --git a/pkg/context/httpcontext.go b/pkg/context/httpcontext.go
index 0fb1e489..35ad9b01 100644
--- a/pkg/context/httpcontext.go
+++ b/pkg/context/httpcontext.go
@@ -44,6 +44,7 @@ type (
 	// It is not goroutine-safe, callers must use Lock/Unlock
 	// to protect it by themselves.
 	HTTPContext interface {
+		Context
 		Lock()
 		Unlock()
 
@@ -52,7 +53,6 @@ type (
 		Request() HTTPRequest
 		Response() HTTPResponse
 
-		stdcontext.Context
 		Cancel(err error)
 		Cancelled() bool
 		ClientDisconnected() bool
@@ -123,6 +123,11 @@ type (
 		Size() uint64 // bytes
 	}
 
+	// HTTPResult is result for handling http request
+	HTTPResult struct {
+		Err error
+	}
+
 	// FinishFunc is the type of function to be called back
 	// when HTTPContext is finishing.
 	FinishFunc = func()
@@ -177,6 +182,10 @@ func New(stdw http.ResponseWriter, stdr *http.Request,
 	}
 }
 
+func (ctx *httpContext) Protocol() Protocol {
+	return HTTP
+}
+
 func (ctx *httpContext) CallNextHandler(lastResult string) string {
 	return ctx.caller(lastResult)
 }
diff --git a/pkg/context/statuscode.go b/pkg/context/httpstatuscode.go
similarity index 100%
rename from pkg/context/statuscode.go
rename to pkg/context/httpstatuscode.go
diff --git a/pkg/context/mqttcontext.go b/pkg/context/mqttcontext.go
new file mode 100644
index 00000000..e38186d1
--- /dev/null
+++ b/pkg/context/mqttcontext.go
@@ -0,0 +1,255 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package context
+
+import (
+	stdcontext "context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
+)
+
+type (
+	// MQTTContext is context for MQTT protocol
+	MQTTContext interface {
+		Context
+		Client() MQTTClient
+		Backend() MQTTBackend
+		Cancel(error)
+		Canceled() bool
+		Duration() time.Duration
+		Finish()
+
+		PacketType() MQTTPacketType
+		ConnectPacket() *packets.ConnectPacket       // read only
+		PublishPacket() *packets.PublishPacket       // read only
+		DisconnectPacket() *packets.DisconnectPacket // read only
+
+		SetDrop()         // set drop value to true
+		Drop() bool       // if true, this mqtt packet will be dropped
+		SetDisconnect()   // set disconnect value to true
+		Disconnect() bool // if true, this mqtt client will be disconnected
+		SetEarlyStop()    // set early stop value to true
+		EarlyStop() bool  // if early stop is true, pipeline will skip following filters and return
+	}
+
+	// MQTTBackend is backend of MQTT proxy
+	MQTTBackend interface {
+		Publish(target string, data []byte, headers map[string]string) error
+	}
+
+	// MQTTClient contains client info that send this packet
+	MQTTClient interface {
+		ClientID() string
+		UserName() string
+		Load(key interface{}) (value interface{}, ok bool)
+		Store(key interface{}, value interface{})
+		Delete(key interface{})
+	}
+
+	// MQTTPacketType contains supported mqtt packet type
+	MQTTPacketType int
+
+	mqttContext struct {
+		mu         sync.RWMutex
+		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
+
+		err        error
+		drop       int32
+		disconnect int32
+		earlyStop  int32
+	}
+
+	// MQTTResult is result for handling mqtt request
+	MQTTResult struct {
+		Err error
+	}
+)
+
+const (
+	// MQTTConnect is mqtt packet type of connect
+	MQTTConnect MQTTPacketType = 1
+
+	// MQTTPublish is mqtt packet type of publish
+	MQTTPublish MQTTPacketType = 2
+
+	// MQTTDisconnect is mqtt packet type of disconnect
+	MQTTDisconnect = 3
+
+	// MQTTOther is all other mqtt packet type
+	MQTTOther MQTTPacketType = 99
+)
+
+var _ MQTTContext = (*mqttContext)(nil)
+
+// NewMQTTContext create new MQTTContext
+func NewMQTTContext(ctx stdcontext.Context, backend MQTTBackend, client MQTTClient, packet packets.ControlPacket) MQTTContext {
+	stdctx, cancelFunc := stdcontext.WithCancel(ctx)
+	startTime := time.Now()
+	mqttCtx := &mqttContext{
+		ctx:        stdctx,
+		cancelFunc: cancelFunc,
+		startTime:  startTime,
+		client:     client,
+		backend:    backend,
+	}
+
+	switch packet := 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
+	default:
+		mqttCtx.packetType = MQTTOther
+	}
+	return mqttCtx
+}
+
+// Protocol return protocol of mqttContext
+func (ctx *mqttContext) Protocol() Protocol {
+	return MQTT
+}
+
+// Deadline return deadline of mqttContext
+func (ctx *mqttContext) Deadline() (time.Time, bool) {
+	return ctx.ctx.Deadline()
+}
+
+// Done return done chan of mqttContext
+func (ctx *mqttContext) Done() <-chan struct{} {
+	return ctx.ctx.Done()
+}
+
+// Err return error of mqttContext
+func (ctx *mqttContext) Err() error {
+	ctx.mu.RLock()
+	defer ctx.mu.RUnlock()
+	if ctx.err != nil {
+		return ctx.err
+	}
+	return ctx.ctx.Err()
+}
+
+// Value return value of mqttContext for given key
+func (ctx *mqttContext) Value(key interface{}) interface{} {
+	return ctx.ctx.Value(key)
+}
+
+// Client return mqttContext client
+func (ctx *mqttContext) Client() MQTTClient {
+	return ctx.client
+}
+
+// Cancel cancel mqttContext
+func (ctx *mqttContext) Cancel(err error) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+	if !ctx.canceled() {
+		ctx.err = err
+		ctx.cancelFunc()
+	}
+}
+
+func (ctx *mqttContext) canceled() bool {
+	return ctx.err != nil || ctx.ctx.Err() != nil
+}
+
+// Canceled return if mqttContext is canceled
+func (ctx *mqttContext) Canceled() bool {
+	ctx.mu.RLock()
+	defer ctx.mu.RUnlock()
+	return ctx.canceled()
+}
+
+// Duration return time duration since this context start
+func (ctx *mqttContext) Duration() time.Duration {
+	ctx.mu.RLock()
+	defer ctx.mu.RUnlock()
+	if !ctx.endTime.IsZero() {
+		return ctx.endTime.Sub(ctx.startTime)
+	}
+	return time.Since(ctx.startTime)
+}
+
+// Finish tell this context is finished
+func (ctx *mqttContext) Finish() {
+	ctx.mu.Lock()
+	ctx.endTime = time.Now()
+	ctx.mu.Unlock()
+}
+
+func (ctx *mqttContext) PacketType() MQTTPacketType {
+	return ctx.packetType
+}
+
+func (ctx *mqttContext) ConnectPacket() *packets.ConnectPacket {
+	return ctx.connectPacket
+}
+
+func (ctx *mqttContext) PublishPacket() *packets.PublishPacket {
+	return ctx.publishPacket
+}
+
+func (ctx *mqttContext) DisconnectPacket() *packets.DisconnectPacket {
+	return ctx.disconnectPacket
+}
+
+func (ctx *mqttContext) SetDrop() {
+	atomic.StoreInt32(&ctx.drop, 1)
+}
+
+func (ctx *mqttContext) Drop() bool {
+	return atomic.LoadInt32(&ctx.drop) == 1
+}
+
+func (ctx *mqttContext) SetDisconnect() {
+	atomic.StoreInt32(&ctx.disconnect, 1)
+}
+
+func (ctx *mqttContext) Disconnect() bool {
+	return atomic.LoadInt32(&ctx.disconnect) == 1
+}
+
+func (ctx *mqttContext) SetEarlyStop() {
+	atomic.StoreInt32(&ctx.earlyStop, 1)
+}
+
+func (ctx *mqttContext) EarlyStop() bool {
+	return atomic.LoadInt32(&ctx.earlyStop) == 1
+}
+
+func (ctx *mqttContext) Backend() MQTTBackend {
+	return ctx.backend
+}
diff --git a/pkg/context/mqttmock.go b/pkg/context/mqttmock.go
new file mode 100644
index 00000000..ecc40783
--- /dev/null
+++ b/pkg/context/mqttmock.go
@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package context
+
+import "sync"
+
+// MockMQTTClient is mock client for MQTTContext
+type MockMQTTClient struct {
+	MockClientID string
+	MockUserName string
+	MockKVMap    sync.Map
+}
+
+var _ MQTTClient = (*MockMQTTClient)(nil)
+
+// ClientID return client id of MockMQTTClient
+func (m *MockMQTTClient) ClientID() string {
+	return m.MockClientID
+}
+
+// UserName return username if MockMQTTClient
+func (m *MockMQTTClient) UserName() string {
+	return m.MockUserName
+}
+
+// Load load value keep in MockMQTTClient kv map
+func (m *MockMQTTClient) Load(key interface{}) (value interface{}, ok bool) {
+	return m.MockKVMap.Load(key)
+}
+
+// Store store kv pair into MockMQTTClient kv map
+func (m *MockMQTTClient) Store(key interface{}, value interface{}) {
+	m.MockKVMap.Store(key, value)
+}
+
+// Delete delete key-value pair in MockMQTTClient kv map
+func (m *MockMQTTClient) Delete(key interface{}) {
+	m.MockKVMap.Delete(key)
+}
+
+// MockMQTTMsg is message send to MockMQTTBackend
+type MockMQTTMsg struct {
+	Target  string
+	Data    []byte
+	Headers map[string]string
+}
+
+// MockMQTTBackend is mocked MQTT backend
+type MockMQTTBackend struct {
+	Messages map[string]MockMQTTMsg
+}
+
+var _ MQTTBackend = (*MockMQTTBackend)(nil)
+
+// Publish publish msg to MockMQTTBackend
+func (m *MockMQTTBackend) Publish(target string, data []byte, headers map[string]string) error {
+	m.Messages[target] = MockMQTTMsg{
+		Target:  target,
+		Data:    data,
+		Headers: headers,
+	}
+	return nil
+}
diff --git a/pkg/context/tcpcontext.go b/pkg/context/tcpcontext.go
new file mode 100644
index 00000000..bf674a91
--- /dev/null
+++ b/pkg/context/tcpcontext.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package context
+
+type (
+	// TCPContext is context for TCP protocol
+	TCPContext interface {
+		Context
+
+		// TODO more protocols coming soon
+	}
+
+	// TCPResult is result for handling TCP request
+	TCPResult struct {
+		Err error
+	}
+)
diff --git a/pkg/context/udpcontext.go b/pkg/context/udpcontext.go
new file mode 100644
index 00000000..980a6c46
--- /dev/null
+++ b/pkg/context/udpcontext.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package context
+
+type (
+	// UDPContext is context for UDP protocol
+	UDPContext interface {
+		Context
+
+		// TODO more protocols coming soon
+	}
+
+	// UDPResult is result for handling UDP request
+	UDPResult struct {
+		Err error
+	}
+)
diff --git a/pkg/filter/connectcontrol/connectcontrol.go b/pkg/filter/connectcontrol/connectcontrol.go
new file mode 100644
index 00000000..8afc98eb
--- /dev/null
+++ b/pkg/filter/connectcontrol/connectcontrol.go
@@ -0,0 +1,189 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package connectcontrol
+
+import (
+	"errors"
+	"regexp"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/pipeline"
+)
+
+const (
+	// Kind is the kind of ConnectControl
+	Kind = "ConnectControl"
+
+	resultBannedClientOrTopic = "bannedClientOrTopicError"
+)
+
+// ErrBannedClientOrTopic is error for banned client or topic
+var ErrBannedClientOrTopic = errors.New(resultBannedClientOrTopic)
+
+func init() {
+	pipeline.Register(&ConnectControl{})
+}
+
+type (
+	// ConnectControl is used to control MQTT clients connect status,
+	// if MQTTContext ClientID in bannedClients, the connection will be closed,
+	// if MQTTContext publish topic in bannedTopics, the connection will be closed.
+	ConnectControl struct {
+		filterSpec     *pipeline.FilterSpec
+		spec           *Spec
+		bannedClients  map[string]struct{}
+		bannedTopics   map[string]struct{}
+		bannedClientRe *regexp.Regexp
+		bannedTopicRe  *regexp.Regexp
+		status         *Status
+	}
+
+	// Spec describes the ConnectControl
+	Spec struct {
+		BannedClientRe string   `yaml:"bannedClientRe" jsonschema:"omitempty"`
+		BannedClients  []string `yaml:"bannedClients" jsonschema:"omitempty"`
+		BannedTopicRe  string   `yaml:"bannedTopicRe" jsonschema:"omitempty"`
+		BannedTopics   []string `yaml:"bannedTopics" jsonschema:"omitempty"`
+		EarlyStop      bool     `yaml:"earlyStop" jsonschema:"omitempty"`
+	}
+
+	// Status is ConnectControl filter status
+	Status struct {
+		BannedClientRe  string `yaml:"bannedClientRe" jsonschema:"omitempty"`
+		BannedClientNum int    `yaml:"bannedClientNum" jsonschema:"omitempty"`
+		BannedTopicRe   string `yaml:"bannedTopicRe" jsonschema:"omitempty"`
+		BannedTopicNum  int    `yaml:"bannedTopicNum" jsonschema:"omitempty"`
+	}
+)
+
+var _ pipeline.Filter = (*ConnectControl)(nil)
+var _ pipeline.MQTTFilter = (*ConnectControl)(nil)
+
+// Kind return kind of ConnectControl
+func (cc *ConnectControl) Kind() string {
+	return Kind
+}
+
+// DefaultSpec return default spec of ConnectControl
+func (cc *ConnectControl) DefaultSpec() interface{} {
+	return &Spec{}
+}
+
+// Description return description of ConnectControl
+func (cc *ConnectControl) Description() string {
+	return "ConnectControl control connections of MQTT clients"
+}
+
+// Results return results of ConnectControl
+func (cc *ConnectControl) Results() []string {
+	return []string{resultBannedClientOrTopic}
+}
+
+// Init init ConnectControl with pipeline filter spec
+func (cc *ConnectControl) Init(filterSpec *pipeline.FilterSpec) {
+	if filterSpec.Protocol() != context.MQTT {
+		panic("filter ConnectControl only support MQTT protocol for now")
+	}
+	cc.filterSpec, cc.spec = filterSpec, filterSpec.FilterSpec().(*Spec)
+	cc.bannedClients = make(map[string]struct{})
+	cc.bannedTopics = make(map[string]struct{})
+	cc.reload()
+}
+
+func (cc *ConnectControl) reload() {
+	if len(cc.spec.BannedClientRe) > 0 {
+		r, err := regexp.Compile(cc.spec.BannedClientRe)
+		if err != nil {
+			logger.Errorf("filter ConnectControl compile BannedClientRe %s failed, %s", cc.spec.BannedClientRe, err)
+		} else {
+			cc.bannedClientRe = r
+		}
+	}
+	if len(cc.spec.BannedTopicRe) > 0 {
+		r, err := regexp.Compile(cc.spec.BannedTopicRe)
+		if err != nil {
+			logger.Errorf("filter ConnectControl compile BannedTopicRe %s failed, %s", cc.spec.BannedTopicRe, err)
+		} else {
+			cc.bannedTopicRe = r
+		}
+	}
+
+	for _, c := range cc.spec.BannedClients {
+		cc.bannedClients[c] = struct{}{}
+	}
+	for _, t := range cc.spec.BannedTopics {
+		cc.bannedTopics[t] = struct{}{}
+	}
+
+	cc.status = &Status{
+		BannedClientRe:  cc.spec.BannedClientRe,
+		BannedTopicRe:   cc.spec.BannedTopicRe,
+		BannedClientNum: len(cc.spec.BannedClients),
+		BannedTopicNum:  len(cc.spec.BannedTopics),
+	}
+}
+
+// Inherit init ConnectControl with previous generation
+func (cc *ConnectControl) Inherit(filterSpec *pipeline.FilterSpec, previousGeneration pipeline.Filter) {
+	previousGeneration.Close()
+	cc.Init(filterSpec)
+}
+
+// Status return status of ConnectControl
+func (cc *ConnectControl) Status() interface{} {
+	return cc.status
+}
+
+// Close close ConnectControl gracefully
+func (cc *ConnectControl) Close() {
+}
+
+func (cc *ConnectControl) checkBan(ctx context.MQTTContext) bool {
+	cid := ctx.Client().ClientID()
+	if cc.bannedClientRe != nil && cc.bannedClientRe.MatchString(cid) {
+		return true
+	}
+	if _, ok := cc.bannedClients[cid]; ok {
+		return true
+	}
+	topic := ctx.PublishPacket().TopicName
+	if cc.bannedTopicRe != nil && cc.bannedTopicRe.MatchString(topic) {
+		return true
+	}
+	if _, ok := cc.bannedTopics[topic]; ok {
+		return true
+	}
+	return false
+}
+
+// HandleMQTT handle MQTT request
+func (cc *ConnectControl) HandleMQTT(ctx context.MQTTContext) *context.MQTTResult {
+	if ctx.PacketType() != context.MQTTPublish {
+		return &context.MQTTResult{}
+	}
+
+	if cc.checkBan(ctx) {
+		ctx.SetDisconnect()
+		if cc.spec.EarlyStop {
+			ctx.SetEarlyStop()
+		}
+		return &context.MQTTResult{Err: ErrBannedClientOrTopic}
+	}
+	return &context.MQTTResult{}
+}
diff --git a/pkg/logger/export.go b/pkg/logger/export.go
index 988a5505..4cf33445 100644
--- a/pkg/logger/export.go
+++ b/pkg/logger/export.go
@@ -22,6 +22,8 @@ import (
 	"time"
 
 	"github.com/megaease/easegress/pkg/util/fasttime"
+	"github.com/megaease/easegress/pkg/util/stringtool"
+	"github.com/openzipkin/zipkin-go/model"
 )
 
 type lazyLogBuilder struct {
@@ -141,3 +143,22 @@ func NginxHTTPAccess(remoteAddr, proto, method, path, referer, agent, realIP str
 		requestTime.Seconds(), upstreamResponseTime.Seconds(), upstreamAddr, upstreamCode,
 		clientWriteBodyTime.Seconds(), clientReadBodyTime.Seconds(), routeTime.Seconds())
 }
+
+func getSpanTemplate(context *model.SpanContext, template string) string {
+	if context == nil {
+		return stringtool.Cat("tid=<nil> sid=<nil> ", template)
+	}
+	return stringtool.Cat("tid=", context.TraceID.String(), " sid=", context.ID.String(), " ", template)
+}
+
+// SpanDebugf is the wrapper of default logger Debugf to log tracing message
+func SpanDebugf(context *model.SpanContext, template string, args ...interface{}) {
+	temp := getSpanTemplate(context, template)
+	defaultLogger.Debugf(temp, args...)
+}
+
+// SpanErrorf is the wrapper of default logger Debugf to log tracing message
+func SpanErrorf(context *model.SpanContext, template string, args ...interface{}) {
+	temp := getSpanTemplate(context, template)
+	defaultLogger.Errorf(temp, args...)
+}
diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
index 6e84fc29..02e508c6 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -49,6 +49,18 @@ func InitNop() {
 	stderrLogger = defaultLogger
 }
 
+// InitMock initializes all logger to print stdout, mainly for unit testing
+func InitMock() {
+	mock := zap.NewExample()
+	httpFilterAccessLogger = mock.Sugar()
+	httpFilterDumpLogger = mock.Sugar()
+	restAPILogger = mock.Sugar()
+
+	defaultLogger = mock.Sugar()
+	gressLogger = defaultLogger
+	stderrLogger = defaultLogger
+}
+
 const (
 	stdoutFilename           = "stdout.log"
 	filterHTTPAccessFilename = "filter_http_access.log"
diff --git a/pkg/object/mqttproxy/backend.go b/pkg/object/mqttproxy/backend.go
index 47c0570b..1b883c0d 100644
--- a/pkg/object/mqttproxy/backend.go
+++ b/pkg/object/mqttproxy/backend.go
@@ -19,9 +19,11 @@ package mqttproxy
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/Shopify/sarama"
 	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
 )
 
@@ -29,6 +31,7 @@ type (
 	// BackendMQ is backend message queue for MQTT proxy
 	backendMQ interface {
 		publish(p *packets.PublishPacket) error
+		Publish(target string, data []byte, headers map[string]string) error
 		close()
 	}
 
@@ -40,10 +43,17 @@ type (
 	}
 
 	testMQ struct {
-		ch chan *packets.PublishPacket
+		mu  sync.Mutex
+		ch  chan *packets.PublishPacket
+		msg map[string]map[string]string
 	}
 )
 
+var _ backendMQ = (*KafkaMQ)(nil)
+var _ backendMQ = (*testMQ)(nil)
+var _ context.MQTTBackend = (*KafkaMQ)(nil)
+var _ context.MQTTBackend = (*testMQ)(nil)
+
 const (
 	kafkaType  = "Kafka"
 	testMQType = "TestMQ"
@@ -56,14 +66,15 @@ func newBackendMQ(spec *Spec) backendMQ {
 	case testMQType:
 		t := &testMQ{}
 		t.ch = make(chan *packets.PublishPacket, 100)
+		t.msg = make(map[string]map[string]string)
 		return t
 	default:
-		logger.Errorf("backend type <%s> not support", spec.BackendType)
+		logger.SpanErrorf(nil, "backend type <%s> not support", spec.BackendType)
 		return nil
 	}
 }
 
-func newKafkaMQ(spec *Spec) *KafkaMQ {
+func newKafkaMQ(spec *Spec) backendMQ {
 	k := &KafkaMQ{}
 	k.mapFunc = getTopicMapFunc(spec.TopicMapper)
 	k.done = make(chan struct{})
@@ -73,7 +84,7 @@ func newKafkaMQ(spec *Spec) *KafkaMQ {
 	config.Version = sarama.V1_0_0_0
 	producer, err := sarama.NewAsyncProducer(spec.Kafka.Backend, config)
 	if err != nil {
-		logger.Errorf("start sarama producer with address %v failed: %v", spec.Kafka.Backend, err)
+		logger.SpanErrorf(nil, "start sarama producer with address %v failed: %v", spec.Kafka.Backend, err)
 		return nil
 	}
 
@@ -86,7 +97,7 @@ func newKafkaMQ(spec *Spec) *KafkaMQ {
 				if !ok {
 					return
 				}
-				logger.Errorf("sarama producer failed: %v", err)
+				logger.SpanErrorf(nil, "sarama producer failed: %v", err)
 			}
 		}
 	}()
@@ -97,7 +108,7 @@ func newKafkaMQ(spec *Spec) *KafkaMQ {
 
 func (k *KafkaMQ) publish(p *packets.PublishPacket) error {
 	var msg *sarama.ProducerMessage
-	logger.Debugf("produce msg with topic %s", p.TopicName)
+	logger.SpanDebugf(nil, "produce msg with topic %s", p.TopicName)
 
 	if k.mapFunc != nil {
 		topic, headers, err := k.mapFunc(p.TopicName)
@@ -128,14 +139,38 @@ func (k *KafkaMQ) close() {
 	close(k.done)
 	err := k.producer.Close()
 	if err != nil {
-		logger.Errorf("close kafka producer failed: %v", err)
+		logger.SpanErrorf(nil, "close kafka producer failed: %v", err)
 	}
 }
 
+// Publish publish msg to Kafka backend
+func (k *KafkaMQ) Publish(target string, data []byte, headers map[string]string) error {
+	var msg *sarama.ProducerMessage
+	kafkaHeaders := make([]sarama.RecordHeader, 0, len(headers))
+	for k, v := range headers {
+		kafkaHeaders = append(kafkaHeaders, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)})
+	}
+	msg = &sarama.ProducerMessage{
+		Topic:   target,
+		Headers: kafkaHeaders,
+		Value:   sarama.ByteEncoder(data),
+	}
+	k.producer.Input() <- msg
+	return nil
+}
+
 func (t *testMQ) publish(p *packets.PublishPacket) error {
 	t.ch <- p
 	return nil
 }
 
+// Publish publish msg to testMQ backend
+func (t *testMQ) Publish(target string, data []byte, headers map[string]string) error {
+	t.mu.Lock()
+	t.msg[target] = headers
+	t.mu.Unlock()
+	return nil
+}
+
 func (t *testMQ) close() {
 }
diff --git a/pkg/object/mqttproxy/broker.go b/pkg/object/mqttproxy/broker.go
index e08224d2..9a2ff9aa 100644
--- a/pkg/object/mqttproxy/broker.go
+++ b/pkg/object/mqttproxy/broker.go
@@ -19,6 +19,7 @@ package mqttproxy
 
 import (
 	"bytes"
+	stdcontext "context"
 	"crypto/sha256"
 	"crypto/tls"
 	"encoding/base64"
@@ -27,11 +28,19 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"strconv"
+	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/megaease/easegress/pkg/api"
+	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/pipeline"
+	"github.com/openzipkin/zipkin-go/model"
+	"github.com/openzipkin/zipkin-go/propagation/b3"
 )
 
 type (
@@ -47,13 +56,16 @@ type (
 		clients    map[string]*Client
 		sha256Auth map[string]string
 		tlsCfg     *tls.Config
+		pipeline   string
 
-		sessMgr   *SessionManager
-		topicMgr  *TopicManager
-		memberURL func(string, string) ([]string, error)
+		sessMgr           *SessionManager
+		topicMgr          *TopicManager
+		connectionLimiter *Limiter
+		memberURL         func(string, string) ([]string, error)
 
 		// done is the channel for shutdowning this proxy.
-		done chan struct{}
+		done      chan struct{}
+		closeFlag int32
 	}
 
 	// HTTPJsonData is json data received from http endpoint used to send back to clients
@@ -64,6 +76,17 @@ type (
 		Base64      bool   `json:"base64"`
 		Distributed bool   `json:"distributed"`
 	}
+
+	// HTTPSessions is json data used for session related operations, like get all sessions and delete some sessions
+	HTTPSessions struct {
+		Sessions []*HTTPSession `json:"sessions"`
+	}
+
+	// HTTPSession is json data used for session related operations, like get all sessions and delete some sessions
+	HTTPSession struct {
+		SessionID string `json:"sessionID"`
+		Topic     string `json:"topic"`
+	}
 )
 
 func sha256Sum(data []byte) string {
@@ -76,29 +99,35 @@ func newBroker(spec *Spec, store storage, memberURL func(string, string) ([]stri
 		egName:     spec.EGName,
 		name:       spec.Name,
 		spec:       spec,
+		pipeline:   spec.Pipeline,
 		backend:    newBackendMQ(spec),
 		clients:    make(map[string]*Client),
 		sha256Auth: make(map[string]string),
 		memberURL:  memberURL,
 		done:       make(chan struct{}),
 	}
+	if broker.backend == nil {
+		panic(fmt.Sprintf("mqtt broker %v connect backend failed", broker.name))
+	}
 
-	for _, a := range spec.Auth {
-		passwd, err := base64.StdEncoding.DecodeString(a.PassBase64)
-		if err != nil {
-			logger.Errorf("auth with name %v, base64 password %v decode failed: %v", a.UserName, a.PassBase64, err)
+	if !spec.AuthByPipeline {
+		for _, a := range spec.Auth {
+			passwd, err := base64.StdEncoding.DecodeString(a.PassBase64)
+			if err != nil {
+				logger.SpanErrorf(nil, "auth with name %v, base64 password %v decode failed: %v", a.UserName, a.PassBase64, err)
+				return nil
+			}
+			broker.sha256Auth[a.UserName] = sha256Sum(passwd)
+		}
+		if len(broker.sha256Auth) == 0 {
+			logger.SpanErrorf(nil, "empty valid auth for mqtt proxy")
 			return nil
 		}
-		broker.sha256Auth[a.UserName] = sha256Sum(passwd)
-	}
-	if len(broker.sha256Auth) == 0 {
-		logger.Errorf("empty valid auth for mqtt proxy")
-		return nil
 	}
 
 	err := broker.setListener()
 	if err != nil {
-		logger.Errorf("mqtt broker set listener failed: %v", err)
+		logger.SpanErrorf(nil, "mqtt broker set listener failed: %v", err)
 		return nil
 	}
 
@@ -107,7 +136,15 @@ func newBroker(spec *Spec, store storage, memberURL func(string, string) ([]stri
 	}
 	broker.topicMgr = newTopicManager(spec.TopicCacheSize)
 	broker.sessMgr = newSessionManager(broker, store)
+	broker.connectionLimiter = newLimiter(spec.ConnectionLimit)
 	go broker.run()
+	ch, closeFunc, err := broker.sessMgr.store.watchDelete(sessionStoreKey(""))
+	if err != nil {
+		logger.SpanErrorf(nil, "get watcher for session failed, %v", err)
+	}
+	if ch != nil {
+		go broker.watchDelete(ch, closeFunc)
+	}
 	return broker
 }
 
@@ -136,6 +173,75 @@ func (b *Broker) setListener() error {
 	return err
 }
 
+func (b *Broker) reconnectWatcher() {
+	if b.closed() {
+		return
+	}
+
+	ch, cancelFunc, err := b.sessMgr.store.watchDelete(sessionStoreKey(""))
+	if err != nil {
+		logger.SpanErrorf(nil, "get watcher for session failed, %v", err)
+		time.Sleep(10 * time.Second)
+		go b.reconnectWatcher()
+		return
+	}
+	go b.watchDelete(ch, cancelFunc)
+
+	// check event during reconnect
+	sessions, err := b.sessMgr.store.getPrefix(sessionStoreKey(""), true)
+	if err != nil {
+		logger.SpanErrorf(nil, "get all session prefix failed, %v", err)
+	}
+
+	clients := []*Client{}
+	b.Lock()
+	for sessionID, client := range b.clients {
+		if _, ok := sessions[sessionID]; !ok {
+			clients = append(clients, client)
+		}
+	}
+	b.Unlock()
+
+	for _, c := range clients {
+		c.close()
+	}
+}
+
+func (b *Broker) watchDelete(ch <-chan map[string]*string, closeFunc func()) {
+	defer closeFunc()
+	for {
+		select {
+		case <-b.done:
+			return
+		case m := <-ch:
+			if m == nil {
+				go b.reconnectWatcher()
+				return
+			}
+			for k, v := range m {
+				if v != nil {
+					continue
+				}
+				clientID := strings.TrimPrefix(k, sessionStoreKey(""))
+				logger.SpanDebugf(nil, "client %v recv delete watch %v", clientID, v)
+				go b.deleteSession(clientID)
+			}
+		}
+	}
+}
+
+func (b *Broker) deleteSession(clientID string) {
+	b.Lock()
+	defer b.Unlock()
+	if c, ok := b.clients[clientID]; ok {
+		if !c.disconnected() {
+			logger.SpanDebugf(nil, "broker watch and delete client %v", c.info.cid)
+			c.close()
+		}
+	}
+	delete(b.clients, clientID)
+}
+
 func (b *Broker) run() {
 	for {
 		conn, err := b.listener.Accept()
@@ -161,59 +267,129 @@ func (b *Broker) checkClientAuth(connect *packets.ConnectPacket) bool {
 	return false
 }
 
-func (b *Broker) handleConn(conn net.Conn) {
-	defer conn.Close()
-	packet, err := packets.ReadPacket(conn)
-	if err != nil {
-		logger.Errorf("read connect packet failed: %s", err)
-		return
-	}
-	connect, ok := packet.(*packets.ConnectPacket)
-	if !ok {
-		logger.Errorf("first packet received %s that was not Connect", packet.String())
-		return
+func (b *Broker) checkConnectPermission(connect *packets.ConnectPacket) bool {
+	size := connect.RemainingLength + 8
+	permitted := b.connectionLimiter.acquirePermission(size)
+	if !permitted {
+		return permitted
+	}
+	// check here to do early stop for connection. Later we will check it again to make sure
+	// not exceed MaxAllowedConnection
+	if b.spec.MaxAllowedConnection > 0 {
+		b.Lock()
+		connNum := len(b.clients)
+		b.Unlock()
+		if connNum >= b.spec.MaxAllowedConnection {
+			return false
+		}
 	}
-	logger.Debugf("connection from client %s", connect.ClientIdentifier)
+	return true
+}
 
+func (b *Broker) connectionValidation(connect *packets.ConnectPacket, conn net.Conn) (*Client, *packets.ConnackPacket, bool) {
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
 	connack.SessionPresent = connect.CleanSession
 	connack.ReturnCode = connect.Validate()
 	if connack.ReturnCode != packets.Accepted {
-		err = connack.Write(conn)
-		logger.Errorf("invalid connection %v, write connack failed: %s", connack.ReturnCode, err)
-		return
+		err := connack.Write(conn)
+		logger.SpanErrorf(nil, "invalid connection %v, write connack failed: %s", connack.ReturnCode, err)
+		return nil, nil, false
+	}
+	// check rate limiter and max allowed connection
+	if !b.checkConnectPermission(connect) {
+		logger.SpanDebugf(nil, "client %v not get connect permission from rate limiter", connect.ClientIdentifier)
+		connack.ReturnCode = packets.ErrRefusedServerUnavailable
+		err := connack.Write(conn)
+		if err != nil {
+			logger.SpanErrorf(nil, "connack back to client %s failed: %s", connect.ClientIdentifier, err)
+		}
+		return nil, nil, false
 	}
 
-	if !b.checkClientAuth(connect) {
+	client := newClient(connect, b, conn, b.spec.ClientPublishLimit)
+	// check auth
+	authFail := false
+	if b.spec.AuthByPipeline {
+		pipe, err := pipeline.GetPipeline(b.pipeline, context.MQTT)
+		if err != nil {
+			logger.SpanErrorf(nil, "get pipeline %v failed, %v", b.pipeline, err)
+			authFail = true
+		} else {
+			ctx := context.NewMQTTContext(stdcontext.Background(), b.backend, client, connect)
+			pipe.HandleMQTT(ctx)
+			if ctx.Disconnect() {
+				logger.SpanErrorf(nil, "client %v not get connect permission from pipeline", connect.ClientIdentifier)
+				authFail = true
+			}
+		}
+	} else if !b.checkClientAuth(connect) {
+		authFail = true
+	}
+	if authFail {
 		connack.ReturnCode = packets.ErrRefusedNotAuthorised
-		err = connack.Write(conn)
-		logger.Errorf("invalid connection %v, connack back failed: %s", connack.ReturnCode, err)
-		return
+		err := connack.Write(conn)
+		if err != nil {
+			logger.SpanErrorf(nil, "connack back to client %s failed: %s", connect.ClientIdentifier, err)
+		}
+		logger.SpanErrorf(nil, "invalid connection %v, client %s auth failed", connack.ReturnCode, connect.ClientIdentifier)
+		return nil, nil, false
 	}
+	return client, connack, true
+}
 
-	err = connack.Write(conn)
+func (b *Broker) handleConn(conn net.Conn) {
+	defer conn.Close()
+	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		logger.Errorf("send connack to client %s failed: %s", connect.ClientIdentifier, err)
+		logger.SpanErrorf(nil, "read connect packet failed: %s", err)
+		return
+	}
+	connect, ok := packet.(*packets.ConnectPacket)
+	if !ok {
+		logger.SpanErrorf(nil, "first packet received %s that was not Connect", packet.String())
 		return
 	}
+	logger.SpanDebugf(nil, "connection from client %s", connect.ClientIdentifier)
 
-	client := newClient(connect, b, conn)
+	client, connack, valid := b.connectionValidation(connect, conn)
+	if !valid {
+		return
+	}
 	cid := client.info.cid
 
 	b.Lock()
 	if oldClient, ok := b.clients[cid]; ok {
+		logger.SpanDebugf(nil, "client %v take over by new client with same name", oldClient.info.cid)
 		go oldClient.close()
+
+	} else if b.spec.MaxAllowedConnection > 0 {
+		if len(b.clients) >= b.spec.MaxAllowedConnection {
+			logger.SpanDebugf(nil, "client %v not get connect permission from rate limiter", connect.ClientIdentifier)
+			connack.ReturnCode = packets.ErrRefusedServerUnavailable
+			err = connack.Write(conn)
+			if err != nil {
+				logger.SpanErrorf(nil, "connack back to client %s failed: %s", connect.ClientIdentifier, err)
+			}
+			b.Unlock()
+			return
+		}
 	}
 	b.clients[client.info.cid] = client
 	b.setSession(client, connect)
 	b.Unlock()
 
+	err = connack.Write(conn)
+	if err != nil {
+		logger.SpanErrorf(nil, "send connack to client %s failed: %s", connect.ClientIdentifier, err)
+		return
+	}
+
 	client.session.updateEGName(b.egName, b.name)
 	topics, qoss, _ := client.session.allSubscribes()
 	if len(topics) > 0 {
 		err = b.topicMgr.subscribe(topics, qoss, client.info.cid)
 		if err != nil {
-			logger.Errorf("client %v use previous session topics %v to subscribe failed: %v", client.info.cid, topics, err)
+			logger.SpanErrorf(nil, "client %v use previous session topics %v to subscribe failed: %v", client.info.cid, topics, err)
 		}
 	}
 	go client.writeLoop()
@@ -234,38 +410,39 @@ func (b *Broker) setSession(client *Client, connect *packets.ConnectPacket) {
 	}
 }
 
-func (b *Broker) requestTransfer(egName, name string, data HTTPJsonData) {
+func (b *Broker) requestTransfer(span *model.SpanContext, egName, name string, data HTTPJsonData, header http.Header) {
 	urls, err := b.memberURL(egName, name)
 	if err != nil {
-		logger.Errorf("find urls for other egs failed:%v", err)
+		logger.SpanErrorf(span, "eg %v find urls for other egs failed:%v", b.egName, err)
 		return
 	}
 	jsonData, err := json.Marshal(data)
 	if err != nil {
-		logger.Errorf("json data marshal failed: %v", err)
+		logger.SpanErrorf(span, "json data marshal failed: %v", err)
 		return
 	}
 	for _, url := range urls {
 		req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(jsonData))
+		req.Header = header.Clone()
 		if err != nil {
-			logger.Errorf("make new request failed: %v", err)
+			logger.SpanErrorf(span, "make new request failed: %v", err)
 			continue
 		}
 		resp, err := http.DefaultClient.Do(req)
 		if err != nil {
-			logger.Errorf("http client send msg failed:%v", err)
+			logger.SpanErrorf(span, "http client send msg failed:%v", err)
 		} else {
 			resp.Body.Close()
 		}
 	}
-	logger.Debugf("http transfer data %v to %v", data, urls)
+	logger.SpanDebugf(span, "eg %v http transfer data %v to %v", b.egName, data, urls)
 }
 
-func (b *Broker) sendMsgToClient(topic string, payload []byte, qos byte) {
-	logger.Debugf("send topic %v to client", topic)
+func (b *Broker) sendMsgToClient(span *model.SpanContext, topic string, payload []byte, qos byte) {
 	subscribers, _ := b.topicMgr.findSubscribers(topic)
+	logger.SpanDebugf(span, "eg %v send topic %v to client %v", b.egName, topic, subscribers)
 	if subscribers == nil {
-		logger.Errorf("not find subscribers for topic %s", topic)
+		logger.SpanErrorf(span, "eg %v not find subscribers for topic %s", b.egName, topic)
 		return
 	}
 
@@ -275,9 +452,9 @@ func (b *Broker) sendMsgToClient(topic string, payload []byte, qos byte) {
 		}
 		client := b.getClient(clientID)
 		if client == nil {
-			logger.Debugf("client %v not on broker %v", clientID, b.name)
+			logger.SpanDebugf(span, "client %v not on broker %v in eg %v", clientID, b.name, b.egName)
 		} else {
-			client.session.publish(topic, payload, qos)
+			client.session.publish(span, topic, payload, qos)
 		}
 	}
 }
@@ -301,7 +478,7 @@ func (b *Broker) removeClient(clientID string) {
 	b.Unlock()
 }
 
-func (b *Broker) topicsPublishHandler(w http.ResponseWriter, r *http.Request) {
+func (b *Broker) httpTopicsPublishHandler(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("suppose POST request but got %s", r.Method))
 		return
@@ -327,30 +504,170 @@ func (b *Broker) topicsPublishHandler(w http.ResponseWriter, r *http.Request) {
 		payload = []byte(data.Payload)
 	}
 
-	logger.Debugf("http endpoint received json data: %v", data)
+	span, _ := b3.ExtractHTTP(r)()
+	logger.SpanDebugf(span, "http endpoint received json data: %v", data)
 	if !data.Distributed {
 		data.Distributed = true
-		b.requestTransfer(b.egName, b.name, data)
+		headers := r.Header.Clone()
+		b.requestTransfer(span, b.egName, b.name, data, headers)
 	}
-	go b.sendMsgToClient(data.Topic, payload, byte(data.QoS))
+	go b.sendMsgToClient(span, data.Topic, payload, byte(data.QoS))
 }
 
-func (b *Broker) mqttAPIPrefix() string {
-	return fmt.Sprintf(mqttAPIPrefix, b.name)
+func (b *Broker) mqttAPIPrefix(path string) string {
+	return fmt.Sprintf(path, b.name)
+}
+
+func (b *Broker) httpGetAllSessionHandler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("suppose Get request but got %s", r.Method))
+		return
+	}
+	span, _ := b3.ExtractHTTP(r)()
+	logger.SpanDebugf(span, "http endpoint receive request to get all session")
+
+	query := r.URL.Query()
+	page := 0
+	pageSize := 0
+	var topic string
+	var err error
+	if len(query) != 0 {
+		pageStr := query.Get("page")
+		pageSizeStr := query.Get("page_size")
+		topic = query.Get("q")
+		if len(pageStr) == 0 || len(pageSizeStr) == 0 {
+			api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("if use query, please provide both page and page_size"))
+			return
+		}
+		page, err = strconv.Atoi(pageStr)
+		if err != nil || page <= 0 {
+			api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("page in query should be number and >0"))
+			return
+		}
+		pageSize, err = strconv.Atoi(pageSizeStr)
+		if err != nil || pageSize <= 0 {
+			api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("page_size in query should be number and >0"))
+			return
+		}
+	}
+
+	allSession, err := b.sessMgr.store.getPrefix(sessionStoreKey(""), false)
+	if err != nil {
+		logger.SpanErrorf(span, "get all sessions with prefix %v failed, %v", sessionStoreKey(""), err)
+		api.HandleAPIError(w, r, http.StatusInternalServerError, fmt.Errorf("get all sessions failed, %v", err))
+		return
+	}
+
+	res := b.queryAllSessions(allSession, len(query) != 0, page, pageSize, topic)
+
+	jsonData, err := json.Marshal(res)
+	if err != nil {
+		logger.SpanErrorf(span, "all session data json marshal failed, %v", err)
+		api.HandleAPIError(w, r, http.StatusInternalServerError, fmt.Errorf("all sessions json marshal failed, %v", err))
+		return
+	}
+	_, err = w.Write(jsonData)
+	if err != nil {
+		logger.SpanErrorf(span, "write json data to http response writer failed, %v", err)
+		api.HandleAPIError(w, r, http.StatusInternalServerError, fmt.Errorf("write json data failed"))
+	}
+}
+
+func (b *Broker) queryAllSessions(allSession map[string]string, query bool, page, pageSize int, topic string) *HTTPSessions {
+	res := &HTTPSessions{}
+	if !query {
+		for k := range allSession {
+			httpSession := &HTTPSession{
+				SessionID: strings.TrimPrefix(k, sessionStoreKey("")),
+			}
+			res.Sessions = append(res.Sessions, httpSession)
+		}
+		return res
+	}
+
+	index := 0
+	start := page*pageSize - pageSize
+	end := page * pageSize
+	for _, v := range allSession {
+		if index >= start && index < end {
+			session := &Session{}
+			session.info = &SessionInfo{}
+			session.decode(v)
+			for k := range session.info.Topics {
+				if strings.Contains(k, topic) {
+					httpSession := &HTTPSession{
+						SessionID: session.info.ClientID,
+						Topic:     k,
+					}
+					res.Sessions = append(res.Sessions, httpSession)
+					break
+				}
+			}
+		}
+		if index > end {
+			break
+		}
+		index++
+	}
+	return res
+}
+
+func (b *Broker) httpDeleteSessionHandler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodDelete {
+		api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("suppose POST request but got %s", r.Method))
+		return
+	}
+	var data HTTPSessions
+	err := json.NewDecoder(r.Body).Decode(&data)
+	if err != nil {
+		api.HandleAPIError(w, r, http.StatusBadRequest, fmt.Errorf("invalid json data from request body"))
+		return
+	}
+
+	span, _ := b3.ExtractHTTP(r)()
+	logger.SpanDebugf(span, "http endpoint received delete session data: %v", data)
+	for _, s := range data.Sessions {
+		err := b.sessMgr.store.delete(sessionStoreKey(s.SessionID))
+		if err != nil {
+			logger.SpanErrorf(span, "delete session %v failed, %v", s, err)
+		}
+	}
+}
+
+func (b *Broker) currentClients() map[string]struct{} {
+	ans := make(map[string]struct{})
+	b.Lock()
+	for k := range b.clients {
+		ans[k] = struct{}{}
+	}
+	b.Unlock()
+	return ans
 }
 
 func (b *Broker) registerAPIs() {
 	group := &api.Group{
 		Group: b.name,
 		Entries: []*api.Entry{
-			{Path: b.mqttAPIPrefix(), Method: http.MethodPost, Handler: b.topicsPublishHandler},
+			{Path: b.mqttAPIPrefix(mqttAPITopicPublishPrefix), Method: http.MethodPost, Handler: b.httpTopicsPublishHandler},
+			{Path: b.mqttAPIPrefix(mqttAPISessionQueryPrefix), Method: http.MethodGet, Handler: b.httpGetAllSessionHandler},
+			{Path: b.mqttAPIPrefix(mqttAPISessionDeletePrefix), Method: http.MethodDelete, Handler: b.httpDeleteSessionHandler},
 		},
 	}
 
 	api.RegisterAPIs(group)
 }
 
+func (b *Broker) setClose() {
+	atomic.StoreInt32(&b.closeFlag, 1)
+}
+
+func (b *Broker) closed() bool {
+	flag := atomic.LoadInt32(&b.closeFlag)
+	return flag == 1
+}
+
 func (b *Broker) close() {
+	b.setClose()
 	close(b.done)
 	b.listener.Close()
 	b.backend.close()
diff --git a/pkg/object/mqttproxy/client.go b/pkg/object/mqttproxy/client.go
index 96fabcf9..ce7999da 100644
--- a/pkg/object/mqttproxy/client.go
+++ b/pkg/object/mqttproxy/client.go
@@ -18,6 +18,7 @@
 package mqttproxy
 
 import (
+	stdcontext "context"
 	"errors"
 	"net"
 	"sync"
@@ -25,7 +26,9 @@ import (
 	"time"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/pipeline"
 )
 
 const (
@@ -56,18 +59,49 @@ type (
 	Client struct {
 		sync.Mutex
 
-		broker  *Broker
-		session *Session
-		conn    net.Conn
+		broker       *Broker
+		session      *Session
+		publishLimit *Limiter
+		conn         net.Conn
 
 		info       ClientInfo
 		statusFlag int32
 		writeCh    chan packets.ControlPacket
 		done       chan struct{}
+
+		// kv map is used for pipeline to share messages among filters during whole connection
+		kvMap sync.Map
 	}
 )
 
-func newClient(connect *packets.ConnectPacket, broker *Broker, conn net.Conn) *Client {
+var _ context.MQTTClient = (*Client)(nil)
+
+// ClientID return client id of Client
+func (c *Client) ClientID() string {
+	return c.info.cid
+}
+
+// Load load value keep in Client kv map
+func (c *Client) Load(key interface{}) (value interface{}, ok bool) {
+	return c.kvMap.Load(key)
+}
+
+// Store store key-value pair in Client kv map
+func (c *Client) Store(key interface{}, value interface{}) {
+	c.kvMap.Store(key, value)
+}
+
+// Delete delete key-value pair in Client kv map
+func (c *Client) Delete(key interface{}) {
+	c.kvMap.Delete(key)
+}
+
+// UserName return username of Client
+func (c *Client) UserName() string {
+	return c.info.username
+}
+
+func newClient(connect *packets.ConnectPacket, broker *Broker, conn net.Conn, limitSpec *RateLimit) *Client {
 	var will *packets.PublishPacket
 	if connect.WillFlag {
 		will = packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
@@ -86,12 +120,13 @@ func newClient(connect *packets.ConnectPacket, broker *Broker, conn net.Conn) *C
 		will:      will,
 	}
 	client := &Client{
-		broker:     broker,
-		conn:       conn,
-		info:       info,
-		statusFlag: Connected,
-		writeCh:    make(chan packets.ControlPacket, 50),
-		done:       make(chan struct{}),
+		broker:       broker,
+		conn:         conn,
+		info:         info,
+		statusFlag:   Connected,
+		writeCh:      make(chan packets.ControlPacket, 50),
+		done:         make(chan struct{}),
+		publishLimit: newLimiter(limitSpec),
 	}
 	return client
 }
@@ -115,14 +150,14 @@ func (c *Client) readLoop() {
 
 		if keepAlive > 0 {
 			if err := c.conn.SetDeadline(time.Now().Add(timeOut)); err != nil {
-				logger.Errorf("set read timeout failed: %s", c.info.cid)
+				logger.SpanErrorf(nil, "set read timeout failed: %s", c.info.cid)
 			}
 		}
 
-		logger.Debugf("client %s readLoop read packet", c.info.cid)
+		logger.SpanDebugf(nil, "client %s readLoop read packet", c.info.cid)
 		packet, err := packets.ReadPacket(c.conn)
 		if err != nil {
-			logger.Errorf("client %s read packet failed: %v", c.info.cid, err)
+			logger.SpanErrorf(nil, "client %s read packet failed: %v", c.info.cid, err)
 			return
 		}
 		if _, ok := packet.(*packets.DisconnectPacket); ok {
@@ -131,7 +166,7 @@ func (c *Client) readLoop() {
 		}
 		err = c.processPacket(packet)
 		if err != nil {
-			logger.Errorf("client %s process packet failed: %v", c.info.cid, err)
+			logger.SpanErrorf(nil, "client %s process packet failed: %v", c.info.cid, err)
 			return
 		}
 	}
@@ -165,14 +200,46 @@ func (c *Client) processPacket(packet packets.ControlPacket) error {
 	default:
 		err = errors.New("unknown packet")
 	}
+	if err != nil {
+		logger.SpanDebugf(nil, "client %v process packet failed, %v", c.info.cid, err)
+	}
 	return err
 }
 
+func (c *Client) checkPublishLimit(publish *packets.PublishPacket) bool {
+	size := publish.RemainingLength + 8
+	return c.publishLimit.acquirePermission(size)
+}
+
 func (c *Client) processPublish(publish *packets.PublishPacket) {
-	logger.Debugf("client %s process publish %v", c.info.cid, publish.TopicName)
+	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
+			}
+		}
+	}
+
 	err := c.broker.backend.publish(publish)
 	if err != nil {
-		logger.Errorf("client %v publish %v failed: %v", c.info.cid, publish.TopicName, err)
+		logger.SpanErrorf(nil, "client %v publish %v failed: %v", c.info.cid, publish.TopicName, err)
 	}
 	switch publish.Qos {
 	case QoS0:
@@ -191,10 +258,10 @@ func (c *Client) processPuback(puback *packets.PubackPacket) {
 }
 
 func (c *Client) processSubscribe(packet *packets.SubscribePacket) {
-	logger.Debugf("client %s processSubscribe %v", c.info.cid, packet.Topics)
+	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.Errorf("client %v subscribe %v failed: %v", c.info.cid, packet.Topics, err)
+		logger.SpanErrorf(nil, "client %v subscribe %v failed: %v", c.info.cid, packet.Topics, err)
 		return
 	}
 	c.session.subscribe(packet.Topics, packet.Qoss)
@@ -209,10 +276,10 @@ func (c *Client) processSubscribe(packet *packets.SubscribePacket) {
 }
 
 func (c *Client) processUnsubscribe(packet *packets.UnsubscribePacket) {
-	logger.Debugf("client %s processUnsubscribe %v", c.info.cid, packet.Topics)
+	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.Errorf("client %v unsubscribe %v failed: %v", c.info.cid, packet.Topics, err)
+		logger.SpanErrorf(nil, "client %v unsubscribe %v failed: %v", c.info.cid, packet.Topics, err)
 	}
 	c.session.unsubscribe(packet.Topics)
 
@@ -236,7 +303,7 @@ func (c *Client) writeLoop() {
 		case p := <-c.writeCh:
 			err := p.Write(c.conn)
 			if err != nil {
-				logger.Errorf("write puback to client %s failed: %s", c.info.cid, err)
+				logger.SpanErrorf(nil, "write packet %v to client %s failed: %s", p.String(), c.info.cid, err)
 				c.closeAndDelSession()
 			}
 		case <-c.done:
@@ -247,12 +314,26 @@ func (c *Client) writeLoop() {
 
 func (c *Client) close() {
 	c.Lock()
-	defer c.Unlock()
 	if c.disconnected() {
+		c.Unlock()
 		return
 	}
+	logger.SpanDebugf(nil, "client %v connection close", c.info.cid)
 	atomic.StoreInt32(&c.statusFlag, Disconnected)
 	close(c.done)
+	c.Unlock()
+
+	// pipeline
+	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 {
+			disconnect := packets.NewControlPacket(packets.Disconnect).(*packets.DisconnectPacket)
+			ctx := context.NewMQTTContext(stdcontext.Background(), c.broker.backend, c, disconnect)
+			pipe.HandleMQTT(ctx)
+		}
+	}
 }
 
 func (c *Client) disconnected() bool {
diff --git a/pkg/object/mqttproxy/mqttproxy.go b/pkg/object/mqttproxy/mqttproxy.go
index 4f4b8a95..5c26ca79 100644
--- a/pkg/object/mqttproxy/mqttproxy.go
+++ b/pkg/object/mqttproxy/mqttproxy.go
@@ -90,10 +90,10 @@ func memberURLFunc(superSpec *supervisor.Spec) func(string, string) ([]string, e
 	c := superSpec.Super().Cluster()
 
 	f := func(egName, name string) ([]string, error) {
-		logger.Debugf("get member url for %v %v", egName, name)
+		logger.SpanDebugf(nil, "get member url for %v %v", egName, name)
 		kv, err := c.GetPrefix(c.Layout().StatusMemberPrefix())
 		if err != nil {
-			logger.Errorf("cluster get member list failed: %v", err)
+			logger.SpanErrorf(nil, "cluster get member list failed: %v", err)
 			return []string{}, err
 		}
 		urls := []string{}
@@ -101,7 +101,7 @@ func memberURLFunc(superSpec *supervisor.Spec) func(string, string) ([]string, e
 			memberStatus := cluster.MemberStatus{}
 			err := yaml.Unmarshal([]byte(v), &memberStatus)
 			if err != nil {
-				logger.Errorf("cluster status unmarshal failed: %v", err)
+				logger.SpanErrorf(nil, "cluster status unmarshal failed: %v", err)
 				return []string{}, err
 			}
 			if memberStatus.Options.Name != egName {
@@ -120,10 +120,10 @@ func memberURLFunc(superSpec *supervisor.Spec) func(string, string) ([]string, e
 				if err != nil {
 					return nil, fmt.Errorf("get url for %v failed: %v", memberStatus.Options.Name, err)
 				}
-				urls = append(urls, newURL+"/apis/v1"+fmt.Sprintf(mqttAPIPrefix, name))
+				urls = append(urls, newURL+"/apis/v1"+fmt.Sprintf(mqttAPITopicPublishPrefix, name))
 			}
 		}
-		logger.Debugf("eg %v %v get urls %v", egName, name, urls)
+		logger.SpanDebugf(nil, "eg %v %v get urls %v", egName, name, urls)
 		return urls, nil
 	}
 	return f
@@ -138,9 +138,10 @@ func (mp *MQTTProxy) Init(superSpec *supervisor.Spec) {
 
 	store := newStorage(superSpec.Super().Cluster())
 	mp.broker = newBroker(spec, store, memberURLFunc(superSpec))
-	if mp.broker != nil {
-		mp.broker.registerAPIs()
+	if mp.broker == nil {
+		panic(fmt.Sprintf("broker %v start failed", spec.Name))
 	}
+	mp.broker.registerAPIs()
 }
 
 // Inherit inherits previous generation of WebSocketServer.
diff --git a/pkg/object/mqttproxy/ratelimiter.go b/pkg/object/mqttproxy/ratelimiter.go
new file mode 100644
index 00000000..2249d577
--- /dev/null
+++ b/pkg/object/mqttproxy/ratelimiter.go
@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mqttproxy
+
+import (
+	"time"
+
+	"github.com/megaease/easegress/pkg/util/ratelimiter"
+)
+
+// Limiter is a rate limiter for MQTTProxy
+type Limiter struct {
+	multiLimiter   *ratelimiter.MultiRateLimiter
+	requestLimiter *ratelimiter.RateLimiter
+	byteLimiter    *ratelimiter.RateLimiter
+}
+
+func newLimiter(spec *RateLimit) *Limiter {
+	limiter := &Limiter{}
+	if spec == nil || (spec.RequestRate == 0 && spec.BytesRate == 0) {
+		return limiter
+	}
+	timePeriod := 1
+	if spec.TimePeriod > 0 {
+		timePeriod = spec.TimePeriod
+	}
+	if spec.RequestRate > 0 && spec.BytesRate > 0 {
+		policy := ratelimiter.NewMultiPolicy(0, time.Duration(timePeriod)*time.Second, []int{spec.RequestRate, spec.BytesRate})
+		l := ratelimiter.NewMulti(policy)
+		limiter.multiLimiter = l
+
+	} else if spec.RequestRate > 0 {
+		policy := ratelimiter.NewPolicy(0, time.Duration(timePeriod)*time.Second, spec.RequestRate)
+		l := ratelimiter.New(policy)
+		limiter.requestLimiter = l
+
+	} else if spec.BytesRate > 0 {
+		policy := ratelimiter.NewPolicy(0, time.Duration(timePeriod)*time.Second, spec.BytesRate)
+		l := ratelimiter.New(policy)
+		limiter.byteLimiter = l
+	}
+	return limiter
+}
+
+func (l *Limiter) acquirePermission(byteNum int) bool {
+	if l.multiLimiter != nil {
+		permitted, _, _ := l.multiLimiter.AcquirePermission([]int{1, byteNum})
+		return permitted
+	}
+	if l.requestLimiter != nil {
+		permitted, _ := l.requestLimiter.AcquirePermission()
+		return permitted
+	}
+	if l.byteLimiter != nil {
+		permitted, _ := l.byteLimiter.AcquireNPermission(byteNum)
+		return permitted
+	}
+	return true
+}
diff --git a/pkg/object/mqttproxy/session.go b/pkg/object/mqttproxy/session.go
index 34f02359..e289bfab 100644
--- a/pkg/object/mqttproxy/session.go
+++ b/pkg/object/mqttproxy/session.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/megaease/easegress/pkg/logger"
+	"github.com/openzipkin/zipkin-go/model"
 )
 
 type (
@@ -70,10 +71,10 @@ func newMsg(topic string, payload []byte, qos byte) *Message {
 }
 
 func (s *Session) store() {
-	logger.Debugf("session %v store", s.info.ClientID)
+	logger.SpanDebugf(nil, "session %v store", s.info.ClientID)
 	str, err := s.encode()
 	if err != nil {
-		logger.Errorf("encode session %+v failed: %v", s, err)
+		logger.SpanErrorf(nil, "encode session %+v failed: %v", s, err)
 		return
 	}
 	ss := SessionStore{
@@ -122,7 +123,7 @@ func (s *Session) updateEGName(egName, name string) {
 }
 
 func (s *Session) subscribe(topics []string, qoss []byte) error {
-	logger.Debugf("session %s sub %v", s.info.ClientID, topics)
+	logger.SpanDebugf(nil, "session %s sub %v", s.info.ClientID, topics)
 	s.Lock()
 	for i, t := range topics {
 		s.info.Topics[t] = int(qoss[i])
@@ -133,7 +134,7 @@ func (s *Session) subscribe(topics []string, qoss []byte) error {
 }
 
 func (s *Session) unsubscribe(topics []string) error {
-	logger.Debugf("session %s unsub %v", s.info.ClientID, topics)
+	logger.SpanDebugf(nil, "session %s unsub %v", s.info.ClientID, topics)
 	s.Lock()
 	for _, t := range topics {
 		delete(s.info.Topics, t)
@@ -144,7 +145,6 @@ func (s *Session) unsubscribe(topics []string) error {
 }
 
 func (s *Session) allSubscribes() ([]string, []byte, error) {
-	logger.Debugf("session %s all sub", s.info.ClientID)
 	s.Lock()
 
 	var sub []string
@@ -169,17 +169,17 @@ func (s *Session) getPacketFromMsg(topic string, payload []byte, qos byte) *pack
 	return p
 }
 
-func (s *Session) publish(topic string, payload []byte, qos byte) {
+func (s *Session) publish(span *model.SpanContext, topic string, payload []byte, qos byte) {
 	client := s.broker.getClient(s.info.ClientID)
 	if client == nil {
-		logger.Errorf("client %s is offline", s.info.ClientID)
+		logger.SpanErrorf(span, "client %s is offline in eg %v", s.info.ClientID, s.broker.egName)
 		return
 	}
 
 	s.Lock()
 	defer s.Unlock()
 
-	logger.Debugf("session %v publish %v", s.info.ClientID, topic)
+	logger.SpanDebugf(span, "session %v publish %v", s.info.ClientID, topic)
 	p := s.getPacketFromMsg(topic, payload, qos)
 	if qos == QoS0 {
 		select {
@@ -192,7 +192,7 @@ func (s *Session) publish(topic string, payload []byte, qos byte) {
 		s.pendingQueue = append(s.pendingQueue, p.MessageID)
 		client.writePacket(p)
 	} else {
-		logger.Errorf("publish message with qos=2 is not supported currently")
+		logger.SpanErrorf(span, "publish message with qos=2 is not supported currently")
 	}
 }
 
@@ -228,7 +228,7 @@ func (s *Session) doResend() {
 			p.TopicName = val.Topic
 			payload, err := base64.StdEncoding.DecodeString(val.B64Payload)
 			if err != nil {
-				logger.Errorf("base64 decode error for Message B64Payload %s", err)
+				logger.SpanErrorf(nil, "base64 decode error for Message B64Payload %s", err)
 				return
 			}
 			p.Payload = payload
@@ -236,7 +236,7 @@ func (s *Session) doResend() {
 			if client != nil {
 				client.writePacket(p)
 			} else {
-				logger.Debugf("session %v do resend but client is nil", s.info.ClientID)
+				logger.SpanDebugf(nil, "session %v do resend but client is nil", s.info.ClientID)
 			}
 			return
 		}
@@ -256,7 +256,7 @@ func (s *Session) backgroundResendPending() {
 			s.doResend()
 		}
 		if time.Now().After(debugLogTime) {
-			logger.Debugf("session %v resend", s.info.ClientID)
+			logger.SpanDebugf(nil, "session %v resend", s.info.ClientID)
 			debugLogTime = time.Now().Add(time.Minute)
 		}
 	}
diff --git a/pkg/object/mqttproxy/session_manager.go b/pkg/object/mqttproxy/session_manager.go
index f08516be..6c8f7b3d 100644
--- a/pkg/object/mqttproxy/session_manager.go
+++ b/pkg/object/mqttproxy/session_manager.go
@@ -62,10 +62,10 @@ func (sm *SessionManager) doStore() {
 		case <-sm.done:
 			return
 		case kv := <-sm.storeCh:
-			logger.Debugf("session manager store session %v", kv.key)
+			logger.SpanDebugf(nil, "session manager store session %v", kv.key)
 			err := sm.store.put(sessionStoreKey(kv.key), kv.value)
 			if err != nil {
-				logger.Errorf("put session %v into storage failed: %v", kv.key, err)
+				logger.SpanErrorf(nil, "put session %v into storage failed: %v", kv.key, err)
 			}
 		}
 	}
@@ -123,6 +123,6 @@ func (sm *SessionManager) delLocal(clientID string) {
 func (sm *SessionManager) delDB(clientID string) {
 	err := sm.store.delete(sessionStoreKey(clientID))
 	if err != nil {
-		logger.Errorf("delete session %v failed, %v", err)
+		logger.SpanErrorf(nil, "delete session %v failed, %v", err)
 	}
 }
diff --git a/pkg/object/mqttproxy/spec.go b/pkg/object/mqttproxy/spec.go
index c055842b..1d4ed5ef 100644
--- a/pkg/object/mqttproxy/spec.go
+++ b/pkg/object/mqttproxy/spec.go
@@ -23,24 +23,41 @@ import (
 )
 
 const (
-	sessionPrefix = "/mqtt/sessionMgr/clientID/%s"
-	topicPrefix   = "/mqtt/topicMgr/topic/%s"
-	mqttAPIPrefix = "/mqttproxy/%s/topics/publish"
+	sessionPrefix              = "/mqtt/sessionMgr/clientID/%s"
+	topicPrefix                = "/mqtt/topicMgr/topic/%s"
+	mqttAPITopicPublishPrefix  = "/mqttproxy/%s/topics/publish"
+	mqttAPISessionQueryPrefix  = "/mqttproxy/%s/session/query"
+	mqttAPISessionDeletePrefix = "/mqttproxy/%s/sessions"
 )
 
 type (
 	// Spec describes the MQTTProxy.
 	Spec struct {
-		EGName         string        `yaml:"-"`
-		Name           string        `yaml:"-"`
-		Port           uint16        `yaml:"port" jsonschema:"required"`
-		BackendType    string        `yaml:"backendType" jsonschema:"required"`
-		Auth           []Auth        `yaml:"auth" jsonschema:"required"`
-		TopicMapper    *TopicMapper  `yaml:"topicMapper" jsonschema:"omitempty"`
-		Kafka          *KafkaSpec    `yaml:"kafkaBroker" jsonschema:"omitempty"`
-		UseTLS         bool          `yaml:"useTLS" jsonschema:"omitempty"`
-		Certificate    []Certificate `yaml:"certificate" jsonschema:"omitempty"`
-		TopicCacheSize int           `yaml:"topicCacheSize" jsonschema:"omitempty"`
+		EGName               string        `yaml:"-"`
+		Name                 string        `yaml:"-"`
+		Port                 uint16        `yaml:"port" jsonschema:"required"`
+		BackendType          string        `yaml:"backendType" jsonschema:"required"`
+		Auth                 []Auth        `yaml:"auth" jsonschema:"required"`
+		TopicMapper          *TopicMapper  `yaml:"topicMapper" jsonschema:"omitempty"`
+		Kafka                *KafkaSpec    `yaml:"kafkaBroker" jsonschema:"omitempty"`
+		UseTLS               bool          `yaml:"useTLS" jsonschema:"omitempty"`
+		Certificate          []Certificate `yaml:"certificate" jsonschema:"omitempty"`
+		TopicCacheSize       int           `yaml:"topicCacheSize" jsonschema:"omitempty"`
+		Pipeline             string        `yaml:"pipeline" jsonschema:"omitempty"`
+		AuthByPipeline       bool          `yaml:"authByPipeline" jsonschema:"omitempty"`
+		MaxAllowedConnection int           `yaml:"maxAllowedConnection" jsonschema:"omitempty"`
+		ConnectionLimit      *RateLimit    `yaml:"connectionLimit" jsonschema:"omitempty"`
+		ClientPublishLimit   *RateLimit    `yaml:"clientPublishLimit" jsonschema:"omitempty"`
+	}
+
+	// RateLimit describes rate limit for connection or publish.
+	// requestRate: max allowed request in time period
+	// timePeriod: max allowed bytes in time period
+	// timePeriod: time of seconds to count requestRate and bytesRate, default 1 second
+	RateLimit struct {
+		RequestRate int `yaml:"requestRate" jsonschema:"omitempty"`
+		BytesRate   int `yaml:"bytesRate" jsonschema:"omitempty"`
+		TimePeriod  int `yaml:"timePeriod" jsonschema:"omitempty"`
 	}
 
 	// Certificate describes TLS certifications.
diff --git a/pkg/object/mqttproxy/storage.go b/pkg/object/mqttproxy/storage.go
index 40e9b494..412146d1 100644
--- a/pkg/object/mqttproxy/storage.go
+++ b/pkg/object/mqttproxy/storage.go
@@ -20,6 +20,7 @@ package mqttproxy
 import (
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/megaease/easegress/pkg/cluster"
 	etcderror "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
@@ -28,14 +29,17 @@ import (
 type (
 	storage interface {
 		get(key string) (*string, error)
-		getPrefix(prefix string) (map[string]string, error)
+		getPrefix(prefix string, keysOnly bool) (map[string]string, error)
 		put(key, value string) error
 		delete(key string) error
+		watchDelete(prefix string) (<-chan map[string]*string, func(), error)
 	}
 
 	mockStorage struct {
-		mu    sync.RWMutex
-		store map[string]string
+		mu        sync.RWMutex
+		store     map[string]string
+		watchCh   chan map[string]*string
+		watchFlag int32
 	}
 
 	clusterStorage struct {
@@ -48,9 +52,14 @@ var _ storage = (*clusterStorage)(nil)
 
 func newStorage(cls cluster.Cluster) storage {
 	if cls != nil {
-		return &clusterStorage{cls: cls}
+		return &clusterStorage{
+			cls: cls,
+		}
+	}
+	return &mockStorage{
+		store:   make(map[string]string),
+		watchCh: make(chan map[string]*string),
 	}
-	return &mockStorage{store: make(map[string]string)}
 }
 
 func (m *mockStorage) get(key string) (*string, error) {
@@ -62,12 +71,16 @@ func (m *mockStorage) get(key string) (*string, error) {
 	return nil, etcderror.ErrKeyNotFound
 }
 
-func (m *mockStorage) getPrefix(prefix string) (map[string]string, error) {
+func (m *mockStorage) getPrefix(prefix string, keysOnly bool) (map[string]string, error) {
 	m.mu.RLock()
 	out := make(map[string]string)
 	for k, v := range m.store {
 		if strings.HasPrefix(k, prefix) {
-			out[k] = v
+			if keysOnly {
+				out[k] = ""
+			} else {
+				out[k] = v
+			}
 		}
 	}
 	m.mu.RUnlock()
@@ -84,15 +97,35 @@ func (m *mockStorage) put(key, value string) error {
 func (m *mockStorage) delete(key string) error {
 	m.mu.Lock()
 	delete(m.store, key)
+	if m.watched() {
+		go func() {
+			ans := make(map[string]*string)
+			ans[key] = nil
+			m.watchCh <- ans
+		}()
+	}
 	m.mu.Unlock()
 	return nil
 }
 
+func (m *mockStorage) watched() bool {
+	flag := atomic.LoadInt32(&m.watchFlag)
+	return flag == 1
+}
+
+func (m *mockStorage) watchDelete(prefix string) (<-chan map[string]*string, func(), error) {
+	atomic.StoreInt32(&m.watchFlag, 1)
+	return m.watchCh, func() {}, nil
+}
+
 func (cs *clusterStorage) get(key string) (*string, error) {
 	return cs.cls.Get(key)
 }
 
-func (cs *clusterStorage) getPrefix(prefix string) (map[string]string, error) {
+func (cs *clusterStorage) getPrefix(prefix string, keysOnly bool) (map[string]string, error) {
+	if keysOnly {
+		return cs.cls.GetWithOp(prefix, cluster.OpPrefix, cluster.OpKeysOnly)
+	}
 	return cs.cls.GetPrefix(prefix)
 }
 
@@ -103,3 +136,16 @@ func (cs *clusterStorage) put(key, value string) error {
 func (cs *clusterStorage) delete(key string) error {
 	return cs.cls.Delete(key)
 }
+
+func (cs *clusterStorage) watchDelete(prefix string) (<-chan map[string]*string, func(), error) {
+	watcher, err := cs.cls.Watcher()
+	if err != nil {
+		return nil, nil, err
+	}
+	ch, err := watcher.WatchWithOp(prefix, cluster.OpPrefix, cluster.OpNotWatchPut)
+	if err != nil {
+		watcher.Close()
+		return nil, nil, err
+	}
+	return ch, watcher.Close, nil
+}
diff --git a/pkg/object/mqttproxy/topic.go b/pkg/object/mqttproxy/topic.go
index d7ea7a95..1f188673 100644
--- a/pkg/object/mqttproxy/topic.go
+++ b/pkg/object/mqttproxy/topic.go
@@ -248,7 +248,7 @@ func getPolicyRoute(routes []*PolicyRe) map[string]*regexp.Regexp {
 	for _, route := range routes {
 		r, err := regexp.Compile(route.MatchExpr)
 		if err != nil {
-			logger.Errorf("topicMapper policy <%s> match expr <%s> compile failed: %v", route.Name, route.MatchExpr, err)
+			logger.SpanErrorf(nil, "topicMapper policy <%s> match expr <%s> compile failed: %v", route.Name, route.MatchExpr, err)
 		} else {
 			ans[route.Name] = r
 		}
@@ -264,7 +264,7 @@ func getTopicRoute(p *Policy) topicRouteType {
 		for _, expr := range route.Exprs {
 			r, err := regexp.Compile(expr)
 			if err != nil {
-				logger.Errorf("topicMapper policy <%s> topic route expr <%s> compile failed: %v", p.Name, expr, err)
+				logger.SpanErrorf(nil, "topicMapper policy <%s> topic route expr <%s> compile failed: %v", p.Name, expr, err)
 			} else {
 				m[route.Topic] = append(m[route.Topic], r)
 			}
diff --git a/pkg/object/pipeline/mock.go b/pkg/object/pipeline/mock.go
new file mode 100644
index 00000000..a432702a
--- /dev/null
+++ b/pkg/object/pipeline/mock.go
@@ -0,0 +1,187 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"sync"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/supervisor"
+)
+
+type mockFilter struct {
+}
+
+type mockSpec struct {
+}
+
+var _ Filter = (*mockFilter)(nil)
+
+func (f *mockFilter) Kind() string                                              { return "MockFilter" }
+func (f *mockFilter) DefaultSpec() interface{}                                  { return &mockSpec{} }
+func (f *mockFilter) Description() string                                       { return "mock filter" }
+func (f *mockFilter) Results() []string                                         { return nil }
+func (f *mockFilter) Init(filterSpec *FilterSpec)                               {}
+func (f *mockFilter) Inherit(filterSpec *FilterSpec, previousGeneration Filter) {}
+func (f *mockFilter) Status() interface{}                                       { return nil }
+func (f *mockFilter) Close()                                                    {}
+func (f *mockFilter) APIs() []*APIEntry                                         { return nil }
+
+// MockMQTTFilter is used for test pipeline, which will count the client number of MQTTContext
+type MockMQTTFilter struct {
+	mockFilter
+	mu sync.Mutex
+
+	spec       *MockMQTTSpec
+	clients    map[string]int
+	disconnect map[string]struct{}
+}
+
+// MockMQTTSpec is spec of MockMQTTFilter
+type MockMQTTSpec struct {
+	UserName               string   `yaml:"userName" jsonschema:"required"`
+	Password               string   `yaml:"password" jsonschema:"required"`
+	Port                   uint16   `yaml:"port" jsonschema:"required"`
+	BackendType            string   `yaml:"backendType" jsonschema:"required"`
+	EarlyStop              bool     `yaml:"earlyStop" jsonschema:"omitempty"`
+	KeysToStore            []string `yaml:"keysToStore" jsonschema:"omitempty"`
+	ConnectKey             string   `yaml:"connectKey" jsonschema:"omitempty"`
+	PublishBackendClientID bool     `yaml:"publishBackendClientID" jsonschema:"omitempty"`
+}
+
+// MockMQTTStatus is status of MockMQTTFilter
+type MockMQTTStatus struct {
+	ClientCount      map[string]int
+	ClientDisconnect map[string]struct{}
+}
+
+var _ MQTTFilter = (*MockMQTTFilter)(nil)
+
+// Kind retrun kind of MockMQTTFilter
+func (m *MockMQTTFilter) Kind() string {
+	return "MockMQTTFilter"
+}
+
+// DefaultSpec retrun default spec of MockMQTTFilter
+func (m *MockMQTTFilter) DefaultSpec() interface{} {
+	return &MockMQTTSpec{}
+}
+
+// Init init MockMQTTFilter
+func (m *MockMQTTFilter) Init(filterSpec *FilterSpec) {
+	m.spec = filterSpec.FilterSpec().(*MockMQTTSpec)
+	m.clients = make(map[string]int)
+	m.disconnect = make(map[string]struct{})
+}
+
+// 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 {
+		if ctx.ConnectPacket().Username != m.spec.UserName || string(ctx.ConnectPacket().Password) != m.spec.Password {
+			ctx.SetDisconnect()
+		}
+	}
+	if ctx.PacketType() == context.MQTTDisconnect {
+		m.disconnect[ctx.Client().ClientID()] = struct{}{}
+	}
+	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)
+	}
+	return nil
+}
+
+func (m *MockMQTTFilter) clientCount() map[string]int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	ans := make(map[string]int)
+	for k, v := range m.clients {
+		ans[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
+	}
+	return ans
+}
+
+// Status return status of MockMQTTFilter
+func (m *MockMQTTFilter) Status() interface{} {
+	return MockMQTTStatus{
+		ClientCount:      m.clientCount(),
+		ClientDisconnect: m.clientDisconnect(),
+	}
+}
+
+type mockMQTTClient struct {
+	cid      string
+	userName string
+	kvMap    sync.Map
+}
+
+var _ context.MQTTClient = (*mockMQTTClient)(nil)
+
+// ClientID return client id of mockMQTTClient
+func (c *mockMQTTClient) ClientID() string {
+	return c.cid
+}
+
+// UserName return username of mockMQTTClient
+func (c *mockMQTTClient) UserName() string {
+	return c.userName
+}
+
+func (c *mockMQTTClient) Load(key interface{}) (interface{}, bool) {
+	return c.kvMap.Load(key)
+}
+
+func (c *mockMQTTClient) Store(key interface{}, value interface{}) {
+	c.kvMap.Store(key, value)
+}
+
+func (c *mockMQTTClient) Delete(key interface{}) {
+	c.kvMap.Delete(key)
+}
+
+// MockFilterSpec help to create FilterSpec for test
+func MockFilterSpec(super *supervisor.Supervisor, rawSpec map[string]interface{}, yamlConfig string,
+	meta *FilterMetaSpec, filterSpec interface{}) *FilterSpec {
+	return &FilterSpec{
+		super:      super,
+		rawSpec:    rawSpec,
+		yamlConfig: yamlConfig,
+		meta:       meta,
+		filterSpec: filterSpec,
+	}
+}
diff --git a/pkg/object/pipeline/pipeline.go b/pkg/object/pipeline/pipeline.go
new file mode 100644
index 00000000..233ec042
--- /dev/null
+++ b/pkg/object/pipeline/pipeline.go
@@ -0,0 +1,196 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/supervisor"
+)
+
+const (
+	// Category is the category of Pipeline.
+	Category = supervisor.CategoryBusinessController
+
+	// Kind is the kind of Pipeline.
+	Kind = "Pipeline"
+)
+
+func init() {
+	supervisor.Register(&Pipeline{})
+}
+
+type (
+	// Pipeline is general pipeline of Easegress
+	Pipeline struct {
+		superSpec      *supervisor.Spec
+		spec           *Spec
+		runningFilters []*runningFilter
+	}
+
+	runningFilter struct {
+		spec   *FilterSpec
+		filter Filter
+	}
+)
+
+var _ supervisor.Controller = (*Pipeline)(nil)
+
+// Category return category of pipeline
+func (p *Pipeline) Category() supervisor.ObjectCategory {
+	return Category
+}
+
+// Kind return kind of pipeline
+func (p *Pipeline) Kind() string {
+	return Kind
+}
+
+// DefaultSpec return default spec of pipeline
+func (p *Pipeline) DefaultSpec() interface{} {
+	return &Spec{}
+}
+
+// Status return status of pipeline
+func (p *Pipeline) Status() *supervisor.Status {
+	s := &Status{
+		Filters: make(map[string]interface{}),
+	}
+	for _, rf := range p.runningFilters {
+		s.Filters[rf.spec.Name()] = rf.filter.Status()
+	}
+	return &supervisor.Status{
+		ObjectStatus: s,
+	}
+}
+
+// Close close pipeline
+func (p *Pipeline) Close() {
+	deletePipeline(p.spec.Name, p.spec.Protocol)
+}
+
+// Init init pipeline
+func (p *Pipeline) Init(superSpec *supervisor.Spec) {
+	p.superSpec, p.spec = superSpec, superSpec.ObjectSpec().(*Spec)
+	p.spec.Name = superSpec.Name()
+	if err := p.spec.Validate(); err != nil {
+		panic(err)
+	}
+	p.reload(nil /*no previous generation*/)
+	storePipeline(p.spec.Name, p.spec.Protocol, p)
+}
+
+// Inherit init new pipeline based on previous pipeline
+func (p *Pipeline) Inherit(superSpec *supervisor.Spec, previousGeneration supervisor.Object) {
+	p.superSpec, p.spec = superSpec, superSpec.ObjectSpec().(*Spec)
+	p.spec.Name = superSpec.Name()
+
+	prev := previousGeneration.(*Pipeline)
+	deletePipeline(prev.spec.Name, p.spec.Protocol)
+	p.reload(prev)
+	storePipeline(p.spec.Name, p.spec.Protocol, p)
+}
+
+// HandleMQTT used to handle MQTT context
+func (p *Pipeline) HandleMQTT(ctx context.MQTTContext) {
+	if p.spec.Protocol != context.MQTT {
+		logger.Errorf("pipeline %s not support protocol MQTT but %s", p.spec.Name, p.spec.Protocol)
+		return
+	}
+	for _, rf := range p.runningFilters {
+		f := rf.filter.(MQTTFilter)
+		f.HandleMQTT(ctx)
+		if ctx.EarlyStop() {
+			return
+		}
+	}
+}
+
+func (p *Pipeline) reload(previousGeneration *Pipeline) {
+	runningFilters := make([]*runningFilter, 0)
+	if len(p.spec.Flow) == 0 {
+		for _, filterSpec := range p.spec.Filters {
+			spec, err := NewFilterSpec(filterSpec, p.superSpec.Super())
+			if err != nil {
+				panic(err)
+			}
+
+			runningFilters = append(runningFilters, &runningFilter{
+				spec: spec,
+			})
+		}
+	} else {
+		filterMap := make(map[string]*FilterSpec)
+		for _, filterSpec := range p.spec.Filters {
+			spec, err := NewFilterSpec(filterSpec, p.superSpec.Super())
+			if err != nil {
+				panic(err)
+			}
+			filterMap[spec.Name()] = spec
+		}
+		for _, f := range p.spec.Flow {
+			if spec, ok := filterMap[f.Filter]; ok {
+				runningFilters = append(runningFilters, &runningFilter{
+					spec: spec,
+				})
+			} else {
+				panic(fmt.Errorf("flow filter %s not found in filters", f.Filter))
+			}
+		}
+	}
+
+	for _, runningFilter := range runningFilters {
+		name, kind := runningFilter.spec.Name(), runningFilter.spec.Kind()
+		rootFilter, exists := filterRegistry[kind]
+		if !exists {
+			panic(fmt.Errorf("kind %s not found", kind))
+		}
+
+		var prevInstance Filter
+		if previousGeneration != nil {
+			runningFilter := previousGeneration.getRunningFilter(name)
+			if runningFilter != nil {
+				prevInstance = runningFilter.filter
+			}
+		}
+
+		filter := reflect.New(reflect.TypeOf(rootFilter).Elem()).Interface().(Filter)
+		runningFilter.spec.meta.Pipeline = p.spec.Name
+		runningFilter.spec.meta.Protocol = p.spec.Protocol
+		if prevInstance == nil {
+			filter.Init(runningFilter.spec)
+		} else {
+			filter.Inherit(runningFilter.spec, prevInstance)
+		}
+		runningFilter.filter = filter
+
+	}
+	p.runningFilters = runningFilters
+}
+
+func (p *Pipeline) getRunningFilter(name string) *runningFilter {
+	for _, filter := range p.runningFilters {
+		if filter.spec.Name() == name {
+			return filter
+		}
+	}
+	return nil
+}
diff --git a/pkg/object/pipeline/registry.go b/pkg/object/pipeline/registry.go
new file mode 100644
index 00000000..cdcdd6a3
--- /dev/null
+++ b/pkg/object/pipeline/registry.go
@@ -0,0 +1,201 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"fmt"
+	"net/http"
+	"reflect"
+	"sync"
+
+	"github.com/megaease/easegress/pkg/context"
+)
+
+type (
+	// Filter is the common interface for filters.
+	Filter interface {
+		// Kind returns the unique kind name to represent itself.
+		Kind() string
+
+		// DefaultSpec returns the default spec.
+		DefaultSpec() interface{}
+
+		// Description returns the description of the filter.
+		Description() string
+
+		// Results returns all possible results, excluding the default result value (empty string).
+		Results() []string
+
+		// Init initializes the Filter.
+		Init(filterSpec *FilterSpec)
+
+		// Inherit also initializes the Filter.
+		// But it needs to handle the lifecycle of the previous generation.
+		// So it is Filter's responsibility to inherit and clean the previous generation.
+		// The http pipeline won't call Close for the previous generation.
+		Inherit(filterSpec *FilterSpec, previousGeneration Filter)
+
+		// Status returns its runtime status.
+		// It could return nil.
+		Status() interface{}
+
+		// Close closes itself.
+		Close()
+	}
+
+	// HTTPFilter is the common interface for filters to handle http traffic.
+	HTTPFilter interface {
+		Filter
+
+		// Handle handles one HTTP request, all possible results
+		// need be registered in Results.
+		HandleHTTP(context.HTTPContext) *context.HTTPResult
+	}
+
+	// MQTTFilter is the common interface for filters to handle mqtt traffic.
+	MQTTFilter interface {
+		Filter
+
+		// Handle handles one MQTT request, all possible results
+		// need be registered in Results.
+		HandleMQTT(context.MQTTContext) *context.MQTTResult
+	}
+
+	// TCPFilter is the common interface for filters to handle tcp traffic.
+	TCPFilter interface {
+		Filter
+
+		HandleTCP(context.TCPContext) *context.TCPResult
+	}
+
+	// UDPFilter is the common interface for filters to handle udp traffic.
+	UDPFilter interface {
+		Filter
+		HandleUDP(context.UDPContext) *context.UDPResult
+	}
+
+	// APIEntry contains filter api information
+	APIEntry struct {
+		Path    string
+		Method  string
+		Handler http.HandlerFunc
+	}
+)
+
+func getProtocols(f Filter) (map[context.Protocol]struct{}, error) {
+	ans := map[context.Protocol]struct{}{}
+	if _, ok := f.(HTTPFilter); ok {
+		ans[context.HTTP] = struct{}{}
+	}
+	if _, ok := f.(MQTTFilter); ok {
+		ans[context.MQTT] = struct{}{}
+	}
+	if _, ok := f.(TCPFilter); ok {
+		ans[context.TCP] = struct{}{}
+	}
+	if _, ok := f.(UDPFilter); ok {
+		ans[context.UDP] = struct{}{}
+	}
+	if len(ans) == 0 {
+		return nil, fmt.Errorf("filter %v protocol not found, currently only support HTTP, MQTT and TCP", f.Kind())
+	}
+	return ans, nil
+}
+
+var runningPipelines = sync.Map{}
+var filterRegistry = map[string]Filter{}
+
+func pipelineName(name string, protocol context.Protocol) string {
+	return string(protocol) + "/" + name
+}
+
+// GetPipeline is used to get pipeline with given name and protocol
+func GetPipeline(name string, protocol context.Protocol) (*Pipeline, error) {
+	key := pipelineName(name, protocol)
+	if value, ok := runningPipelines.Load(key); ok {
+		return value.(*Pipeline), nil
+	}
+	return nil, fmt.Errorf("no running pipeline for %v %v", protocol, name)
+}
+
+func storePipeline(name string, protocol context.Protocol, pipeline *Pipeline) {
+	_, loaded := runningPipelines.LoadOrStore(pipelineName(name, protocol), pipeline)
+	if loaded {
+		panic(fmt.Errorf("pipeline %v %v already exists", name, protocol))
+	}
+}
+
+func deletePipeline(name string, protocol context.Protocol) {
+	_, loaded := runningPipelines.LoadAndDelete(pipelineName(name, protocol))
+	if !loaded {
+		panic(fmt.Errorf("pipeline %v %v not exists", name, protocol))
+	}
+}
+
+// Register registers filter.
+func Register(f Filter) {
+	if f.Kind() == "" {
+		panic(fmt.Errorf("%T: empty kind", f))
+	}
+
+	existedFilter, existed := filterRegistry[f.Kind()]
+	if existed {
+		panic(fmt.Errorf("%T and %T got same kind: %s", f, existedFilter, f.Kind()))
+	}
+
+	// Checking filter type.
+	filterType := reflect.TypeOf(f)
+	if filterType.Kind() != reflect.Ptr {
+		panic(fmt.Errorf("%s: want a pointer, got %s", f.Kind(), filterType.Kind()))
+	}
+	if filterType.Elem().Kind() != reflect.Struct {
+		panic(fmt.Errorf("%s elem: want a struct, got %s", f.Kind(), filterType.Kind()))
+	}
+
+	// Checking spec type.
+	specType := reflect.TypeOf(f.DefaultSpec())
+	if specType.Kind() != reflect.Ptr {
+		panic(fmt.Errorf("%s spec: want a pointer, got %s", f.Kind(), specType.Kind()))
+	}
+	if specType.Elem().Kind() != reflect.Struct {
+		panic(fmt.Errorf("%s spec elem: want a struct, got %s", f.Kind(), specType.Elem().Kind()))
+	}
+
+	// Checking results.
+	results := make(map[string]struct{})
+	for _, result := range f.Results() {
+		_, exists := results[result]
+		if exists {
+			panic(fmt.Errorf("repeated result: %s", result))
+		}
+		results[result] = struct{}{}
+	}
+
+	filterRegistry[f.Kind()] = f
+}
+
+// GetFilterRegistry get the filter registry.
+func GetFilterRegistry() map[string]Filter {
+	result := map[string]Filter{}
+
+	for kind, f := range filterRegistry {
+		result[kind] = f
+	}
+
+	return result
+}
diff --git a/pkg/object/pipeline/spec.go b/pkg/object/pipeline/spec.go
new file mode 100644
index 00000000..488e050d
--- /dev/null
+++ b/pkg/object/pipeline/spec.go
@@ -0,0 +1,230 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"fmt"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+	"github.com/megaease/easegress/pkg/v"
+)
+
+type (
+	// Spec describes the Pipeline.
+	Spec struct {
+		Name     string                   `yaml:"-" jsonschema:"-"`
+		Protocol context.Protocol         `yaml:"protocol" jsonschema:"required"`
+		Flow     []Flow                   `yaml:"flow" jsonschema:"omitempty"`
+		Filters  []map[string]interface{} `yaml:"filters" jsonschema:"required"`
+	}
+
+	// Flow controls the flow of pipeline.
+	Flow struct {
+		Filter string `yaml:"filter" jsonschema:"required,format=urlname"`
+	}
+
+	// FilterSpec is the universal spec for all filters.
+	FilterSpec struct {
+		super *supervisor.Supervisor
+
+		rawSpec    map[string]interface{}
+		yamlConfig string
+		meta       *FilterMetaSpec
+		filterSpec interface{}
+	}
+
+	// FilterMetaSpec is metadata for all specs.
+	FilterMetaSpec struct {
+		Name     string           `yaml:"name" jsonschema:"required,format=urlname"`
+		Kind     string           `yaml:"kind" jsonschema:"required"`
+		Pipeline string           `yaml:"-" jsonschema:"-"`
+		Protocol context.Protocol `yaml:"-" jsonschema:"-"`
+	}
+
+	// Status is the status of HTTPPipeline.
+	Status struct {
+		Health string `yaml:"health"`
+
+		Filters map[string]interface{} `yaml:"filters"`
+	}
+)
+
+func extractFiltersData(config []byte) interface{} {
+	var whole map[string]interface{}
+	yamltool.Unmarshal(config, &whole)
+	return whole["filters"]
+}
+
+// creates FilterSpecs from a list of filters
+func filtersToFilterSpecs(filters []map[string]interface{}, super *supervisor.Supervisor) (map[string]*FilterSpec, []string) {
+	filterMap := make(map[string]*FilterSpec)
+	filterNames := []string{}
+	for _, filter := range filters {
+		spec, err := NewFilterSpec(filter, super)
+		if err != nil {
+			panic(err)
+		}
+		if _, exists := filterMap[spec.Name()]; exists {
+			panic(fmt.Errorf("conflict name: %s", spec.Name()))
+		}
+		filterMap[spec.Name()] = spec
+		filterNames = append(filterNames, spec.Name())
+	}
+	return filterMap, filterNames
+}
+
+// Validate validates Spec.
+func (s Spec) Validate() (err error) {
+	errPrefix := "filters"
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("%s: %s", errPrefix, r)
+		}
+	}()
+
+	config := yamltool.Marshal(s)
+
+	filtersData := extractFiltersData(config)
+	if filtersData == nil {
+		return fmt.Errorf("filters is required")
+	}
+
+	filterSpecs, _ := filtersToFilterSpecs(
+		s.Filters,
+		nil, /*NOTE: Nil supervisor is fine in spec validating phrase.*/
+	)
+
+	for _, spec := range filterSpecs {
+		kind := spec.Kind()
+		filter, exists := filterRegistry[kind]
+		if !exists {
+			panic(fmt.Errorf("kind %s not found", kind))
+		}
+		protocols, err := getProtocols(filter)
+		if err != nil {
+			panic(fmt.Errorf("filter %v get protocols failed, %v", kind, err))
+		}
+		if _, ok := protocols[s.Protocol]; !ok {
+			panic(fmt.Errorf("filter %v not support pipeline protocol %s", spec.Name(), s.Protocol))
+		}
+	}
+
+	errPrefix = "flow"
+	filters := make(map[string]struct{})
+	for _, f := range s.Flow {
+		if _, exists := filters[f.Filter]; exists {
+			panic(fmt.Errorf("repeated filter %s", f.Filter))
+		}
+	}
+	return nil
+}
+
+// NewFilterSpec creates a filter spec and validates it.
+func NewFilterSpec(originalRawSpec map[string]interface{}, super *supervisor.Supervisor) (
+	s *FilterSpec, err error) {
+
+	s = &FilterSpec{super: super}
+
+	defer func() {
+		if r := recover(); r != nil {
+			s = nil
+			err = fmt.Errorf("%v", r)
+		} else {
+			err = nil
+		}
+	}()
+
+	yamlBuff := yamltool.Marshal(originalRawSpec)
+
+	// Meta part.
+	meta := &FilterMetaSpec{}
+	yamltool.Unmarshal(yamlBuff, meta)
+	verr := v.Validate(meta)
+	if !verr.Valid() {
+		panic(verr)
+	}
+
+	// Filter self part.
+	rootFilter, exists := filterRegistry[meta.Kind]
+	if !exists {
+		panic(fmt.Errorf("kind %s not found", meta.Kind))
+	}
+	filterSpec := rootFilter.DefaultSpec()
+	yamltool.Unmarshal(yamlBuff, filterSpec)
+	verr = v.Validate(filterSpec)
+	if !verr.Valid() {
+		// TODO: Make the invalid part more accurate. e,g:
+		// filters: jsonschemaErrs:
+		// - 'policies.0: name is required'
+		// to
+		// filters: jsonschemaErrs:
+		// - 'rateLimiter.policies.0: name is required'
+		panic(verr)
+	}
+
+	// Build final yaml config and raw spec.
+	var rawSpec map[string]interface{}
+	filterBuff := yamltool.Marshal(filterSpec)
+	yamltool.Unmarshal(filterBuff, &rawSpec)
+
+	metaBuff := yamltool.Marshal(meta)
+	yamltool.Unmarshal(metaBuff, &rawSpec)
+
+	yamlConfig := string(yamltool.Marshal(rawSpec))
+
+	s.meta = meta
+	s.filterSpec = filterSpec
+	s.rawSpec = rawSpec
+	s.yamlConfig = yamlConfig
+
+	return
+}
+
+// Super returns super
+func (s *FilterSpec) Super() *supervisor.Supervisor {
+	return s.super
+}
+
+// Name returns name.
+func (s *FilterSpec) Name() string { return s.meta.Name }
+
+// Kind returns kind.
+func (s *FilterSpec) Kind() string { return s.meta.Kind }
+
+// Pipeline returns the name of the pipeline this filter belongs to.
+func (s *FilterSpec) Pipeline() string { return s.meta.Pipeline }
+
+// Protocol return protocol for this filter
+func (s *FilterSpec) Protocol() context.Protocol { return s.meta.Protocol }
+
+// YAMLConfig returns the config in yaml format.
+func (s *FilterSpec) YAMLConfig() string {
+	return s.yamlConfig
+}
+
+// RawSpec returns raw spec in type map[string]interface{}.
+func (s *FilterSpec) RawSpec() map[string]interface{} {
+	return s.rawSpec
+}
+
+// FilterSpec returns the filter spec in its own type.
+func (s *FilterSpec) FilterSpec() interface{} {
+	return s.filterSpec
+}
diff --git a/pkg/registry/registry.go b/pkg/registry/registry.go
index 24ff157e..8cabf6b5 100644
--- a/pkg/registry/registry.go
+++ b/pkg/registry/registry.go
@@ -23,6 +23,7 @@ import (
 	_ "github.com/megaease/easegress/pkg/filter/apiaggregator"
 	_ "github.com/megaease/easegress/pkg/filter/bridge"
 	_ "github.com/megaease/easegress/pkg/filter/circuitbreaker"
+	_ "github.com/megaease/easegress/pkg/filter/connectcontrol"
 	_ "github.com/megaease/easegress/pkg/filter/corsadaptor"
 	_ "github.com/megaease/easegress/pkg/filter/fallback"
 	_ "github.com/megaease/easegress/pkg/filter/meshadaptor"
@@ -51,6 +52,7 @@ import (
 	_ "github.com/megaease/easegress/pkg/object/meshcontroller"
 	_ "github.com/megaease/easegress/pkg/object/mqttproxy"
 	_ "github.com/megaease/easegress/pkg/object/nacosserviceregistry"
+	_ "github.com/megaease/easegress/pkg/object/pipeline"
 	_ "github.com/megaease/easegress/pkg/object/rawconfigtrafficcontroller"
 	_ "github.com/megaease/easegress/pkg/object/trafficcontroller"
 	_ "github.com/megaease/easegress/pkg/object/websocketserver"
diff --git a/pkg/util/ratelimiter/multiratelimiter.go b/pkg/util/ratelimiter/multiratelimiter.go
new file mode 100644
index 00000000..37e72fcb
--- /dev/null
+++ b/pkg/util/ratelimiter/multiratelimiter.go
@@ -0,0 +1,172 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ratelimiter
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+type (
+	// MultiPolicy defines the policy of a rate limiter
+	MultiPolicy struct {
+		TimeoutDuration    time.Duration
+		LimitRefreshPeriod time.Duration
+		LimitForPeriod     []int
+	}
+
+	// MultiRateLimiter defines a rate limiter
+	MultiRateLimiter struct {
+		lock      sync.Mutex
+		state     State
+		policy    *MultiPolicy
+		startTime time.Time
+		cycle     int
+		tokens    []int
+	}
+)
+
+// NewMultiPolicy create and initialize a policy
+func NewMultiPolicy(timeout, refresh time.Duration, limit []int) *MultiPolicy {
+	return &MultiPolicy{
+		TimeoutDuration:    timeout,
+		LimitRefreshPeriod: refresh,
+		LimitForPeriod:     limit,
+	}
+}
+
+// NewMulti creates a multi rate limiter based on `policy`,
+func NewMulti(policy *MultiPolicy) *MultiRateLimiter {
+	rl := &MultiRateLimiter{
+		policy:    policy,
+		startTime: nowFunc(),
+		tokens:    make([]int, len(policy.LimitForPeriod)),
+	}
+	return rl
+}
+
+// SetState sets the state of the rate limiter to `state`
+func (rl *MultiRateLimiter) SetState(state State) {
+	rl.lock.Lock()
+	defer rl.lock.Unlock()
+	if rl.state == state {
+		return
+	}
+	if rl.state == StateDisabled {
+		rl.cycle = 0
+		rl.tokens = make([]int, len(rl.policy.LimitForPeriod))
+		rl.startTime = nowFunc()
+	}
+	rl.state = state
+}
+
+// AcquirePermission acquires a permission from the rate limiter.
+// returns true if the request is permitted and false otherwise.
+// when permitted, the caller should wait returned duration before action.
+func (rl *MultiRateLimiter) AcquirePermission(count []int) (bool, time.Duration, error) {
+	rl.lock.Lock()
+	defer rl.lock.Unlock()
+
+	if rl.state == StateDisabled {
+		return true, 0, nil
+	}
+	if len(count) != len(rl.policy.LimitForPeriod) {
+		return false, 0, fmt.Errorf("MultiRateLimiter policy contain %v tokens but get %v tokens", rl.policy.LimitForPeriod, count)
+	}
+
+	now := nowFunc()
+
+	// max tokens could be permitted(including reserved) from the beginning of current cycle
+	maxTokens := make([]int, len(rl.policy.LimitForPeriod))
+	copy(maxTokens, rl.policy.LimitForPeriod)
+	scale := int(rl.policy.TimeoutDuration/rl.policy.LimitRefreshPeriod) + 1
+	for i := range maxTokens {
+		maxTokens[i] *= scale
+	}
+
+	// current cycle index
+	cycle := int(now.Sub(rl.startTime) / rl.policy.LimitRefreshPeriod)
+
+	// rl.tokens is the number of permitted tokens from the beginning of rl.cycle
+	// tokens is the number of tokens have already been permitted from the beginning
+	// of current cycle. note tokens could be less than zero
+	tokens := make([]int, len(rl.tokens))
+	for i, token := range rl.tokens {
+		newToken := token - (cycle-rl.cycle)*rl.policy.LimitForPeriod[i]
+		if newToken < 0 {
+			newToken = 0
+		}
+		tokens[i] = newToken
+	}
+
+	// reject if already reached the permission limitation
+	for i, token := range tokens {
+		if token >= maxTokens[i] {
+			return false, rl.policy.TimeoutDuration, nil
+		}
+	}
+
+	// permit another token
+	for i := range rl.tokens {
+		rl.tokens[i] = tokens[i] + count[i]
+	}
+	rl.cycle = cycle
+
+	// if there are still free tokens in current cycle
+	allFree := true
+	for i, token := range tokens {
+		if token >= rl.policy.LimitForPeriod[i] {
+			allFree = false
+			break
+		}
+	}
+	if allFree {
+		if rl.state != StateNormal {
+			rl.state = StateNormal
+		}
+		return true, 0, nil
+	}
+
+	// no free tokens in current cycle, we can permit the request, but we need also
+	// return a duration which the caller must wait before proceed.
+
+	var timeToWait time.Duration
+	for i, token := range tokens {
+		tmpCycle := cycle + token/rl.policy.LimitForPeriod[i]
+		d := rl.policy.LimitRefreshPeriod * time.Duration(tmpCycle)
+		wait := rl.startTime.Add(d).Sub(now)
+		if wait > timeToWait {
+			timeToWait = wait
+		}
+	}
+	return true, timeToWait, nil
+}
+
+// WaitPermission waits a permission from the rate limiter
+// returns true if the request is permitted and false if timed out
+func (rl *MultiRateLimiter) WaitPermission(counts []int) (bool, error) {
+	permitted, d, err := rl.AcquirePermission(counts)
+	if err != nil {
+		return false, err
+	}
+	if d > 0 {
+		time.Sleep(d)
+	}
+	return permitted, nil
+}
diff --git a/pkg/util/ratelimiter/ratelimiter.go b/pkg/util/ratelimiter/ratelimiter.go
index 147fef94..2650c72d 100644
--- a/pkg/util/ratelimiter/ratelimiter.go
+++ b/pkg/util/ratelimiter/ratelimiter.go
@@ -73,15 +73,15 @@ var stateStrings = []string{
 // NewPolicy create and initialize a policy
 func NewPolicy(timeout, refresh time.Duration, limit int) *Policy {
 	return &Policy{
-		TimeoutDuration:    timeout * time.Millisecond,
-		LimitRefreshPeriod: refresh * time.Millisecond,
+		TimeoutDuration:    timeout,
+		LimitRefreshPeriod: refresh,
 		LimitForPeriod:     limit,
 	}
 }
 
 // NewDefaultPolicy create and initialize a policy with default configuration
 func NewDefaultPolicy() *Policy {
-	return NewPolicy(100, 10, 50)
+	return NewPolicy(100*time.Millisecond, 10*time.Millisecond, 50)
 }
 
 // New creates a rate limiter based on `policy`,
@@ -125,10 +125,7 @@ func (rl *RateLimiter) notifyListener(tm time.Time, state State) {
 	}
 }
 
-// AcquirePermission acquires a permission from the rate limiter.
-// returns true if the request is permitted and false otherwise.
-// when permitted, the caller should wait returned duration before action.
-func (rl *RateLimiter) AcquirePermission() (bool, time.Duration) {
+func (rl *RateLimiter) acquirePermission(count int) (bool, time.Duration) {
 	rl.lock.Lock()
 	defer rl.lock.Unlock()
 
@@ -154,12 +151,12 @@ func (rl *RateLimiter) AcquirePermission() (bool, time.Duration) {
 	}
 
 	// reject if already reached the permission limitation
-	if tokens == maxTokens {
+	if tokens >= maxTokens {
 		return false, rl.policy.TimeoutDuration
 	}
 
 	// permit another token
-	rl.tokens = tokens + 1
+	rl.tokens = tokens + count
 	rl.cycle = cycle
 
 	// if there are still free tokens in current cycle
@@ -186,6 +183,20 @@ func (rl *RateLimiter) AcquirePermission() (bool, time.Duration) {
 	return true, timeToWait
 }
 
+// AcquirePermission acquires a permission from the rate limiter.
+// returns true if the request is permitted and false otherwise.
+// when permitted, the caller should wait returned duration before action.
+func (rl *RateLimiter) AcquirePermission() (bool, time.Duration) {
+	return rl.acquirePermission(1)
+}
+
+// AcquireNPermission acquires N permission from the rate limiter.
+// returns true if the request is permitted and false otherwise.
+// when permitted, the caller should wait returned duration before action.
+func (rl *RateLimiter) AcquireNPermission(n int) (bool, time.Duration) {
+	return rl.acquirePermission(n)
+}
+
 // WaitPermission waits a permission from the rate limiter
 // returns true if the request is permitted and false if timed out
 func (rl *RateLimiter) WaitPermission() bool {


Test patch:
diff --git a/pkg/cluster/cluster_interface_test.go b/pkg/cluster/cluster_interface_test.go
new file mode 100644
index 00000000..8d954612
--- /dev/null
+++ b/pkg/cluster/cluster_interface_test.go
@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cluster
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+func TestGetOpOption(t *testing.T) {
+	tests := []struct {
+		wantFunc clientv3.OpOption
+		op       ClientOp
+	}{
+		{clientv3.WithPrefix(), OpPrefix},
+		{clientv3.WithFilterPut(), OpNotWatchPut},
+		{clientv3.WithFilterDelete(), OpNotWatchDelete},
+		{clientv3.WithKeysOnly(), OpKeysOnly},
+	}
+	for _, tc := range tests {
+		op1 := clientv3.Op{}
+		op2 := clientv3.Op{}
+		tc.wantFunc(&op1)
+		getOpOption(tc.op)(&op2)
+		assert.Equal(t, op1, op2)
+	}
+}
diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go
index f85267ba..7424597b 100644
--- a/pkg/cluster/cluster_test.go
+++ b/pkg/cluster/cluster_test.go
@@ -24,7 +24,6 @@ import (
 	"time"
 
 	"go.etcd.io/etcd/api/v3/mvccpb"
-	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
 )
 
@@ -225,22 +224,14 @@ func TestClusterSyncer(t *testing.T) {
 
 	c.Put("/akey", "avalue")
 
-	for {
-		var value *string
-		select {
-		case value = <-schan:
-			fmt.Printf("sync value is %v\n", value)
-		}
-		break
+	{
+		value := <-schan
+		fmt.Printf("sync value is %v\n", value)
 	}
 
-	for {
-		var value *mvccpb.KeyValue
-		select {
-		case value = <-rschan:
-			fmt.Printf("sync raw value is %v\n", value)
-		}
-		break
+	{
+		value := <-rschan
+		fmt.Printf("sync raw value is %v\n", value)
 	}
 
 	pchan, err := syncer.SyncPrefix("/abcd")
@@ -258,23 +249,18 @@ func TestClusterSyncer(t *testing.T) {
 	if _, err = c.GetPrefix("/abcd"); err != nil {
 		t.Errorf("cluster get prefix failed: %v", err)
 	}
+	if _, err = c.GetWithOp("/abcd", OpPrefix); err != nil {
+		t.Errorf("cluster get prefix failed: %v", err)
+	}
 
-	for {
-		value := make(map[string]string, 1)
-		select {
-		case value = <-pchan:
-			fmt.Printf("sync prefix value is %v\n", value)
-		}
-		break
+	{
+		value := <-pchan
+		fmt.Printf("sync prefix value is %v\n", value)
 	}
 
-	for {
-		value := make(map[string]*mvccpb.KeyValue)
-		select {
-		case value = <-rpchan:
-			fmt.Printf("sync raw pvalue is %v\n", value)
-		}
-		break
+	{
+		value := <-rpchan
+		fmt.Printf("sync raw pvalue is %v\n", value)
 	}
 
 	syncer.Close()
@@ -318,28 +304,24 @@ func TestClusterWatcher(t *testing.T) {
 
 	c.Put("/akey/value", "yes")
 
-	for {
-		var value *string
-		select {
-		case value = <-wchan:
-			fmt.Printf("watch value is %v\n", value)
-		}
-		break
+	{
+		value := <-wchan
+		fmt.Printf("watch value is %v\n", value)
 	}
 
-	for {
-		var value *clientv3.Event
-		select {
-		case value = <-rchan:
-			fmt.Printf("watch raw value is %v\n", value)
-		}
-		break
+	{
+		value := <-rchan
+		fmt.Printf("watch raw value is %v\n", value)
 	}
 
 	pchan, err := watcher.WatchPrefix("/ab")
 	if err != nil {
 		t.Errorf("watcher watch failed: %v", err)
 	}
+	opchan, err := watcher.WatchWithOp("/ab", OpPrefix)
+	if err != nil {
+		t.Errorf("watcher watch failed: %v", err)
+	}
 
 	rawpchan, err := watcher.WatchRawPrefix("/abcd")
 	if err != nil {
@@ -349,44 +331,39 @@ func TestClusterWatcher(t *testing.T) {
 
 	c.Put("/abc", "kkk")
 
-	for {
-		value := make(map[string]*string, 1)
-		select {
-		case value = <-pchan:
-			fmt.Printf("watch prefix value is %v\n", value)
-		}
-		break
+	{
+		value := <-pchan
+		fmt.Printf("watch prefix value is %v\n", value)
+	}
+
+	{
+		value := <-opchan
+		fmt.Printf("watch prefix value is %v\n", value)
 	}
 	c.Put("/abcd/ef", "jjj")
 
-	for {
-		value := make(map[string]*clientv3.Event, 1)
-		select {
-		case value = <-rawpchan:
-			fmt.Printf("watch prefix raw value is %v\n", value)
-		}
-		break
+	{
+		value := <-rawpchan
+		fmt.Printf("watch prefix raw value is %v\n", value)
 	}
+
 	c.DeletePrefix("/abc")
 
-	for {
-		value := make(map[string]*string, 1)
-		select {
-		case value = <-pchan:
-			fmt.Printf("watch delete prefix value is %v\n", value)
-		}
-		break
+	{
+		value := <-pchan
+		fmt.Printf("watch delete prefix value is %v\n", value)
+	}
+
+	{
+		value := <-opchan
+		fmt.Printf("watch delete prefix value is %v\n", value)
 	}
 
 	c.DeletePrefix("/abcd/")
 
-	for {
-		value := make(map[string]*clientv3.Event, 1)
-		select {
-		case value = <-rawpchan:
-			fmt.Printf("watch delete prefix raw value is %v\n", value)
-		}
-		break
+	{
+		value := <-rawpchan
+		fmt.Printf("watch delete prefix raw value is %v\n", value)
 	}
 
 	watcher.Close()
diff --git a/pkg/context/contexttest/context.go b/pkg/context/contexttest/context.go
index 78b03aec..c00f1405 100644
--- a/pkg/context/contexttest/context.go
+++ b/pkg/context/contexttest/context.go
@@ -55,6 +55,11 @@ type MockedHTTPContext struct {
 	MockedSetHandlerCaller   func(caller context.HandlerCaller)
 }
 
+// Protocol return protocol of MockedHTTPContext
+func (c *MockedHTTPContext) Protocol() context.Protocol {
+	return context.HTTP
+}
+
 // Lock mocks the Lock function of HTTPContext
 func (c *MockedHTTPContext) Lock() {
 	if c.MockedLock != nil {
diff --git a/pkg/filter/connectcontrol/connectcontrol_test.go b/pkg/filter/connectcontrol/connectcontrol_test.go
new file mode 100644
index 00000000..25fb65fc
--- /dev/null
+++ b/pkg/filter/connectcontrol/connectcontrol_test.go
@@ -0,0 +1,191 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package connectcontrol
+
+import (
+	stdcontext "context"
+	"errors"
+	"testing"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/pipeline"
+	"github.com/stretchr/testify/assert"
+)
+
+func init() {
+	logger.InitNop()
+}
+
+func newContext(cid string, topic string) context.MQTTContext {
+	backend := &context.MockMQTTBackend{
+		Messages: make(map[string]context.MockMQTTMsg),
+	}
+	client := &context.MockMQTTClient{
+		MockClientID: cid,
+	}
+	packet := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
+	packet.TopicName = topic
+	return context.NewMQTTContext(stdcontext.Background(), backend, client, packet)
+}
+
+func defaultFilterSpec(spec *Spec) *pipeline.FilterSpec {
+	meta := &pipeline.FilterMetaSpec{
+		Name:     "connect-control-demo",
+		Kind:     Kind,
+		Pipeline: "pipeline-demo",
+		Protocol: context.MQTT,
+	}
+	filterSpec := pipeline.MockFilterSpec(nil, nil, "", meta, spec)
+	return filterSpec
+}
+
+func TestConnectControl(t *testing.T) {
+	assert := assert.New(t)
+
+	cc := &ConnectControl{}
+	assert.Equal(cc.Kind(), Kind, "wrong kind")
+	assert.Equal(cc.DefaultSpec(), &Spec{}, "wrong spec")
+	assert.NotEqual(len(cc.Description()), 0, "description for ConnectControl is empty")
+
+	assert.NotNil(cc.Results(), "if update result, please update this case")
+	checkProtocol := func() (err error) {
+		defer func() {
+			if errMsg := recover(); errMsg != nil {
+				err = errors.New(errMsg.(string))
+				return
+			}
+		}()
+		meta := &pipeline.FilterMetaSpec{
+			Protocol: context.HTTP,
+		}
+		filterSpec := pipeline.MockFilterSpec(nil, nil, "", meta, nil)
+		cc.Init(filterSpec)
+		return
+	}
+	err := checkProtocol()
+	assert.NotNil(err, "if ConnectControl supports more protocol, please update this case")
+
+	spec := &Spec{
+		BannedClients: []string{"banClient1", "banClient2"},
+		BannedTopics:  []string{"banTopic1", "banTopic2"},
+	}
+	filterSpec := defaultFilterSpec(spec)
+	cc.Init(filterSpec)
+	newCc := &ConnectControl{}
+	newCc.Inherit(filterSpec, cc)
+	defer newCc.Close()
+	status := newCc.Status().(*Status)
+	assert.Equal(status.BannedClientNum, len(spec.BannedClients))
+	assert.Equal(status.BannedTopicNum, len(spec.BannedTopics))
+}
+
+type testCase struct {
+	cid        string
+	topic      string
+	err        error
+	disconnect bool
+	earlyStop  bool
+}
+
+func doTest(t *testing.T, spec *Spec, testCases []testCase) {
+	assert := assert.New(t)
+	filterSpec := defaultFilterSpec(spec)
+	cc := &ConnectControl{}
+	cc.Init(filterSpec)
+
+	for _, test := range testCases {
+		ctx := newContext(test.cid, test.topic)
+		res := cc.HandleMQTT(ctx)
+		assert.Equal(res.Err, test.err)
+		assert.Equal(ctx.Disconnect(), test.disconnect)
+		assert.Equal(ctx.EarlyStop(), test.earlyStop)
+	}
+	status := cc.Status().(*Status)
+	assert.Equal(status.BannedClientRe, spec.BannedClientRe)
+	assert.Equal(status.BannedTopicRe, spec.BannedTopicRe)
+	assert.Equal(status.BannedClientNum, len(spec.BannedClients))
+	assert.Equal(status.BannedTopicNum, len(spec.BannedTopics))
+}
+
+func TestHandleMQTT(t *testing.T) {
+	// check BannedClients
+	spec := &Spec{
+		BannedClients: []string{"ban1", "ban2"},
+		EarlyStop:     true,
+	}
+	testCases := []testCase{
+		{cid: "ban1", topic: "unban", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "ban2", topic: "unban", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "unban1", topic: "ban/sport/ball", err: nil, disconnect: false, earlyStop: false},
+		{cid: "unban2", topic: "ban/sport/run", err: nil, disconnect: false, earlyStop: false},
+		{cid: "unban", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+	}
+	doTest(t, spec, testCases)
+
+	// check BannedTopics
+	spec = &Spec{
+		BannedTopics: []string{"ban/sport/ball", "ban/sport/run"},
+		EarlyStop:    true,
+	}
+	testCases = []testCase{
+		{cid: "unban1", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "unban2", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "unban3", topic: "unban/sport", err: nil, disconnect: false, earlyStop: false},
+		{cid: "unban4", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+	}
+	doTest(t, spec, testCases)
+
+	// check BannedClientRe
+	spec = &Spec{
+		BannedClientRe: "phone",
+		EarlyStop:      true,
+	}
+	testCases = []testCase{
+		{cid: "phone123", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "phone256", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "tv", topic: "unban/sport", err: nil, disconnect: false, earlyStop: false},
+		{cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+	}
+	doTest(t, spec, testCases)
+
+	// check BannedTopicRe
+	spec = &Spec{
+		BannedTopicRe: "sport",
+		EarlyStop:     true,
+	}
+	testCases = []testCase{
+		{cid: "phone123", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "phone256", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
+		{cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+		{cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+	}
+	doTest(t, spec, testCases)
+
+	// check re compile fail
+	spec = &Spec{
+		BannedClientRe: "(",
+		BannedTopicRe:  "(?P<name>",
+	}
+	filterSpec := defaultFilterSpec(spec)
+	cc := &ConnectControl{}
+	cc.Init(filterSpec)
+	assert.Nil(t, cc.bannedClientRe)
+	assert.Nil(t, cc.bannedTopicRe)
+}
diff --git a/pkg/object/mqttproxy/mock_test.go b/pkg/object/mqttproxy/mock_test.go
index 605d5ebf..7d480cc8 100644
--- a/pkg/object/mqttproxy/mock_test.go
+++ b/pkg/object/mqttproxy/mock_test.go
@@ -30,12 +30,14 @@ import (
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/megaease/easegress/pkg/cluster"
 	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
 )
 
 type mockCluster struct {
 	sync.RWMutex
-	kv map[string]string
+	kv    map[string]string
+	delCh chan map[string]*string
 }
 
 var _ cluster.Cluster = (*mockCluster)(nil)
@@ -49,10 +51,8 @@ func (m *mockCluster) GetRawPrefix(prefix string) (map[string]*mvccpb.KeyValue,
 func (m *mockCluster) PutUnderLease(key, value string) error                      { return nil }
 func (m *mockCluster) PutAndDelete(map[string]*string) error                      { return nil }
 func (m *mockCluster) PutAndDeleteUnderLease(map[string]*string) error            { return nil }
-func (m *mockCluster) Delete(key string) error                                    { return nil }
 func (m *mockCluster) DeletePrefix(prefix string) error                           { return nil }
 func (m *mockCluster) STM(apply func(concurrency.STM) error) error                { return nil }
-func (m *mockCluster) Watcher() (cluster.Watcher, error)                          { return nil, nil }
 func (m *mockCluster) Syncer(pullInterval time.Duration) (*cluster.Syncer, error) { return nil, nil }
 func (m *mockCluster) Mutex(name string) (cluster.Mutex, error)                   { return nil, nil }
 func (m *mockCluster) CloseServer(wg *sync.WaitGroup)                             {}
@@ -60,6 +60,45 @@ func (m *mockCluster) StartServer() (chan struct{}, chan struct{}, error)
 func (m *mockCluster) Close(wg *sync.WaitGroup)                                   {}
 func (m *mockCluster) PurgeMember(member string) error                            { return nil }
 
+func (m *mockCluster) Watcher() (cluster.Watcher, error) {
+	m.Lock()
+	defer m.Unlock()
+	if m.delCh == nil {
+		m.delCh = make(chan map[string]*string, 100)
+	}
+	return &mockWatcher{delCh: m.delCh}, nil
+}
+
+func (m *mockCluster) Delete(key string) error {
+	m.Lock()
+	defer m.Unlock()
+	if v, ok := m.kv[key]; ok {
+		if m.delCh != nil {
+			kv := map[string]*string{key: &v}
+			m.delCh <- kv
+		}
+	}
+	delete(m.kv, key)
+	return nil
+}
+
+type mockWatcher struct {
+	delCh chan map[string]*string
+}
+
+var _ cluster.Watcher = (*mockWatcher)(nil)
+
+func (w *mockWatcher) Watch(key string) (<-chan *string, error)                     { return nil, nil }
+func (w *mockWatcher) WatchPrefix(prefix string) (<-chan map[string]*string, error) { return nil, nil }
+func (w *mockWatcher) WatchRaw(key string) (<-chan *clientv3.Event, error)          { return nil, nil }
+func (w *mockWatcher) WatchRawPrefix(prefix string) (<-chan map[string]*clientv3.Event, error) {
+	return nil, nil
+}
+func (w *mockWatcher) WatchWithOp(key string, ops ...cluster.ClientOp) (<-chan map[string]*string, error) {
+	return w.delCh, nil
+}
+func (w *mockWatcher) Close() {}
+
 func (m *mockCluster) Get(key string) (*string, error) {
 	m.RLock()
 	defer m.RUnlock()
@@ -81,6 +120,26 @@ func (m *mockCluster) GetPrefix(prefix string) (map[string]string, error) {
 	return out, nil
 }
 
+func (m *mockCluster) GetWithOp(key string, op ...cluster.ClientOp) (map[string]string, error) {
+	prefix := false
+	for _, o := range op {
+		if o == cluster.OpPrefix {
+			prefix = true
+		}
+	}
+	ans := make(map[string]string)
+	if prefix {
+		kvs, _ := m.GetPrefix(key)
+		for k, v := range kvs {
+			ans[k] = v
+		}
+	} else {
+		value, _ := m.Get(key)
+		ans[key] = *value
+	}
+	return ans, nil
+}
+
 func (m *mockCluster) Put(key, value string) error {
 	m.Lock()
 	m.kv[key] = value
@@ -101,10 +160,24 @@ func TestStorage(t *testing.T) {
 	if err != nil || *val != "1" {
 		t.Errorf("get wrong val")
 	}
-	valmap, err := store.getPrefix("prefix")
+	valmap, err := store.getPrefix("prefix", false)
 	if err != nil || !reflect.DeepEqual(valmap, map[string]string{"prefix_1": "1", "prefix_2": "2"}) {
 		t.Errorf("get wrong prefix val")
 	}
+	ch, _, err := store.watchDelete("prefix")
+	if err != nil {
+		t.Errorf("create watch delete failed %v", err)
+	}
+	store.delete("prefix_1")
+	delKv := <-ch
+	v, ok := delKv["prefix_1"]
+	if !ok {
+		t.Errorf("watch delete failed")
+	} else {
+		if *v != "1" {
+			t.Errorf("get wrong watch result, expected %v, got %v", "1", *v)
+		}
+	}
 }
 
 func TestMockStorage(t *testing.T) {
@@ -116,7 +189,7 @@ func TestMockStorage(t *testing.T) {
 	if err != nil || *val != "val1" {
 		t.Errorf("mock storage get return wrong value")
 	}
-	valMap, err := store.getPrefix("key")
+	valMap, err := store.getPrefix("key", false)
 	if err != nil || !reflect.DeepEqual(valMap, map[string]string{"key1": "val1", "key2": "val2", "key3": "val3"}) {
 		t.Errorf("mock storage get prefix return wrong value %v %v", valMap, map[string]string{"key1": "val1", "key2": "val2", "key3": "val3"})
 	}
@@ -141,7 +214,7 @@ var _ sarama.AsyncProducer = (*mockAsyncProducer)(nil)
 
 func newMockAsyncProducer() sarama.AsyncProducer {
 	return &mockAsyncProducer{
-		ch: make(chan *sarama.ProducerMessage, 1),
+		ch: make(chan *sarama.ProducerMessage, 100),
 	}
 }
 
@@ -152,7 +225,7 @@ func TestKafka(t *testing.T) {
 			Backend: []string{"localhost:1234"},
 		},
 	})
-	if k.(*KafkaMQ) != nil {
+	if k != nil {
 		t.Errorf("should return nil for invalid broker address, %v", k)
 	}
 	k = newBackendMQ(&Spec{
@@ -189,7 +262,13 @@ func TestKafka(t *testing.T) {
 	kafka.publish(p)
 	msg = <-kafka.producer.(*mockAsyncProducer).ch
 	if msg.Topic != p.TopicName || len(msg.Headers) != 0 {
-		t.Errorf("kafka producer produce wrong msg")
+		t.Errorf("kafka producer publish wrong msg")
+	}
+	kafka.Publish("a/b/c/d", []byte("abcd"), map[string]string{"key": "value"})
+	msg = <-kafka.producer.(*mockAsyncProducer).ch
+	if msg.Topic != "a/b/c/d" || len(msg.Headers) != 1 {
+		t.Errorf("kafka producer Publish wrong msg")
+
 	}
 	kafka.close()
 }
diff --git a/pkg/object/mqttproxy/mqtt_test.go b/pkg/object/mqttproxy/mqtt_test.go
index 343c9856..ca7ef049 100644
--- a/pkg/object/mqttproxy/mqtt_test.go
+++ b/pkg/object/mqttproxy/mqtt_test.go
@@ -36,6 +36,10 @@ import (
 	paho "github.com/eclipse/paho.mqtt.golang"
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/pipeline"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/openzipkin/zipkin-go/propagation/b3"
+	"github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v2"
 )
 
@@ -46,6 +50,8 @@ func (t *testMQ) get() *packets.PublishPacket {
 
 func init() {
 	logger.InitNop()
+	// logger.InitMock()
+	pipeline.Register(&pipeline.MockMQTTFilter{})
 }
 
 func getMQTTClient(t *testing.T, clientID, userName, password string, cleanSession bool) paho.Client {
@@ -57,6 +63,12 @@ func getMQTTClient(t *testing.T, clientID, userName, password string, cleanSessi
 	return c
 }
 
+func getUnConnectClient(clientID, userName, password string, cleanSession bool) paho.Client {
+	opts := paho.NewClientOptions().AddBroker("tcp://0.0.0.0:1883").SetClientID(clientID).SetUsername(userName).SetPassword(password).SetCleanSession(cleanSession)
+	c := paho.NewClient(opts)
+	return c
+}
+
 func getMQTTSubscribeHandler(ch chan CheckMsg) paho.MessageHandler {
 	handler := func(client paho.Client, message paho.Message) {
 		msg := CheckMsg{
@@ -716,7 +728,7 @@ func TestSendMsgBack(t *testing.T) {
 			for j := 0; j < msgNum; j++ {
 				topic := r.ClientID()
 				text := fmt.Sprintf("sub %d", j)
-				broker.sendMsgToClient(topic, []byte(text), QoS1)
+				broker.sendMsgToClient(nil, topic, []byte(text), QoS1)
 			}
 		}(clients[i])
 	}
@@ -905,7 +917,7 @@ func TestHTTPRequest(t *testing.T) {
 	broker := getBroker("test", "test", b64passwd, 1883)
 
 	srv := newServer(":8888")
-	srv.addHandlerFunc("/mqtt", broker.topicsPublishHandler)
+	srv.addHandlerFunc("/mqtt", broker.httpTopicsPublishHandler)
 	if err := srv.start(); err != nil {
 		t.Errorf("couldn't start server: %s", err)
 	}
@@ -990,7 +1002,7 @@ func TestHTTPPublish(t *testing.T) {
 	broker := getBroker("test", "test", b64passwd, 1883)
 
 	srv := newServer(":8888")
-	srv.addHandlerFunc("/mqtt", broker.topicsPublishHandler)
+	srv.addHandlerFunc("/mqtt", broker.httpTopicsPublishHandler)
 	srv.start()
 
 	subscribeCh := make(chan CheckMsg)
@@ -1049,7 +1061,7 @@ func TestHTTPTransfer(t *testing.T) {
 	passBase64 := base64.StdEncoding.EncodeToString([]byte("test"))
 	broker0 := getBroker("test", "test", passBase64, 1883)
 	srv0 := newServer(":8888")
-	srv0.addHandlerFunc("/mqtt", broker0.topicsPublishHandler)
+	srv0.addHandlerFunc("/mqtt", broker0.httpTopicsPublishHandler)
 	srv0.start()
 
 	spec := &Spec{
@@ -1076,7 +1088,7 @@ func TestHTTPTransfer(t *testing.T) {
 		return urls, nil
 	})
 	srv1 := newServer(":8889")
-	srv1.addHandlerFunc("/mqtt", broker1.topicsPublishHandler)
+	srv1.addHandlerFunc("/mqtt", broker1.httpTopicsPublishHandler)
 	srv1.start()
 
 	// auto connect to broker0, not broker1
@@ -1261,6 +1273,8 @@ func TestWildCard(t *testing.T) {
 	checkSubscriptions("+", []string{}, []string{"/finance"})
 }
 
+// this certPem and keyPem come from golang crypto/tls/testdata
+// with original name: example-key.pem and example-key.pem
 const certPem = `
 -----BEGIN CERTIFICATE-----
 MIIBhTCCASugAwIBAgIQIRi6zePL6mKjOipn+dNuaTAKBggqhkjOPQQDAjASMRAw
@@ -1336,17 +1350,20 @@ func TestSessMgr(t *testing.T) {
 }
 
 func TestClient(t *testing.T) {
+	assert := assert.New(t)
 	svcConn, clientConn := net.Pipe()
 	go func() {
 		io.ReadAll(svcConn)
 	}()
 	connect := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
+	connect.ClientIdentifier = "cid"
+	connect.Username = "username"
 	connect.WillFlag = true
 	connect.WillQos = 1
 	connect.WillTopic = "will"
 	connect.WillMessage = []byte("i am gone")
 
-	client := newClient(connect, nil, clientConn)
+	client := newClient(connect, nil, clientConn, nil)
 	will := client.info.will
 	if (will.Qos != connect.WillQos) || (will.TopicName != connect.WillTopic) || string(will.Payload) != string(connect.WillMessage) {
 		t.Error("produce wrong will msg")
@@ -1389,6 +1406,17 @@ func TestClient(t *testing.T) {
 	if err == nil {
 		t.Errorf("broker not ping")
 	}
+
+	assert.Equal("username", client.UserName())
+	client.Store("key", "value")
+
+	value, ok := client.Load("key")
+	assert.Equal(true, ok)
+	assert.Equal("value", value)
+
+	client.Delete("key")
+	_, ok = client.Load("key")
+	assert.Equal(false, ok)
 }
 
 func TestBrokerListen(t *testing.T) {
@@ -1459,7 +1487,7 @@ func TestBrokerListen(t *testing.T) {
 	if broker2 != nil {
 		t.Errorf("not valid port should return nil broker")
 	}
-	broker.mqttAPIPrefix()
+	broker.mqttAPIPrefix(mqttAPITopicPublishPrefix)
 	broker.registerAPIs()
 	broker.close()
 }
@@ -1497,5 +1525,502 @@ func TestMQTTProxy(t *testing.T) {
 	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
 	broker := getBroker("test", "test", b64passwd, 1883)
 	mp.broker = broker
+	broker.reconnectWatcher()
 	mp.Close()
+
+	ans, err := updatePort("http://example.com:1234", "demo.com:2345")
+	assert.Nil(t, err)
+	assert.Equal(t, "http://example.com:2345", ans)
+}
+
+func TestPipeline(t *testing.T) {
+	// create test pipeline first
+	yamlStr := `
+name: mqtt-test-pipeline
+kind: Pipeline
+protocol: MQTT
+flow:
+- filter: mqtt-filter
+filters:
+- name: mqtt-filter
+  kind: MockMQTTFilter
+  userName: test
+  port: 1234
+  backendType: Kafka
+  keysToStore:
+  - filter
+  publishBackendClientID: true
+`
+
+	super := supervisor.NewDefaultMock()
+	superSpec, err := super.NewSpec(yamlStr)
+	if err != nil {
+		t.Errorf("supervisor unmarshal yaml failed, %s", err)
+		t.Skip()
+	}
+	pipe := pipeline.Pipeline{}
+	pipe.Init(superSpec)
+	defer pipe.Close()
+
+	// get broker
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	broker := getBroker("test", "test", b64passwd, 1883)
+	broker.pipeline = "mqtt-test-pipeline"
+	defer broker.close()
+
+	// set some client
+	clientNum := 20
+	for i := 0; i < clientNum; i++ {
+		client := getMQTTClient(t, strconv.Itoa(i), "test", "test", true)
+		topic := fmt.Sprintf("client-%d", i)
+		text := "text"
+		token := client.Publish(topic, 1, false, text)
+		token.Wait()
+		p := broker.backend.(*testMQ).get()
+		if p.TopicName != topic || string(p.Payload) != text {
+			t.Errorf("get wrong publish")
+		}
+		c := broker.getClient(strconv.Itoa(i))
+		if _, ok := c.Load("filter"); !ok {
+			t.Errorf("filter write key value failed")
+		}
+		client.Disconnect(200)
+	}
+	// wait all close
+	for i := 0; i < 10; i++ {
+		num := len(broker.currentClients())
+		if num == 0 {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	backend := broker.backend.(*testMQ)
+	if len(backend.msg) != clientNum {
+		t.Errorf("filter publish message to backend failed")
+	}
+
+	filterStatus := pipe.Status().ObjectStatus.(*pipeline.Status).Filters["mqtt-filter"].(pipeline.MockMQTTStatus)
+	if len(filterStatus.ClientCount) != clientNum {
+		t.Errorf("filter get wrong result %v for client num", len(filterStatus.ClientCount))
+	}
+	if len(filterStatus.ClientDisconnect) != clientNum {
+		t.Errorf("filter get wrong result %v for client disconnect", filterStatus.ClientDisconnect)
+	}
+}
+
+func TestAuthByPipeline(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	broker := getBroker("test", "test", b64passwd, 1883)
+	broker.spec.AuthByPipeline = true
+	broker.pipeline = "mqtt-test-pipeline"
+	defer broker.close()
+
+	// since there are no such pipeline
+	// connect to broker will fail
+	option := paho.NewClientOptions().AddBroker("tcp://0.0.0.0:1883").SetClientID("test").SetUsername("fakeuser").SetPassword("fakepasswd")
+	client := paho.NewClient(option)
+	if token := client.Connect(); token.Wait() && token.Error() == nil {
+		t.Errorf("non auth user should fail%s", token.Error())
+	}
+	client.Disconnect(200)
+
+	// filter with auth username and passwd
+	yamlStr := `
+name: mqtt-test-pipeline
+kind: Pipeline
+protocol: MQTT
+flow:
+- filter: mqtt-filter
+filters:
+- name: mqtt-filter
+  kind: MockMQTTFilter
+  userName: filter-auth-name
+  password: filter-auth-passwd
+  connectKey: connect`
+
+	super := supervisor.NewDefaultMock()
+	superSpec, err := super.NewSpec(yamlStr)
+	if err != nil {
+		t.Errorf("supervisor unmarshal yaml failed, %s", err)
+		t.Skip()
+	}
+	pipe := pipeline.Pipeline{}
+	pipe.Init(superSpec)
+	defer pipe.Close()
+
+	// same username and passwd with filter, success
+	option = paho.NewClientOptions().AddBroker("tcp://0.0.0.0:1883").SetClientID("test").SetUsername("filter-auth-name").SetPassword("filter-auth-passwd")
+	client = paho.NewClient(option)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		t.Errorf("auth user should success %s", token.Error())
+	}
+	c := broker.getClient("test")
+	if _, ok := c.Load("connect"); !ok {
+		t.Errorf("filter set connect key failed")
+	}
+
+	client.Disconnect(200)
+
+	// different username and passwd with filer, fail
+	option = paho.NewClientOptions().AddBroker("tcp://0.0.0.0:1883").SetClientID("test").SetUsername("fake").SetPassword("fakepasswd")
+	client = paho.NewClient(option)
+	if token := client.Connect(); token.Wait() && token.Error() == nil {
+		t.Errorf("non auth user should fail %s", token.Error())
+	}
+	client.Disconnect(200)
+}
+
+func TestEmptyAuthAndAuthByPipeline(t *testing.T) {
+	spec := &Spec{
+		Name:           "test",
+		EGName:         "test",
+		Port:           1883,
+		BackendType:    testMQType,
+		AuthByPipeline: true,
+	}
+	store := newStorage(nil)
+	broker := newBroker(spec, store, func(s, ss string) ([]string, error) {
+		m := map[string]string{
+			"test":  "http://localhost:8888/mqtt",
+			"test1": "http://localhost:8889/mqtt",
+		}
+		urls := []string{}
+		for k, v := range m {
+			if k != s {
+				urls = append(urls, v)
+			}
+		}
+		return urls, nil
+	})
+	if broker == nil {
+		t.Errorf("broker with empty auth should not be nil when AuthByPipeline is true")
+	}
+	broker.close()
+
+	spec = &Spec{
+		Name:           "test",
+		EGName:         "test",
+		Port:           1883,
+		BackendType:    testMQType,
+		AuthByPipeline: false,
+	}
+	store = newStorage(nil)
+	broker = newBroker(spec, store, func(s, ss string) ([]string, error) {
+		m := map[string]string{
+			"test":  "http://localhost:8888/mqtt",
+			"test1": "http://localhost:8889/mqtt",
+		}
+		urls := []string{}
+		for k, v := range m {
+			if k != s {
+				urls = append(urls, v)
+			}
+		}
+		return urls, nil
+	})
+	if broker != nil {
+		t.Errorf("broker with empty auth should be nil when AuthByPipeline is false")
+	}
+}
+
+func TestMaxAllowedConnection(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	spec := &Spec{
+		Name:        "test",
+		EGName:      "test",
+		Port:        1883,
+		BackendType: testMQType,
+		Auth: []Auth{
+			{UserName: "test", PassBase64: b64passwd},
+		},
+		MaxAllowedConnection: 10,
+	}
+	store := newStorage(nil)
+	broker := newBroker(spec, store, func(s, ss string) ([]string, error) {
+		return nil, errors.New("empty urls for test")
+	})
+	defer broker.close()
+
+	clients := []paho.Client{}
+	clientNum := 10
+	for i := 0; i < clientNum; i++ {
+		client := getMQTTClient(t, strconv.Itoa(i), "test", "test", true)
+		clients = append(clients, client)
+	}
+	var num int
+	for i := 0; i < 10; i++ {
+		num = len(broker.currentClients())
+		if num == clientNum {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	if num != clientNum {
+		t.Fatalf("wrong client connection number, got %v, expected %v", num, clientNum)
+	}
+	for i := clientNum; i < 2*clientNum; i++ {
+		client := getUnConnectClient(strconv.Itoa(i), "test", "test", true)
+		if token := client.Connect(); token.Wait() && token.Error() == nil {
+			t.Errorf("client %v connect should fail but got nil error, %v", i, token.Error())
+		}
+	}
+	for _, c := range clients {
+		c.Disconnect(200)
+	}
+}
+
+func TestConnectionLimit(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	spec := &Spec{
+		Name:        "test",
+		EGName:      "test",
+		Port:        1883,
+		BackendType: testMQType,
+		Auth: []Auth{
+			{UserName: "test", PassBase64: b64passwd},
+		},
+		ConnectionLimit: &RateLimit{
+			RequestRate: 10,
+			TimePeriod:  1000,
+		},
+	}
+	store := newStorage(nil)
+	broker := newBroker(spec, store, func(s, ss string) ([]string, error) {
+		return nil, errors.New("empty urls for test")
+	})
+	defer broker.close()
+
+	// use all rate
+	for i := 0; i < 10; i++ {
+		broker.connectionLimiter.acquirePermission(1000)
+	}
+	client := getUnConnectClient("test", "test", "test", true)
+	if token := client.Connect(); token.Wait() && token.Error() == nil {
+		t.Errorf("client test connect should fail but got nil error, %v", token.Error())
+	}
+}
+
+func TestClientPublishLimit(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	spec := &Spec{
+		Name:        "test",
+		EGName:      "test",
+		Port:        1883,
+		BackendType: testMQType,
+		Auth: []Auth{
+			{UserName: "test", PassBase64: b64passwd},
+		},
+		ClientPublishLimit: &RateLimit{
+			RequestRate: 10,
+			TimePeriod:  1000,
+		},
+	}
+	store := newStorage(nil)
+	broker := newBroker(spec, store, func(s, ss string) ([]string, error) {
+		return nil, errors.New("empty urls for test")
+	})
+	defer broker.close()
+
+	client := getMQTTClient(t, "test", "test", "test", true)
+	time.Sleep(50 * time.Millisecond)
+
+	// acquire all permission
+	c := broker.getClient("test")
+	for i := 0; i < 10; i++ {
+		c.publishLimit.acquirePermission(100)
+	}
+
+	token := client.Publish("123", 1, false, []byte("test"))
+	if token.WaitTimeout(1 * time.Second) {
+		t.Errorf("client publish should fail, since we set client publish limit")
+	}
+}
+
+func TestHTTPGetAllSession(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	broker := getBroker("test", "test", b64passwd, 1883)
+	defer broker.close()
+
+	// connect 10 clients
+	clients := []paho.Client{}
+	clientNum := 10
+	for i := 0; i < clientNum; i++ {
+		client := getMQTTClient(t, strconv.Itoa(i), "test", "test", true)
+		if token := client.Subscribe("topic", 1, nil); token.Wait() && token.Error() != nil {
+			t.Errorf("subscribe qos0 error %s", token.Error())
+		}
+		clients = append(clients, client)
+	}
+
+	// start server
+	srv := newServer(":8888")
+	srv.addHandlerFunc("/session/query", broker.httpGetAllSessionHandler)
+	if err := srv.start(); err != nil {
+		t.Errorf("couldn't start server: %s", err)
+	}
+	defer srv.shutdown()
+
+	tests := []struct {
+		url    string
+		ok     bool
+		ansLen int
+	}{
+		{"http://localhost:8888/session/query?page=1&page_size=20&q=", true, 10},
+		{"http://localhost:8888/session/query?page=1&page_size=8&q=", true, 8},
+		{"http://localhost:8888/session/query", true, 10},
+		{"http://localhost:8888/session/query?page=1", false, 0},
+		{"http://localhost:8888/session/query?page_size=2", false, 0},
+		{"http://localhost:8888/session/query?q=", false, 0},
+		{"http://localhost:8888/session/query?page=0&page_size=10&q=", false, 0},
+		{"http://localhost:8888/session/query?page=1&page_size=0&q=", false, 0},
+		{"http://localhost:8888/session/query?page=2&page_size=20&q=", true, 0},
+		{"http://localhost:8888/session/query?page=1&page_size=10&q=233", true, 0},
+	}
+	for _, test := range tests {
+		req, err := http.NewRequest(http.MethodGet, test.url, nil)
+		if err != nil {
+			t.Errorf("get request failed, %v", err)
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Errorf("client request failed, %v", err)
+		}
+		ok := resp.StatusCode == http.StatusOK
+		if ok != test.ok {
+			t.Errorf("get wrong result")
+		}
+		if ok {
+			sessions := &HTTPSessions{}
+			json.NewDecoder(resp.Body).Decode(sessions)
+			if len(sessions.Sessions) != test.ansLen {
+				t.Errorf("get wrong session number wanted %v, got %v", test.ansLen, len(sessions.Sessions))
+			}
+		}
+		resp.Body.Close()
+	}
+	for _, c := range clients {
+		c.Disconnect(200)
+	}
+}
+
+func TestHTTPDeleteSession(t *testing.T) {
+	b64passwd := base64.StdEncoding.EncodeToString([]byte("test"))
+	broker := getBroker("test", "test", b64passwd, 1883)
+	defer broker.close()
+
+	// connect 10 clients
+	clients := []paho.Client{}
+	clientNum := 10
+	for i := 0; i < clientNum; i++ {
+		client := getMQTTClient(t, strconv.Itoa(i), "test", "test", true)
+		clients = append(clients, client)
+	}
+
+	// start server
+	srv := newServer(":8888")
+	srv.addHandlerFunc("/sessions", broker.httpDeleteSessionHandler)
+	if err := srv.start(); err != nil {
+		t.Errorf("couldn't start server: %s", err)
+	}
+	defer srv.shutdown()
+
+	data := HTTPSessions{
+		Sessions: []*HTTPSession{
+			{SessionID: "1"},
+			{SessionID: "2"},
+		},
+	}
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		t.Errorf("marshal http session %v failed, %v", data, err)
+	}
+	req, _ := http.NewRequest(http.MethodDelete, "http://localhost:8888/sessions", bytes.NewBuffer(jsonData))
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("delete session failed, %v", err)
+	}
+	defer resp.Body.Close()
+
+	for i := 0; i < 10; i++ {
+		if len(broker.currentClients()) == clientNum-2 {
+			break
+		}
+		if i == 9 {
+			t.Errorf("session delete failed %v", broker.currentClients())
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	for _, c := range clients {
+		c.Disconnect(200)
+	}
+}
+
+func TestHTTPTransferHeaderCopy(t *testing.T) {
+	done := make(chan bool, 2)
+
+	passBase64 := base64.StdEncoding.EncodeToString([]byte("test"))
+	broker0 := getBroker("test", "test", passBase64, 1883)
+	srv0 := newServer(":8888")
+	srv0.addHandlerFunc("/mqtt", func(w http.ResponseWriter, r *http.Request) {
+		broker0.httpTopicsPublishHandler(w, r)
+		done <- true
+	})
+	srv0.start()
+
+	spec := &Spec{
+		Name:        "test1",
+		EGName:      "test1",
+		Port:        1884,
+		BackendType: testMQType,
+		Auth: []Auth{
+			{UserName: "test", PassBase64: passBase64},
+		},
+	}
+	store := broker0.sessMgr.store
+	broker1 := newBroker(spec, store, func(s, ss string) ([]string, error) {
+		m := map[string]string{
+			"test":  "http://localhost:8888/mqtt",
+			"test1": "http://localhost:8889/mqtt",
+		}
+		urls := []string{}
+		for k, v := range m {
+			if k != s {
+				urls = append(urls, v)
+			}
+		}
+		return urls, nil
+	})
+	srv1 := newServer(":8889")
+	srv1.addHandlerFunc("/mqtt", func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get(b3.TraceID) != "123" {
+			t.Errorf("wrong trace id received")
+		}
+		done <- true
+	})
+	srv1.start()
+
+	// set data to broker1
+	data := HTTPJsonData{
+		Topic:   "client",
+		QoS:     1,
+		Payload: "data",
+		Base64:  false,
+	}
+	jsonData, _ := json.Marshal(data)
+	req, _ := http.NewRequest(http.MethodPost, "http://localhost:8888/mqtt", bytes.NewBuffer(jsonData))
+	req.Header.Add(b3.TraceID, "123")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Errorf("mqtt client do error")
+	}
+	defer resp.Body.Close()
+
+	<-done
+	<-done
+	broker0.close()
+	broker1.close()
+	srv0.shutdown()
+	srv1.shutdown()
 }
diff --git a/pkg/object/mqttproxy/ratelimiter_test.go b/pkg/object/mqttproxy/ratelimiter_test.go
new file mode 100644
index 00000000..73d8e5da
--- /dev/null
+++ b/pkg/object/mqttproxy/ratelimiter_test.go
@@ -0,0 +1,126 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mqttproxy
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLimiter(t *testing.T) {
+	// default spec will not block anything
+	spec := &RateLimit{}
+	l := newLimiter(spec)
+	for i := 0; i < 1000; i++ {
+		permitted := l.acquirePermission(100)
+		assert.Equal(t, true, permitted)
+	}
+
+	// only provide rate should only check rate
+	spec = &RateLimit{
+		RequestRate: 10,
+		TimePeriod:  60,
+	}
+	l = newLimiter(spec)
+	for i := 0; i < 20; i++ {
+		permitted := l.acquirePermission(1000)
+		if i < 10 {
+			assert.Equal(t, true, permitted)
+		} else {
+			assert.Equal(t, false, permitted)
+		}
+	}
+
+	// only provide burst should only check burst
+	spec = &RateLimit{
+		BytesRate:  100,
+		TimePeriod: 60,
+	}
+	l = newLimiter(spec)
+	for i := 0; i < 20; i++ {
+		permitted := l.acquirePermission(10)
+		if i < 10 {
+			assert.Equal(t, true, permitted)
+		} else {
+			assert.Equal(t, false, permitted)
+		}
+	}
+
+	// not provide time period use one second
+	spec = &RateLimit{
+		BytesRate: 100,
+	}
+	l = newLimiter(spec)
+	permitted := l.acquirePermission(100)
+	assert.Equal(t, true, permitted)
+	permitted = l.acquirePermission(1)
+	assert.Equal(t, false, permitted)
+	time.Sleep(1100 * time.Millisecond)
+	permitted = l.acquirePermission(100)
+	assert.Equal(t, true, permitted)
+
+	// provide both and use both
+	spec = &RateLimit{
+		RequestRate: 10,
+		BytesRate:   100,
+		TimePeriod:  60,
+	}
+	l = newLimiter(spec)
+	for i := 0; i < 20; i++ {
+		permitted := l.acquirePermission(10)
+		if i < 10 {
+			assert.Equal(t, true, permitted)
+		} else {
+			assert.Equal(t, false, permitted)
+		}
+	}
+
+	// provide both and use both
+	spec = &RateLimit{
+		RequestRate: 10,
+		BytesRate:   100,
+		TimePeriod:  60,
+	}
+	l = newLimiter(spec)
+	for i := 0; i < 20; i++ {
+		permitted := l.acquirePermission(0)
+		if i < 10 {
+			assert.Equal(t, true, permitted)
+		} else {
+			assert.Equal(t, false, permitted)
+		}
+	}
+
+	// provide both and use both
+	spec = &RateLimit{
+		RequestRate: 100,
+		BytesRate:   10,
+		TimePeriod:  60,
+	}
+	l = newLimiter(spec)
+	for i := 0; i < 20; i++ {
+		permitted := l.acquirePermission(1)
+		if i < 10 {
+			assert.Equal(t, true, permitted)
+		} else {
+			assert.Equal(t, false, permitted)
+		}
+	}
+}
diff --git a/pkg/object/pipeline/mock_test.go b/pkg/object/pipeline/mock_test.go
new file mode 100644
index 00000000..e451db8b
--- /dev/null
+++ b/pkg/object/pipeline/mock_test.go
@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMockFilter(t *testing.T) {
+	assert := assert.New(t)
+
+	// mockFilter do nothing, used for create other filters
+	mock := &mockFilter{}
+	assert.Equal(mock.Kind(), "MockFilter")
+	assert.Equal(mock.DefaultSpec(), &mockSpec{})
+	assert.Equal(mock.Description(), "mock filter")
+	assert.Nil(mock.Results())
+	assert.Nil(mock.Status())
+	assert.Nil(mock.APIs())
+	mock.Init(nil)
+	newMock := &mockFilter{}
+	newMock.Inherit(nil, mock)
+	newMock.Close()
+}
+
+func TestMockFilterSpec(t *testing.T) {
+	assert := assert.New(t)
+
+	super := supervisor.NewDefaultMock()
+	rawSpec := map[string]interface{}{"name": "testMock"}
+	yamlConfig := "fake yaml config string"
+	meta := &FilterMetaSpec{
+		Name:     "testMock",
+		Kind:     "test",
+		Pipeline: "pipeline-demo",
+		Protocol: context.TCP,
+	}
+	filterSpec := &MockMQTTSpec{}
+	mockSpec := MockFilterSpec(super, rawSpec, yamlConfig, meta, filterSpec)
+	assert.Equal(mockSpec.Super(), super)
+	assert.Equal(mockSpec.RawSpec(), rawSpec)
+	assert.Equal(mockSpec.Kind(), meta.Kind)
+	assert.Equal(mockSpec.Name(), meta.Name)
+	assert.Equal(mockSpec.Pipeline(), meta.Pipeline)
+	assert.Equal(mockSpec.Protocol(), meta.Protocol)
+	assert.Equal(mockSpec.FilterSpec(), filterSpec)
+}
diff --git a/pkg/object/pipeline/pipeline_test.go b/pkg/object/pipeline/pipeline_test.go
new file mode 100644
index 00000000..0cd02ec7
--- /dev/null
+++ b/pkg/object/pipeline/pipeline_test.go
@@ -0,0 +1,198 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	stdcontext "context"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/stretchr/testify/assert"
+)
+
+func init() {
+	Register(&MockMQTTFilter{})
+	logger.InitNop()
+}
+
+func getPipeline(yamlStr string, t *testing.T) *Pipeline {
+	super := supervisor.NewDefaultMock()
+	superSpec, err := super.NewSpec(yamlStr)
+	if err != nil {
+		t.Errorf("supervisor unmarshal yaml failed, %s", err)
+		t.Skip()
+	}
+	p := &Pipeline{}
+	p.Init(superSpec)
+	return p
+}
+
+func TestPipeline(t *testing.T) {
+	assert := assert.New(t)
+
+	yamlStr := `
+    name: pipeline
+    kind: Pipeline
+    protocol: MQTT
+    flow:
+    - filter: mqtt-filter
+    - filter: mqtt-filter2
+    filters:
+    - name: mqtt-filter
+      kind: MockMQTTFilter
+      userName: test
+      port: 1234
+      backendType: Kafka
+    - name: mqtt-filter2
+      kind: MockMQTTFilter`
+	p := getPipeline(yamlStr, t)
+
+	assert.Equal(p.spec.Name, "pipeline", "wrong name")
+	assert.Equal(p.spec.Protocol, context.MQTT, "wrong protocol")
+	assert.Equal(len(p.spec.Flow), 2, "wrong flow length")
+	assert.Equal(p.spec.Flow[0].Filter, "mqtt-filter", "wrong filter name")
+	assert.Equal(len(p.runningFilters), 2, "wrong running filters")
+
+	s := p.runningFilters[0].spec
+	assert.Equal(s.Name(), "mqtt-filter", "wrong filter name")
+	assert.Equal(s.Kind(), "MockMQTTFilter", "wrong filter kind")
+	assert.Equal(s.Pipeline(), "pipeline", "wrong filter pipeline")
+	assert.Equal(s.Protocol(), context.MQTT, "wrong filter protocol")
+
+	f := p.getRunningFilter("mqtt-filter").filter.(*MockMQTTFilter)
+	assert.Equal(f.spec.UserName, "test", "wrong filter username")
+	assert.Equal(f.spec.Port, uint16(1234), "wrong filter port")
+	assert.Equal(f.spec.BackendType, "Kafka", "wrong filter BackendType")
+
+	f = p.getRunningFilter("mqtt-filter2").filter.(*MockMQTTFilter)
+	assert.Equal(f.spec.UserName, "", "wrong filter username")
+	assert.Equal(f.spec.Port, uint16(0), "wrong filter port")
+	assert.Equal(f.spec.BackendType, "", "wrong filter BackendType")
+
+	pipeline, err := GetPipeline("pipeline", context.MQTT)
+	assert.Nil(err, "get pipeline failed")
+	assert.Equal(pipeline, p, "get wrong pipeline")
+
+	status := p.Status().ObjectStatus.(*Status)
+	assert.Equal(len(status.Filters), 2)
+
+	filter := p.getRunningFilter("not-exist-filter")
+	assert.Nil(filter)
+
+	newP := &Pipeline{}
+	newP.Inherit(p.superSpec, p)
+	newP.Close()
+}
+
+func TestHandleMQTT(t *testing.T) {
+	assert := assert.New(t)
+
+	yamlStr := `
+name: pipeline
+kind: Pipeline
+protocol: MQTT
+flow:
+- filter: mqtt-filter
+filters:
+- name: mqtt-filter
+  kind: MockMQTTFilter
+  userName: test
+  port: 1234
+  earlyStop: true
+  backendType: Kafka
+  keysToStore:
+  - mock
+  - mqtt
+  - filter
+  publishBackendClientID: true`
+	p := getPipeline(yamlStr, t)
+	defer p.Close()
+
+	backend := &context.MockMQTTBackend{
+		Messages: make(map[string]context.MockMQTTMsg),
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		go func(i int) {
+			c := &mockMQTTClient{cid: strconv.Itoa(i), userName: strconv.Itoa(i + 1)}
+			publish := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
+			ctx := context.NewMQTTContext(stdcontext.Background(), backend, c, publish)
+			assert.Equal(ctx.Client().UserName(), strconv.Itoa(i+1))
+			p.HandleMQTT(ctx)
+			_, ok := ctx.Client().Load("mock")
+			assert.Equal(true, ok)
+			_, ok = ctx.Client().Load("mqtt")
+			assert.Equal(true, ok)
+			_, ok = ctx.Client().Load("filter")
+			assert.Equal(true, ok)
+			wg.Done()
+		}(i)
+	}
+	wg.Wait()
+	f := p.getRunningFilter("mqtt-filter").filter.(*MockMQTTFilter)
+	assert.Equal(len(f.Status().(MockMQTTStatus).ClientCount), 1000, "wrong client count")
+
+	assert.Equal(1000, len(backend.Messages))
+
+	newP := &Pipeline{}
+	newP.spec = &Spec{Protocol: context.HTTP}
+	assert.NotPanics(func() { newP.HandleMQTT(nil) }, "handle mqtt will log and return since pipeline protocol is http")
+
+	yamlStr = `
+    name: pipeline-no-flow
+    kind: Pipeline
+    protocol: MQTT
+    filters:
+    - name: mqtt-filter
+      kind: MockMQTTFilter
+      userName: test
+      port: 1234
+      earlyStop: true
+      backendType: Kafka`
+	assert.NotPanics(func() { getPipeline(yamlStr, t) }, "no flow should work")
+
+	yamlStr = `
+    name: pipeline-flow-no-filter
+    kind: Pipeline
+    protocol: MQTT
+    flow:
+    - filter: mqtt-filter
+    filters:
+    - name: http-filter
+      kind: MockMQTTFilter
+      userName: test
+      port: 1234
+      earlyStop: true
+      backendType: Kafka`
+	assert.Panics(func() { getPipeline(yamlStr, t) }, "flow and filter have different name")
+
+	yamlStr = `
+    name: pipeline-flow-no-filter
+    kind: Pipeline
+    protocol: MQTT
+    flow:
+    - filter: mqtt-filter`
+	assert.Panics(func() { getPipeline(yamlStr, t) }, "flow and no filter should panic")
+}
diff --git a/pkg/object/pipeline/registry_test.go b/pkg/object/pipeline/registry_test.go
new file mode 100644
index 00000000..feed22d0
--- /dev/null
+++ b/pkg/object/pipeline/registry_test.go
@@ -0,0 +1,143 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pipeline
+
+import (
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/stretchr/testify/assert"
+)
+
+// allProtocolFilter can handle http, mqtt, tcp
+type allProtocolFilter struct {
+	mockFilter
+}
+
+var _ HTTPFilter = (*allProtocolFilter)(nil)
+var _ MQTTFilter = (*allProtocolFilter)(nil)
+var _ TCPFilter = (*allProtocolFilter)(nil)
+
+func (f *allProtocolFilter) Kind() string {
+	return "allProtocolFilter"
+}
+
+func (f *allProtocolFilter) HandleHTTP(ctx context.HTTPContext) *context.HTTPResult {
+	return nil
+}
+
+func (f *allProtocolFilter) HandleMQTT(ctx context.MQTTContext) *context.MQTTResult {
+	return nil
+}
+
+func (f *allProtocolFilter) HandleTCP(ctx context.TCPContext) *context.TCPResult {
+	return nil
+}
+
+func (f *allProtocolFilter) Results() []string {
+	return []string{"protocolError", "contextError"}
+}
+
+// errResultFilter return repeat result
+type errResultFilter struct {
+	mockFilter
+}
+
+func (f *errResultFilter) Kind() string      { return "errResultFilter" }
+func (f *errResultFilter) Results() []string { return []string{"error", "error"} }
+
+// emptyKindFilter kind is empty
+type emptyKindFilter struct {
+	mockFilter
+}
+
+func (f *emptyKindFilter) Kind() string { return "" }
+
+// noPtrFilter is filter but not a pointer
+type nonPtrFilter struct {
+}
+
+type noPtrSpec struct {
+}
+
+var _ Filter = nonPtrFilter{}
+
+func (f nonPtrFilter) Kind() string                                              { return "nonPtrFilter" }
+func (f nonPtrFilter) DefaultSpec() interface{}                                  { return &noPtrSpec{} }
+func (f nonPtrFilter) Description() string                                       { return "nonPtrFilter" }
+func (f nonPtrFilter) Results() []string                                         { return nil }
+func (f nonPtrFilter) Init(filterSpec *FilterSpec)                               {}
+func (f nonPtrFilter) Inherit(filterSpec *FilterSpec, previousGeneration Filter) {}
+func (f nonPtrFilter) Status() interface{}                                       { return nil }
+func (f nonPtrFilter) Close()                                                    {}
+func (f nonPtrFilter) APIs() []*APIEntry                                         { return nil }
+
+// intSpecFilter return int default spec
+type intSpecFilter struct {
+	mockFilter
+}
+
+func (f *intSpecFilter) Kind() string             { return "nilSpecFilter" }
+func (f *intSpecFilter) DefaultSpec() interface{} { return 2 }
+
+// nonStructSpecFilter
+type nonStructSpecFilter struct {
+	mockFilter
+}
+
+func (f *nonStructSpecFilter) Kind() string             { return "nonStructSpecFilter" }
+func (f *nonStructSpecFilter) DefaultSpec() interface{} { return map[string]string{} }
+
+func TestRegistry(t *testing.T) {
+	assert := assert.New(t)
+
+	// test getProtocols
+	allFilter := &allProtocolFilter{}
+	protocols, err := getProtocols(allFilter)
+	assert.Nil(err)
+	assert.Equal(protocols, map[context.Protocol]struct{}{context.HTTP: {}, context.MQTT: {}, context.TCP: {}}, "allProtocolFilter support HTTP, MQTT and TCP")
+
+	nonFilter := &mockFilter{}
+	_, err = getProtocols(nonFilter)
+	assert.NotNil(err, "nonFilter not support any protocol")
+
+	// test GetPipeline
+	_, err = GetPipeline("pipeline-no-exist", context.HTTP)
+	assert.NotNil(err, "GetPipeline not exist pipeline should fail")
+
+	// test store and delete pipeline
+	assert.Panics(func() { deletePipeline("not-exist", context.HTTP) }, "delete not-exist pipeline should panic")
+	assert.NotPanics(func() { storePipeline("pipeline-store", context.HTTP, &Pipeline{}) }, "store pipeline should success")
+	assert.Panics(func() { storePipeline("pipeline-store", context.HTTP, &Pipeline{}) }, "store exist pipeline should panic")
+	assert.NotPanics(func() { storePipeline("pipeline-store", context.MQTT, &Pipeline{}) }, "store same pipeline name but different protocol name should success")
+	assert.NotPanics(func() { deletePipeline("pipeline-store", context.HTTP) }, "delete exist pipeline should success")
+
+	Register(&mockFilter{})
+	Register(&allProtocolFilter{})
+	filters := GetFilterRegistry()
+	assert.Contains(filters, nonFilter.Kind())
+	assert.Contains(filters, allFilter.Kind())
+
+	// check register
+	assert.Panics(func() { Register(&errResultFilter{}) }, "errResultFilter return repeat results, should panic")
+	assert.Panics(func() { Register(&emptyKindFilter{}) }, "emptyKindFilter return empty kind, should panic")
+	assert.Panics(func() { Register(&mockFilter{}) }, "mockFilter already registered, should panic")
+	assert.Panics(func() { Register(nonPtrFilter{}) }, "nonPtrFilter is not a ptr should panic")
+	assert.Panics(func() { Register(&intSpecFilter{}) }, "intSpecFilter return int default spec, should panic")
+	assert.Panics(func() { Register(&nonStructSpecFilter{}) }, "nonStructSpecFilter return non struct default spec, should panic")
+}
diff --git a/pkg/util/ratelimiter/multiratelimiter_test.go b/pkg/util/ratelimiter/multiratelimiter_test.go
new file mode 100644
index 00000000..14688261
--- /dev/null
+++ b/pkg/util/ratelimiter/multiratelimiter_test.go
@@ -0,0 +1,371 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ratelimiter
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestMultiConcurrent(t *testing.T) {
+	policy := NewMultiPolicy(50*time.Millisecond, 10*time.Millisecond, []int{5})
+
+	var wg sync.WaitGroup
+	limiter := NewMulti(policy)
+	waitPermission := func() bool {
+		permitted, err := limiter.WaitPermission([]int{1})
+		if err != nil {
+			panic(err)
+		}
+		return permitted
+	}
+
+	fn := func() {
+		permitted := waitPermission()
+		if !permitted {
+			t.Errorf("WaitPermission should succeed")
+		}
+		wg.Done()
+	}
+
+	wg.Add(30)
+	for i := 0; i < 30; i++ {
+		go fn()
+	}
+	wg.Wait()
+
+	if waitPermission() {
+		t.Errorf("WaitPermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 5)
+	if waitPermission() {
+		t.Errorf("WaitPermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 5)
+	wg.Add(5)
+	for i := 0; i < 5; i++ {
+		go fn()
+	}
+	wg.Wait()
+
+	if waitPermission() {
+		t.Errorf("WaitPermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 100)
+	wg.Add(30)
+	for i := 0; i < 30; i++ {
+		go fn()
+	}
+	wg.Wait()
+
+	if waitPermission() {
+		t.Errorf("WaitPermission should fail")
+	}
+}
+
+func TestMultiRateLimiter(t *testing.T) {
+	policy := NewMultiPolicy(50*time.Millisecond, 10*time.Millisecond, []int{5})
+
+	limiter := NewMulti(policy)
+	limiter.SetState(StateNormal)
+	for i := 0; i < 30; i++ {
+		permitted, d, err := limiter.AcquirePermission([]int{1})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+		if d != time.Duration(i/policy.LimitForPeriod[0])*policy.LimitRefreshPeriod {
+			t.Errorf("wait duration of %d should not be: %s", i, d.String())
+		}
+	}
+
+	limiter.SetState(StateDisabled)
+	permitted, d, err := limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if !permitted {
+		t.Errorf("AcquirePermission should succeeded")
+	} else if d != 0 {
+		t.Errorf("wait duration should not be: %s", d.String())
+	}
+
+	limiter.SetState(StateNormal)
+	for i := 0; i < 30; i++ {
+		permitted, d, err := limiter.AcquirePermission([]int{1})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+		if d != time.Duration(i/policy.LimitForPeriod[0])*policy.LimitRefreshPeriod {
+			t.Errorf("wait duration of %d should not be: %s", i, d.String())
+		}
+	}
+
+	limiter.SetState(StateLimiting)
+	permitted, d, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	} else if d != policy.TimeoutDuration {
+		t.Errorf("wait duration should not be: %s", d.String())
+	}
+
+	limiter.SetState(StateLimiting)
+	now = now.Add(time.Millisecond * 5)
+	permitted, d, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	} else if d != policy.TimeoutDuration {
+		t.Errorf("wait duration should not be: %s", d.String())
+	}
+
+	now = now.Add(time.Millisecond * 6)
+	for i := 0; i < 5; i++ {
+		permitted, d, err := limiter.AcquirePermission([]int{1})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		} else if d != policy.TimeoutDuration-time.Millisecond {
+			t.Errorf("wait duration of %d should not be: %s", i, d.String())
+		}
+	}
+
+	permitted, d, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	} else if d != policy.TimeoutDuration {
+		t.Errorf("wait duration should not be: %s", d.String())
+	}
+
+	now = now.Add(time.Millisecond * 89)
+	for i := 0; i < 30; i++ {
+		permitted, d, err := limiter.AcquirePermission([]int{1})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		} else if d != time.Duration(i/policy.LimitForPeriod[0])*policy.LimitRefreshPeriod {
+			t.Errorf("wait duration of %d should not be: %s", i, d.String())
+		}
+	}
+
+	permitted, d, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	} else if d != policy.TimeoutDuration {
+		t.Errorf("wait duration should not be: %s", d.String())
+	}
+	limiter.SetState(StateDisabled)
+}
+
+func TestMultiRateLimiter2(t *testing.T) {
+	policy := NewMultiPolicy(0, 10*time.Millisecond, []int{10, 50})
+
+	limiter := NewMulti(policy)
+
+	// both exceed rate
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{1, 5})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{1, 5})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if permitted {
+			t.Errorf("AcquirePermission should fail: %d", i)
+		}
+	}
+
+	// second exceed rate
+	now = now.Add(time.Millisecond * 10)
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{0, 5})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{1, 0})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if permitted {
+			t.Errorf("AcquirePermission should fail: %d", i)
+		}
+	}
+
+	// first exceed rate
+	now = now.Add(time.Millisecond * 10)
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{1, 0})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{0, 5})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if permitted {
+			t.Errorf("AcquirePermission should fail: %d", i)
+		}
+	}
+
+	// both not exceed rate
+	now = now.Add(time.Millisecond * 10)
+	for i := 0; i < 100; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{0, 0})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+}
+
+func TestMultiRateLimiterN(t *testing.T) {
+	policy := NewMultiPolicy(50*time.Millisecond, 10*time.Millisecond, []int{5})
+
+	limiter := NewMulti(policy)
+	limiter.SetState(StateNormal)
+	for i := 0; i < 15; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{2})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	limiter.SetState(StateDisabled)
+	permitted, _, err := limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if !permitted {
+		t.Errorf("AcquirePermission should succeeded")
+	}
+
+	limiter.SetState(StateNormal)
+	for i := 0; i < 10; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{3})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	limiter.SetState(StateLimiting)
+	permitted, _, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	limiter.SetState(StateLimiting)
+	now = now.Add(time.Millisecond * 5)
+	permitted, _, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 6)
+	for i := 0; i < 5; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{1})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	permitted, _, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 89)
+	for i := 0; i < 5; i++ {
+		permitted, _, err := limiter.AcquirePermission([]int{6})
+		if err != nil {
+			t.Errorf("AcquirePermission fail, %v", err)
+		}
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	permitted, _, err = limiter.AcquirePermission([]int{1})
+	if err != nil {
+		t.Errorf("AcquirePermission fail, %v", err)
+	}
+	if permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+	limiter.SetState(StateDisabled)
+}
diff --git a/pkg/util/ratelimiter/ratelimiter_test.go b/pkg/util/ratelimiter/ratelimiter_test.go
index d604152d..8016acb9 100644
--- a/pkg/util/ratelimiter/ratelimiter_test.go
+++ b/pkg/util/ratelimiter/ratelimiter_test.go
@@ -49,7 +49,7 @@ func TestNewDefaultPolicy(t *testing.T) {
 	}
 }
 func TestConcurrent(t *testing.T) {
-	policy := NewPolicy(50, 10, 5)
+	policy := NewPolicy(50*time.Millisecond, 10*time.Millisecond, 5)
 
 	var wg sync.WaitGroup
 	limiter := New(policy)
@@ -100,7 +100,7 @@ func TestConcurrent(t *testing.T) {
 }
 
 func TestRateLimiter(t *testing.T) {
-	policy := NewPolicy(50, 10, 5)
+	policy := NewPolicy(50*time.Millisecond, 10*time.Millisecond, 5)
 
 	limiter := New(policy)
 	limiter.SetStateListener(func(event *Event) {
@@ -181,3 +181,66 @@ func TestRateLimiter(t *testing.T) {
 	}
 	limiter.SetState(StateDisabled)
 }
+
+func TestRateLimiterN(t *testing.T) {
+	policy := NewPolicy(50*time.Millisecond, 10*time.Millisecond, 5)
+
+	limiter := New(policy)
+	limiter.SetStateListener(func(event *Event) {
+		fmt.Printf("%v\n", event)
+	})
+	limiter.SetState(StateNormal)
+	for i := 0; i < 15; i++ {
+		permitted, _ := limiter.AcquireNPermission(2)
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	limiter.SetState(StateDisabled)
+	if permitted, _ := limiter.AcquirePermission(); !permitted {
+		t.Errorf("AcquirePermission should succeeded")
+	}
+
+	limiter.SetState(StateNormal)
+	for i := 0; i < 10; i++ {
+		permitted, _ := limiter.AcquireNPermission(3)
+		if !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	limiter.SetState(StateLimiting)
+	if permitted, _ := limiter.AcquirePermission(); permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	limiter.SetState(StateLimiting)
+	now = now.Add(time.Millisecond * 5)
+	if permitted, _ := limiter.AcquirePermission(); permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 6)
+	for i := 0; i < 5; i++ {
+		if permitted, _ := limiter.AcquirePermission(); !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	if permitted, _ := limiter.AcquirePermission(); permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+
+	now = now.Add(time.Millisecond * 89)
+	for i := 0; i < 5; i++ {
+		if permitted, _ := limiter.AcquireNPermission(6); !permitted {
+			t.Errorf("AcquirePermission should succeed: %d", i)
+		}
+	}
+
+	if permitted, _ := limiter.AcquirePermission(); permitted {
+		t.Errorf("AcquirePermission should fail")
+	}
+	limiter.SetState(StateDisabled)
+}