Repository: megaease/easegress
Pull request #500 (2022-02-08 00:49:50)

Custom data
refactor custom data, now Easegress & EaseMesh uses the same implementation

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

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

```diff
@@            Coverage Diff             @@
##             main     #500      +/-   ##
==========================================
- Coverage   80.18%   78.42%   -1.77%     
==========================================
  Files          83       85       +2     
  Lines        9676     9900     +224     
==========================================
+ Hits         7759     7764       +5     
- Misses       1472     1693     +221     
+ Partials      445      443       -2     
```


| [Impacted Files](https://codecov.io/gh/megaease/easegress/pull/500?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | Coverage Δ | |
|---|---|---|
| [pkg/cluster/customdatastore.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY3VzdG9tZGF0YXN0b3JlLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/cluster/layout.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvbGF5b3V0Lmdv) | `89.47% <0.00%> (ø)` | |
| [pkg/object/meshcontroller/spec/spec.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tZXNoY29udHJvbGxlci9zcGVjL3NwZWMuZ28=) | `86.62% <ø> (-0.45%)` | :arrow_down: |
| [pkg/util/dynamicobject/dynamicobject.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL3V0aWwvZHluYW1pY29iamVjdC9keW5hbWljb2JqZWN0Lmdv) | `91.66% <91.66%> (ø)` | |
| [pkg/object/mqttproxy/client.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvY2xpZW50Lmdv) | `80.54% <0.00%> (+0.90%)` | :arrow_up: |
| [pkg/cluster/cluster.go](https://codecov.io/gh/megaease/easegress/pull/500/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY2x1c3Rlci5nbw==) | `52.89% <0.00%> (+1.02%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/megaease/easegress/pull/500?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/500?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Last update [dc61784...117b756](https://codecov.io/gh/megaease/easegress/pull/500?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/cmd/client/command/common.go b/cmd/client/command/common.go
index 37aacd54..261cb1a8 100644
--- a/cmd/client/command/common.go
+++ b/cmd/client/command/common.go
@@ -63,8 +63,10 @@ const (
 	wasmCodeURL = apiURL + "/wasm/code"
 	wasmDataURL = apiURL + "/wasm/data/%s/%s"
 
-	customDataKindURL = apiURL + "/customdata/%s"
-	customDataURL     = apiURL + "/customdata/%s/%s"
+	customDataKindURL     = apiURL + "/customdatakinds"
+	customDataKindItemURL = apiURL + "/customdatakinds/%s"
+	customDataURL         = apiURL + "/customdata/%s"
+	customDataItemURL     = apiURL + "/customdata/%s/%s"
 
 	// MeshTenantsURL is the mesh tenant prefix.
 	MeshTenantsURL = apiURL + "/mesh/tenants"
diff --git a/cmd/client/command/customdata.go b/cmd/client/command/customdata.go
index 4f7e07de..ac9c82f7 100644
--- a/cmd/client/command/customdata.go
+++ b/cmd/client/command/customdata.go
@@ -24,6 +24,118 @@ import (
 	"github.com/spf13/cobra"
 )
 
+// CustomDataKindCmd defines custom data kind command.
+func CustomDataKindCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "custom-data-kind",
+		Short: "View and change custom data kind",
+	}
+
+	cmd.AddCommand(listCustomDataKindCmd())
+	cmd.AddCommand(getCustomDataKindCmd())
+	cmd.AddCommand(createCustomDataKindCmd())
+	cmd.AddCommand(updateCustomDataKindCmd())
+	cmd.AddCommand(deleteCustomDataKindCmd())
+
+	return cmd
+}
+
+func listCustomDataKindCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "list",
+		Short:   "List all custom data kinds",
+		Example: "egctl custom-data-kind list",
+
+		Run: func(cmd *cobra.Command, args []string) {
+			handleRequest(http.MethodGet, makeURL(customDataKindURL), nil, cmd)
+		},
+	}
+
+	return cmd
+}
+
+func getCustomDataKindCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "get",
+		Short:   "Get a custom data kind",
+		Example: "egctl custom-data-kind get <kind>",
+		Args: func(cmd *cobra.Command, args []string) error {
+			if len(args) != 1 {
+				return errors.New("requires custom data kind to be retrieved")
+			}
+			return nil
+		},
+
+		Run: func(cmd *cobra.Command, args []string) {
+			handleRequest(http.MethodGet, makeURL(customDataKindItemURL, args[0]), nil, cmd)
+		},
+	}
+
+	return cmd
+}
+
+func createCustomDataKindCmd() *cobra.Command {
+	var specFile string
+	cmd := &cobra.Command{
+		Use:     "create",
+		Short:   "Create a custom data kind from a yaml file or stdin",
+		Example: "egctl custom-data-kind create -f <kind file>",
+		Run: func(cmd *cobra.Command, args []string) {
+			visitor := buildYAMLVisitor(specFile, cmd)
+			visitor.Visit(func(yamlDoc []byte) error {
+				handleRequest(http.MethodPost, makeURL(customDataKindURL), yamlDoc, cmd)
+				return nil
+			})
+			visitor.Close()
+		},
+	}
+
+	cmd.Flags().StringVarP(&specFile, "file", "f", "", "A yaml file specifying the change request.")
+
+	return cmd
+}
+
+func updateCustomDataKindCmd() *cobra.Command {
+	var specFile string
+	cmd := &cobra.Command{
+		Use:     "update",
+		Short:   "Update a custom data from a yaml file or stdin",
+		Example: "egctl custom-data-kind update -f <kind file>",
+		Run: func(cmd *cobra.Command, args []string) {
+			visitor := buildYAMLVisitor(specFile, cmd)
+			visitor.Visit(func(yamlDoc []byte) error {
+				handleRequest(http.MethodPut, makeURL(customDataKindURL), yamlDoc, cmd)
+				return nil
+			})
+			visitor.Close()
+		},
+	}
+
+	cmd.Flags().StringVarP(&specFile, "file", "f", "", "A yaml file specifying the change request.")
+
+	return cmd
+}
+
+func deleteCustomDataKindCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "delete",
+		Short:   "Delete a custom data kind",
+		Example: "egctl custom-data-kind delete <kind>",
+		Args: func(cmd *cobra.Command, args []string) error {
+			if len(args) != 1 {
+				return errors.New("requires custom data kind to be retrieved")
+			}
+			return nil
+		},
+
+		Run: func(cmd *cobra.Command, args []string) {
+			handleRequest(http.MethodDelete, makeURL(customDataKindItemURL, args[0]), nil, cmd)
+		},
+	}
+
+	return cmd
+}
+
 // CustomDataCmd defines custom data command.
 func CustomDataCmd() *cobra.Command {
 	cmd := &cobra.Command{
@@ -33,7 +145,10 @@ func CustomDataCmd() *cobra.Command {
 
 	cmd.AddCommand(listCustomDataCmd())
 	cmd.AddCommand(getCustomDataCmd())
+	cmd.AddCommand(createCustomDataCmd())
 	cmd.AddCommand(updateCustomDataCmd())
+	cmd.AddCommand(batchUpdateCustomDataCmd())
+	cmd.AddCommand(deleteCustomDataCmd())
 
 	return cmd
 }
@@ -41,7 +156,7 @@ func CustomDataCmd() *cobra.Command {
 func getCustomDataCmd() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:     "get",
-		Short:   "Get an custom data",
+		Short:   "Get a custom data",
 		Example: "egctl custom-data get <kind> <id>",
 		Args: func(cmd *cobra.Command, args []string) error {
 			if len(args) != 2 {
@@ -51,7 +166,7 @@ func getCustomDataCmd() *cobra.Command {
 		},
 
 		Run: func(cmd *cobra.Command, args []string) {
-			handleRequest(http.MethodGet, makeURL(customDataURL, args[0], args[1]), nil, cmd)
+			handleRequest(http.MethodGet, makeURL(customDataItemURL, args[0], args[1]), nil, cmd)
 		},
 	}
 
@@ -70,19 +185,73 @@ func listCustomDataCmd() *cobra.Command {
 			return nil
 		},
 		Run: func(cmd *cobra.Command, args []string) {
-			handleRequest(http.MethodGet, makeURL(customDataKindURL, args[0]), nil, cmd)
+			handleRequest(http.MethodGet, makeURL(customDataURL, args[0]), nil, cmd)
 		},
 	}
 
 	return cmd
 }
 
+func createCustomDataCmd() *cobra.Command {
+	var specFile string
+	cmd := &cobra.Command{
+		Use:     "create",
+		Short:   "Create a custom data from a yaml file or stdin",
+		Example: "egctl custom-data create <kind> -f <data item file>",
+		Args: func(cmd *cobra.Command, args []string) error {
+			if len(args) != 1 {
+				return errors.New("requires custom data kind to be retrieved")
+			}
+			return nil
+		},
+		Run: func(cmd *cobra.Command, args []string) {
+			visitor := buildYAMLVisitor(specFile, cmd)
+			visitor.Visit(func(yamlDoc []byte) error {
+				handleRequest(http.MethodPost, makeURL(customDataURL, args[0]), yamlDoc, cmd)
+				return nil
+			})
+			visitor.Close()
+		},
+	}
+
+	cmd.Flags().StringVarP(&specFile, "file", "f", "", "A yaml file specifying the change request.")
+
+	return cmd
+}
+
 func updateCustomDataCmd() *cobra.Command {
 	var specFile string
 	cmd := &cobra.Command{
 		Use:     "update",
+		Short:   "Update a custom data from a yaml file or stdin",
+		Example: "egctl custom-data update <kind> -f <data item file>",
+		Args: func(cmd *cobra.Command, args []string) error {
+			if len(args) != 1 {
+				return errors.New("requires custom data kind to be retrieved")
+			}
+			return nil
+		},
+		Run: func(cmd *cobra.Command, args []string) {
+			visitor := buildYAMLVisitor(specFile, cmd)
+			visitor.Visit(func(yamlDoc []byte) error {
+				handleRequest(http.MethodPut, makeURL(customDataURL, args[0]), yamlDoc, cmd)
+				return nil
+			})
+			visitor.Close()
+		},
+	}
+
+	cmd.Flags().StringVarP(&specFile, "file", "f", "", "A yaml file specifying the change request.")
+
+	return cmd
+}
+
+func batchUpdateCustomDataCmd() *cobra.Command {
+	var specFile string
+	cmd := &cobra.Command{
+		Use:     "batch-update",
 		Short:   "Batch update custom data from a yaml file or stdin",
-		Example: "egctl custom-data update <kind> -f <change request file>",
+		Example: "egctl custom-data batch-update <kind> -f <change request file>",
 		Args: func(cmd *cobra.Command, args []string) error {
 			if len(args) != 1 {
 				return errors.New("requires custom data kind to be retrieved")
@@ -92,7 +261,7 @@ func updateCustomDataCmd() *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			visitor := buildYAMLVisitor(specFile, cmd)
 			visitor.Visit(func(yamlDoc []byte) error {
-				handleRequest(http.MethodPost, makeURL(customDataKindURL, args[0]), yamlDoc, cmd)
+				handleRequest(http.MethodPost, makeURL(customDataItemURL, args[0], "items"), yamlDoc, cmd)
 				return nil
 			})
 			visitor.Close()
@@ -103,3 +272,22 @@ func updateCustomDataCmd() *cobra.Command {
 
 	return cmd
 }
+
+func deleteCustomDataCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "delete",
+		Short:   "Delete a custom data item",
+		Example: "egctl custom-data delete <kind> <id>",
+		Args: func(cmd *cobra.Command, args []string) error {
+			if len(args) != 2 {
+				return errors.New("requires custom data kind and id to be retrieved")
+			}
+			return nil
+		},
+		Run: func(cmd *cobra.Command, args []string) {
+			handleRequest(http.MethodDelete, makeURL(customDataItemURL, args[0], args[1]), nil, cmd)
+		},
+	}
+
+	return cmd
+}
diff --git a/cmd/client/main.go b/cmd/client/main.go
index d200591c..1bdd1e5b 100644
--- a/cmd/client/main.go
+++ b/cmd/client/main.go
@@ -110,6 +110,7 @@ func main() {
 		command.ObjectCmd(),
 		command.MemberCmd(),
 		command.WasmCmd(),
+		command.CustomDataKindCmd(),
 		command.CustomDataCmd(),
 		completionCmd,
 	)
diff --git a/doc/README.md b/doc/README.md
index 13c55f5a..b500037a 100644
--- a/doc/README.md
+++ b/doc/README.md
@@ -9,6 +9,7 @@
 			- [4.1.1 System Controllers](#411-system-controllers)
 			- [4.1.2 Business Controllers](#412-business-controllers)
 		- [4.2 Filters](#42-filters)
+		- [4.3 Custom Data](#43-custom-data)
 
 ## 1. Cookbook / How-To Guide
 
@@ -91,4 +92,6 @@ It could be created, updated, deleted by admin operation. They control various r
 - [Validator](./reference/filters.md#Validator) - The Validator filter validates requests, forwards valid ones, and rejects invalid ones. 
 - [WasmHost](./reference/filters.md#WasmHost) - The WasmHost filter implements a host environment for user-developed WebAssembly code. 
 
+### 4.3 Custom Data
 
+- [Custom Data Management](./reference/customdata.md) - Create/Read/Update/Delete custom data kinds and custom data items.
diff --git a/doc/reference/customdata.md b/doc/reference/customdata.md
new file mode 100644
index 00000000..9467bd76
--- /dev/null
+++ b/doc/reference/customdata.md
@@ -0,0 +1,105 @@
+# Custom Data Management
+
+The `Custom Data` feature implements a storage for 'any' data, which can be used by other components for data persistence.
+
+Because the schema of the data being persisted varies from components, a `CustomDataKind` must be defined to distinguish the data.
+
+## CustomDataKind
+
+The YAML example below defines a CustomDataKind:
+
+```yaml
+name: kind1
+idField: name
+jsonSchema:
+  type: object
+  properties:
+    name:
+      type: string
+      required: true
+```
+
+The `name` field is required, it is the kind name of custom data items of this kind.
+The `idField` is optional, and its default value is `name`, the value of this field of a data item is used as its identifier.
+The `jsonSchema` is optional, if provided, data items of this kind will be validated against this [JSON Schema](http://json-schema.org/).
+
+## CustomData
+
+CustomData is a map, the keys of this map must be strings while the values can be any valid JSON values, but the keys of a nested map must be strings too.
+
+A CustomData item must contain the `idField` defined by its corresponding CustomDataKind, for example, the data items of the kind defined in the above example must contain the `name` field as their identifiers.
+
+Below is an example of a CustomData:
+
+```yaml
+name: data1
+field1: 12
+field2: abc
+field3: [1, 2, 3, 4]
+```
+
+## API
+
+* **Create a CustomDataKind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdatakinds
+	* **Method**: POST
+	* **Body**: CustomDataKind definition is YAML.
+
+* **Update a CustomDataKind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdatakinds
+	* **Method**: PUT
+	* **Body**: CustomDataKind definition is YAML.
+
+* **Query the definition of a CustomDataKind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdatakinds/{kind name}
+	* **Method**: GET
+  
+* **List the definition of all CustomDataKind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdatakinds
+	* **Method**: GET
+
+* **Delete a CustomDataKind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdatakinds/{kind name}
+	* **Method**: DELETE
+
+* **Create a CustomData**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}
+	* **Method**: POST
+	* **Body**: CustomData definition is YAML.
+
+* **Update a CustomData**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}
+	* **Method**: PUT
+	* **Body**: CustomData definition is YAML.
+
+* **Query the definition of a CustomData**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}/{data id}
+	* **Method**: GET
+
+* **List the definition of all CustomData of a kind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}
+	* **Method**: GET
+
+* **Delete a CustomData**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}/{data id}
+	* **Method**: DELETE
+
+* **Delete all CustomData of a kind**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}
+	* **Method**: DELETE
+
+* **Bulk update**
+	* **URL**: http://{ip}:{port}/apis/v1/customdata/{kind name}/items
+	* **Method**: POST
+	* **Body**: A change request in YAML, as defined below.
+
+```yaml
+rebuild: false
+delete: [data1, data2]
+list:
+- name: data3
+  field1: 12
+- name: data4
+  field1: foo
+```
+When `rebuild` is true (default is false), all existing data items are deleted before processing the data items in `list`. `delete` is an array of data identifiers to be deleted, this array is ignored when `rebuild` is true. `list` is an array of data items to be created or updated.
diff --git a/doc/reference/filters.md b/doc/reference/filters.md
index f710b9e2..f1697d46 100644
--- a/doc/reference/filters.md
+++ b/doc/reference/filters.md
@@ -52,6 +52,8 @@
   - [HeaderToJSON](#headertojson)
     - [Configuration](#configuration-16)
     - [Results](#results-16)
+  - [CertExtractor](#certextractor)
+    - [Configuration](#configuration-17)
   - [Common Types](#common-types)
     - [apiaggregator.Pipeline](#apiaggregatorpipeline)
     - [pathadaptor.Spec](#pathadaptorspec)
@@ -250,6 +252,8 @@ allowedMethods: [GET]
 | allowedHeaders   | []string | An array of non-simple headers the client is allowed to use with cross-domain requests. If the special `*` value is present in the list, all headers will be allowed. The default value is [] but "Origin" is always appended to the list                                                                                                                               | No       |
 | allowCredentials | bool     | Indicates whether the request can include user credentials like cookies, HTTP authentication, or client-side SSL certificates                                                                                                                                                                                                                                           | No       |
 | exposedHeaders   | []string | Indicates which headers are safe to expose to the API of a CORS API specification                                                                                                                                                                                                                                                                                       | No       |
+| maxAge   | int | Indicates how long (in seconds) the results of a preflight request can be cached. The default is 0 stands for no max age                                                                                                                                                                                                                                                                                       | No       |
+| supportCORSRequest   | bool | When true, support CORS request and CORS preflight requests. By default, support only preflight requests.                                                                                                                                                                                                                                                                                       | No       |
 
 ### Results
 
@@ -379,17 +383,22 @@ path:
 
 ### Configuration
 
-| Name   | Type                                         | Description                                                                                                                                                                                                         | Required |
-| ------ | -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
-| method | string                                       | If provided, the method of the original request is replaced by the value of this option                                                                                                                             | No       |
-| path   | [pathadaptor.Spec](#pathadaptorSpec)         | Rules to revise request path                                                                                                                                                                                        | No       |
-| header | [httpheader.AdaptSpec](#httpheaderAdaptSpec) | Rules to revise request header                                                                                                                                                                                      | No       |
-| body   | string                                       | If provided the body of the original request is replaced by the value of this option. Note: the body can be a template, which means runtime variables (enclosed by `[[` & `]]`) are replaced by their actual values | No       |
-| host   | string                                       | If provided the host of the original request is replaced by the value of this option. Note: the host can be a template, which means runtime variables (enclosed by `[[` & `]]`) are replaced by their actual values | No       |
+| Name       | Type                                         | Description                                                                                                                                                                                                         | Required |
+| -----------| -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
+| method     | string                                       | If provided, the method of the original request is replaced by the value of this option                                                                                                                             | No       |
+| path       | [pathadaptor.Spec](#pathadaptorSpec)         | Rules to revise request path                                                                                                                                                                                        | No       |
+| header     | [httpheader.AdaptSpec](#httpheaderAdaptSpec) | Rules to revise request header                                                                                                                                                                                      | No       |
+| body       | string                                       | If provided the body of the original request is replaced by the value of this option. Note: the body can be a template, which means runtime variables (enclosed by `[[` & `]]`) are replaced by their actual values | No       |
+| host       | string                                       | If provided the host of the original request is replaced by the value of this option. Note: the host can be a template, which means runtime variables (enclosed by `[[` & `]]`) are replaced by their actual values | No       |
+| decompress | string                                       | If provided, the request body is replaced by the value of decompressed body. Now support "gzip" decompress                                                                                                          | No       |
+| compress   | string                                       | If provided, the request body is replaced by the value of compressed body. Now support "gzip" compress                                                                                                              | No       |
 
 ### Results
 
-The RequestAdaptor always returns an empty result.
+| Value          | Description                              |
+| -------------- | ---------------------------------------- |
+| decompressFail | the request body can not be decompressed |
+| compressFail   | the request body can not be compressed   |
 
 ## CircuitBreaker
 
@@ -734,8 +743,7 @@ headerMap:
 
 | Name         | Type     | Description                      | Required |
 | ------------ | -------- | -------------------------------- | -------- |
-| headerMap | [][HeaderToJSON.HeaderMap](#headertojsonheadermap) | headerMap defines a map between HTTP header name and corresponding JSON filed name
-  | Yes      |
+| headerMap | [][HeaderToJSON.HeaderMap](#headertojsonheadermap) | headerMap defines a map between HTTP header name and corresponding JSON field name | Yes      |
 
 
 ### Results
@@ -744,6 +752,31 @@ headerMap:
 | ----------------------- | ------------------------------------ |
 | jsonEncodeDecodeErr     | Failed to convert HTTP headers to JSON. |
 
+## CertExtractor
+
+CertExtractor extracts a value from requests TLS certificates Subject or Issuer metadata (https://pkg.go.dev/crypto/x509/pkix#Name) and adds the value to headers. Request can contain zero or multiple certificates so the position (first, second, last, etc) of the certificate in the chain is required.
+
+Here's an example configuration, that adds a new header `tls-cert-postalcode`, based on the PostalCode of the last TLS certificate's Subject:
+
+```yaml
+kind: "CertExtractor"
+name: "postalcode-extractor"
+certIndex: -1 # take last certificate in chain
+target: "subject"
+field: "PostalCode"
+headerKey: "tls-cert-postalcode"
+```
+
+### Configuration
+
+| Name         | Type     | Description                      | Required |
+| ------------ | -------- | -------------------------------- | -------- |
+| certIndex | int16 | The index of the certificate in the chain. Negative indexes from the end of the chain (-1 is the last index, -2 second last etc.) | Yes      |
+| target | string | Either `subject` or `issuer` of the [x509.Certificate](https://pkg.go.dev/crypto/x509#Certificate) | Yes      |
+| field | string | One of the string or string slice fields from https://pkg.go.dev/crypto/x509/pkix#Name  | Yes      |
+| headerKey | string | Extracted value is added to this request header key. | Yes      |
+
+
 ## Common Types
 
 ### apiaggregator.Pipeline
diff --git a/pkg/api/customdata.go b/pkg/api/customdata.go
index b701d7dc..21af6d35 100644
--- a/pkg/api/customdata.go
+++ b/pkg/api/customdata.go
@@ -18,134 +18,243 @@
 package api
 
 import (
+	"fmt"
 	"net/http"
 
 	"github.com/go-chi/chi/v5"
-	"go.etcd.io/etcd/client/v3/concurrency"
+	"github.com/megaease/easegress/pkg/cluster/customdata"
 	"gopkg.in/yaml.v2"
 )
 
 const (
-	// CustomDataPrefix is the object prefix.
+	// CustomDataKindPrefix is the URL prefix of APIs for custom data kind
+	CustomDataKindPrefix = "/customdatakinds"
+	// CustomDataPrefix is the URL prefix of APIs for custom data
 	CustomDataPrefix = "/customdata/{kind}"
 )
 
-type (
-	// CustomData defines the custom data type
-	CustomData map[string]interface{}
-
-	// ChangeRequest represents a change request to custom data
-	ChangeRequest struct {
-		Rebuild bool         `yaml:"rebuild"`
-		Delete  []string     `yaml:"delete"`
-		List    []CustomData `yaml:"list"`
-	}
-)
-
-// Key returns the 'key' field of the custom data
-func (cd CustomData) Key() string {
-	if v, ok := cd["key"].(string); ok {
-		return v
-	}
-	return ""
+// ChangeRequest represents a change request to custom data
+type ChangeRequest struct {
+	Rebuild bool              `yaml:"rebuild"`
+	Delete  []string          `yaml:"delete"`
+	List    []customdata.Data `yaml:"list"`
 }
 
 func (s *Server) customDataAPIEntries() []*Entry {
 	return []*Entry{
+		{
+			Path:    CustomDataKindPrefix,
+			Method:  http.MethodGet,
+			Handler: s.listCustomDataKind,
+		},
+		{
+			Path:    CustomDataKindPrefix + "/{name}",
+			Method:  http.MethodGet,
+			Handler: s.getCustomDataKind,
+		},
+		{
+			Path:    CustomDataKindPrefix,
+			Method:  http.MethodPost,
+			Handler: s.createCustomDataKind,
+		},
+		{
+			Path:    CustomDataKindPrefix,
+			Method:  http.MethodPut,
+			Handler: s.updateCustomDataKind,
+		},
+		{
+			Path:    CustomDataKindPrefix + "/{name}",
+			Method:  http.MethodDelete,
+			Handler: s.deleteCustomDataKind,
+		},
+
 		{
 			Path:    CustomDataPrefix,
 			Method:  http.MethodGet,
 			Handler: s.listCustomData,
 		},
+		{
+			Path:    CustomDataPrefix + "/{id}",
+			Method:  http.MethodGet,
+			Handler: s.getCustomData,
+		},
 		{
 			Path:    CustomDataPrefix,
 			Method:  http.MethodPost,
+			Handler: s.createCustomData,
+		},
+		{
+			Path:    CustomDataPrefix,
+			Method:  http.MethodPut,
 			Handler: s.updateCustomData,
 		},
 		{
-			Path:    CustomDataPrefix + "/{key}",
-			Method:  http.MethodGet,
-			Handler: s.getCustomData,
+			Path:    CustomDataPrefix + "/{id}",
+			Method:  http.MethodDelete,
+			Handler: s.deleteCustomData,
+		},
+
+		{
+			Path:    CustomDataPrefix + "/items",
+			Method:  http.MethodPost,
+			Handler: s.batchUpdateCustomData,
 		},
 	}
 }
 
-func (s *Server) listCustomData(w http.ResponseWriter, r *http.Request) {
-	kind := chi.URLParam(r, "kind")
-	prefix := s.cluster.Layout().CustomDataPrefix(kind)
-	kvs, err := s.cluster.GetRawPrefix(prefix)
+func (s *Server) listCustomDataKind(w http.ResponseWriter, r *http.Request) {
+	result, err := s.cds.ListKinds()
 	if err != nil {
 		ClusterPanic(err)
 	}
 
-	result := make([]CustomData, 0, len(kvs))
-	for key, kv := range kvs {
-		key = key[len(prefix):]
-		cd := CustomData{}
-		err = yaml.Unmarshal(kv.Value, &cd)
-		if err != nil {
-			panic(err)
-		}
-		result = append(result, cd)
+	w.Header().Set("Content-Type", "text/vnd.yaml")
+	err = yaml.NewEncoder(w).Encode(result)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (s *Server) getCustomDataKind(w http.ResponseWriter, r *http.Request) {
+	name := chi.URLParam(r, "name")
+	k, err := s.cds.GetKind(name)
+	if err != nil {
+		ClusterPanic(err)
 	}
 
 	w.Header().Set("Content-Type", "text/vnd.yaml")
-	err = yaml.NewEncoder(w).Encode(result)
+	err = yaml.NewEncoder(w).Encode(k)
 	if err != nil {
 		panic(err)
 	}
 }
 
-func (s *Server) updateCustomData(w http.ResponseWriter, r *http.Request) {
-	kind := chi.URLParam(r, "kind")
+func (s *Server) createCustomDataKind(w http.ResponseWriter, r *http.Request) {
+	k := customdata.Kind{}
+	err := yaml.NewDecoder(r.Body).Decode(&k)
+	if err != nil {
+		panic(err)
+	}
+	err = s.cds.PutKind(&k, false)
+	if err != nil {
+		ClusterPanic(err)
+	}
 
-	var cr ChangeRequest
-	err := yaml.NewDecoder(r.Body).Decode(&cr)
+	w.WriteHeader(http.StatusCreated)
+	location := fmt.Sprintf("%s/%s", r.URL.Path, k.Name)
+	w.Header().Set("Location", location)
+}
+
+func (s *Server) updateCustomDataKind(w http.ResponseWriter, r *http.Request) {
+	k := customdata.Kind{}
+	err := yaml.NewDecoder(r.Body).Decode(&k)
 	if err != nil {
 		panic(err)
 	}
+	err = s.cds.PutKind(&k, true)
+	if err != nil {
+		ClusterPanic(err)
+	}
+}
 
-	prefix := s.cluster.Layout().CustomDataPrefix(kind)
-	if cr.Rebuild {
-		err = s.cluster.DeletePrefix(prefix)
-		if err != nil {
-			ClusterPanic(err)
-		}
+func (s *Server) deleteCustomDataKind(w http.ResponseWriter, r *http.Request) {
+	name := chi.URLParam(r, "name")
+	err := s.cds.DeleteKind(name)
+	if err != nil {
+		ClusterPanic(err)
 	}
+}
 
-	err = s.cluster.STM(func(stm concurrency.STM) error {
-		if !cr.Rebuild {
-			for _, key := range cr.Delete {
-				key = s.cluster.Layout().CustomDataItem(kind, key)
-				stm.Del(key)
-			}
-		}
-		for _, v := range cr.List {
-			key := v.Key()
-			key = s.cluster.Layout().CustomDataItem(kind, key)
-			data, err := yaml.Marshal(v)
-			if err != nil {
-				return err
-			}
-			stm.Put(key, string(data))
-		}
-		return nil
-	})
+func (s *Server) listCustomData(w http.ResponseWriter, r *http.Request) {
+	kind := chi.URLParam(r, "kind")
+
+	result, err := s.cds.ListData(kind)
 	if err != nil {
 		ClusterPanic(err)
 	}
+
+	w.Header().Set("Content-Type", "text/vnd.yaml")
+	err = yaml.NewEncoder(w).Encode(result)
+	if err != nil {
+		panic(err)
+	}
 }
 
 func (s *Server) getCustomData(w http.ResponseWriter, r *http.Request) {
 	kind := chi.URLParam(r, "kind")
-	key := chi.URLParam(r, "key")
+	id := chi.URLParam(r, "id")
 
-	key = s.cluster.Layout().CustomDataItem(kind, key)
-	kv, err := s.cluster.GetRaw(key)
+	data, err := s.cds.GetData(kind, id)
 	if err != nil {
 		ClusterPanic(err)
 	}
 
 	w.Header().Set("Content-Type", "text/vnd.yaml")
-	w.Write(kv.Value)
+	err = yaml.NewEncoder(w).Encode(data)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (s *Server) createCustomData(w http.ResponseWriter, r *http.Request) {
+	kind := chi.URLParam(r, "kind")
+
+	data := customdata.Data{}
+	err := yaml.NewDecoder(r.Body).Decode(&data)
+	if err != nil {
+		panic(err)
+	}
+	id, err := s.cds.PutData(kind, data, false)
+	if err != nil {
+		ClusterPanic(err)
+	}
+
+	w.WriteHeader(http.StatusCreated)
+	location := fmt.Sprintf("%s/%s", r.URL.Path, id)
+	w.Header().Set("Location", location)
+}
+
+func (s *Server) updateCustomData(w http.ResponseWriter, r *http.Request) {
+	kind := chi.URLParam(r, "kind")
+
+	data := customdata.Data{}
+	err := yaml.NewDecoder(r.Body).Decode(&data)
+	if err != nil {
+		panic(err)
+	}
+	_, err = s.cds.PutData(kind, data, true)
+	if err != nil {
+		ClusterPanic(err)
+	}
+}
+
+func (s *Server) deleteCustomData(w http.ResponseWriter, r *http.Request) {
+	kind := chi.URLParam(r, "kind")
+	id := chi.URLParam(r, "id")
+	err := s.cds.DeleteData(kind, id)
+	if err != nil {
+		ClusterPanic(err)
+	}
+}
+
+func (s *Server) batchUpdateCustomData(w http.ResponseWriter, r *http.Request) {
+	kind := chi.URLParam(r, "kind")
+
+	var cr ChangeRequest
+	err := yaml.NewDecoder(r.Body).Decode(&cr)
+	if err != nil {
+		panic(err)
+	}
+
+	if cr.Rebuild {
+		err = s.cds.DeleteAllData(kind)
+		if err != nil {
+			ClusterPanic(err)
+		}
+	}
+
+	err = s.cds.BatchUpdateData(kind, cr.Delete, cr.List)
+	if err != nil {
+		ClusterPanic(err)
+	}
 }
diff --git a/pkg/api/server.go b/pkg/api/server.go
index dbbdaca0..acb7f57f 100644
--- a/pkg/api/server.go
+++ b/pkg/api/server.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/megaease/easegress/pkg/cluster"
+	"github.com/megaease/easegress/pkg/cluster/customdata"
 	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/option"
 	"github.com/megaease/easegress/pkg/supervisor"
@@ -37,6 +38,7 @@ type (
 		router  *dynamicMux
 		cluster cluster.Cluster
 		super   *supervisor.Supervisor
+		cds     *customdata.Store
 
 		mutex      cluster.Mutex
 		mutexMutex sync.Mutex
@@ -57,10 +59,10 @@ type (
 )
 
 // MustNewServer creates an api server.
-func MustNewServer(opt *option.Options, cluster cluster.Cluster, super *supervisor.Supervisor) *Server {
+func MustNewServer(opt *option.Options, cls cluster.Cluster, super *supervisor.Supervisor) *Server {
 	s := &Server{
 		opt:     opt,
-		cluster: cluster,
+		cluster: cls,
 		super:   super,
 	}
 	s.router = newDynamicMux(s)
@@ -71,6 +73,10 @@ func MustNewServer(opt *option.Options, cluster cluster.Cluster, super *supervis
 		logger.Errorf("get cluster mutex %s failed: %v", lockKey, err)
 	}
 
+	kindPrefix := cls.Layout().CustomDataKindPrefix()
+	dataPrefix := cls.Layout().CustomDataPrefix()
+	s.cds = customdata.NewStore(cls, kindPrefix, dataPrefix)
+
 	s.initMetadata()
 	s.registerAPIs()
 
diff --git a/pkg/cluster/customdata/customdata.go b/pkg/cluster/customdata/customdata.go
new file mode 100644
index 00000000..ed1ebf88
--- /dev/null
+++ b/pkg/cluster/customdata/customdata.go
@@ -0,0 +1,368 @@
+/*
+ * 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 customdata
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/megaease/easegress/pkg/cluster"
+	"github.com/megaease/easegress/pkg/util/dynamicobject"
+	"github.com/xeipuuv/gojsonschema"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	yaml "gopkg.in/yaml.v2"
+)
+
+// Data represents a custom data
+type Data = dynamicobject.DynamicObject
+
+// Kind defines the spec of a custom data kind
+type Kind struct {
+	// Name is the name of the Kind
+	Name string `yaml:"name" jsonschema:"required"`
+	// IDField is a field name of custom data of this kind, this field is the ID
+	// of the data, that's unique among the same kind, the default value is 'name'.
+	IDField string `yaml:"idField" jsonschema:"omitempty"`
+	// JSONSchema is JSON schema to validate a custom data of this kind
+	JSONSchema dynamicobject.DynamicObject `yaml:"jsonSchema" jsonschema:"omitempty"`
+}
+
+func (k *Kind) dataID(data Data) string {
+	var id string
+	if k.IDField == "" {
+		id, _ = data["name"].(string)
+	} else {
+		id, _ = data[k.IDField].(string)
+	}
+	return id
+}
+
+// Store defines the storage for custom data
+type Store struct {
+	cluster    cluster.Cluster
+	KindPrefix string
+	DataPrefix string
+}
+
+// NewStore creates a new custom data store
+func NewStore(cls cluster.Cluster, kindPrefix string, dataPrefix string) *Store {
+	return &Store{
+		cluster:    cls,
+		KindPrefix: kindPrefix,
+		DataPrefix: dataPrefix,
+	}
+}
+
+func unmarshalKind(in []byte) (*Kind, error) {
+	kind := &Kind{}
+	err := yaml.Unmarshal(in, kind)
+	if err != nil {
+		return nil, fmt.Errorf("BUG: unmarshal %s to yaml failed: %v", string(in), err)
+	}
+	return kind, nil
+}
+
+func (s *Store) kindKey(name string) string {
+	return s.KindPrefix + name
+}
+
+// GetKind gets custom data kind by its name
+func (s *Store) GetKind(name string) (*Kind, error) {
+	kvs, err := s.cluster.GetRaw(s.kindKey(name))
+	if err != nil {
+		return nil, err
+	}
+
+	if kvs == nil {
+		return nil, nil
+	}
+
+	return unmarshalKind(kvs.Value)
+}
+
+// ListKinds lists custom data kinds
+func (s *Store) ListKinds() ([]*Kind, error) {
+	kvs, err := s.cluster.GetRawPrefix(s.KindPrefix)
+	if err != nil {
+		return nil, err
+	}
+
+	kinds := make([]*Kind, 0, len(kvs))
+	for _, v := range kvs {
+		kind, err := unmarshalKind(v.Value)
+		if err != nil {
+			return nil, err
+		}
+		kinds = append(kinds, kind)
+	}
+
+	return kinds, nil
+}
+
+// PutKind creates or updates a custom data kind
+func (s *Store) PutKind(kind *Kind, update bool) error {
+	if len(kind.JSONSchema) > 0 {
+		sl := gojsonschema.NewGoLoader(kind.JSONSchema)
+		if _, err := gojsonschema.NewSchema(sl); err != nil {
+			return fmt.Errorf("invalid JSONSchema: %s", err.Error())
+		}
+	}
+
+	oldKind, err := s.GetKind(kind.Name)
+	if err != nil {
+		return err
+	}
+
+	if update && (oldKind == nil) {
+		return fmt.Errorf("%s not found", kind.Name)
+	}
+	if (!update) && (oldKind != nil) {
+		return fmt.Errorf("%s existed", kind.Name)
+	}
+
+	buf, err := yaml.Marshal(kind)
+	if err != nil {
+		return fmt.Errorf("BUG: marshal %#v to yaml failed: %v", kind, err)
+	}
+
+	key := s.kindKey(kind.Name)
+	return s.cluster.Put(key, string(buf))
+}
+
+// DeleteKind deletes a custom data kind
+func (s *Store) DeleteKind(name string) error {
+	kind, err := s.GetKind(name)
+	if err != nil {
+		return err
+	}
+
+	if kind == nil {
+		return fmt.Errorf("%s not found", name)
+	}
+
+	return s.cluster.Delete(s.kindKey(name))
+	// TODO: remove custom data?
+}
+
+func (s *Store) dataPrefix(kind string) string {
+	return s.DataPrefix + kind + "/"
+}
+
+func (s *Store) dataKey(kind string, id string) string {
+	return s.dataPrefix(kind) + id
+}
+
+func unmarshalData(in []byte) (Data, error) {
+	data := Data{}
+	err := yaml.Unmarshal(in, &data)
+	if err != nil {
+		return nil, fmt.Errorf("BUG: unmarshal %s to yaml failed: %v", string(in), err)
+	}
+	return data, nil
+}
+
+// GetData gets custom data by its id
+func (s *Store) GetData(kind string, id string) (Data, error) {
+	kvs, err := s.cluster.GetRaw(s.dataKey(kind, id))
+	if err != nil {
+		return nil, err
+	}
+
+	if kvs == nil {
+		return nil, nil
+	}
+
+	return unmarshalData(kvs.Value)
+}
+
+// ListData lists custom data of specified kind.
+// if kind is empty, it returns custom data of all kinds.
+func (s *Store) ListData(kind string) ([]Data, error) {
+	key := s.DataPrefix
+	if kind != "" {
+		key = s.dataPrefix(kind)
+	}
+	kvs, err := s.cluster.GetRawPrefix(key)
+	if err != nil {
+		return nil, err
+	}
+
+	results := make([]Data, 0, len(kvs))
+	for _, v := range kvs {
+		data, err := unmarshalData(v.Value)
+		if err != nil {
+			return nil, err
+		}
+		results = append(results, data)
+	}
+
+	return results, nil
+}
+
+// PutData creates or updates a custom data item
+func (s *Store) PutData(kind string, data Data, update bool) (string, error) {
+	k, err := s.GetKind(kind)
+	if err != nil {
+		return "", err
+	}
+	if k == nil {
+		return "", fmt.Errorf("kind %s not found", kind)
+	}
+
+	id := k.dataID(data)
+	if id == "" {
+		return "", fmt.Errorf("data id is empty")
+	}
+
+	if len(k.JSONSchema) > 0 {
+		schema := gojsonschema.NewGoLoader(k.JSONSchema)
+		doc := gojsonschema.NewGoLoader(data)
+		res, err := gojsonschema.Validate(schema, doc)
+		if err != nil {
+			return "", fmt.Errorf("error occurs during validation: %v", err)
+		}
+		if !res.Valid() {
+			return "", fmt.Errorf("validation failed: %v", res.Errors())
+		}
+	}
+
+	oldData, err := s.GetData(kind, id)
+	if err != nil {
+		return "", err
+	}
+	if update && (oldData == nil) {
+		return "", fmt.Errorf("%s/%s not found", kind, id)
+	}
+	if (!update) && (oldData != nil) {
+		return "", fmt.Errorf("%s/%s existed", kind, id)
+	}
+
+	buf, err := yaml.Marshal(data)
+	if err != nil {
+		return "", fmt.Errorf("BUG: marshal %#v to yaml failed: %v", data, err)
+	}
+
+	key := s.dataKey(kind, id)
+	err = s.cluster.Put(key, string(buf))
+	if err != nil {
+		return "", err
+	}
+
+	return id, nil
+}
+
+// BatchUpdateData updates multiple custom data in a transaction
+func (s *Store) BatchUpdateData(kind string, del []string, update []Data) error {
+	k, err := s.GetKind(kind)
+	if err != nil {
+		return err
+	}
+	if k == nil {
+		return fmt.Errorf("kind %s not found", kind)
+	}
+
+	if len(update) > 0 && len(k.JSONSchema) > 0 {
+		schema := gojsonschema.NewGoLoader(k.JSONSchema)
+		for _, data := range update {
+			doc := gojsonschema.NewGoLoader(data)
+			res, err := gojsonschema.Validate(schema, doc)
+			if err != nil {
+				return fmt.Errorf("error occurs during validation: %v", err)
+			}
+			if !res.Valid() {
+				return fmt.Errorf("validation failed: %v", res.Errors())
+			}
+		}
+	}
+
+	for _, data := range update {
+		if k.dataID(data) == "" {
+			return fmt.Errorf("data id is empty")
+		}
+	}
+
+	return s.cluster.STM(func(stm concurrency.STM) error {
+		for _, id := range del {
+			key := s.dataKey(kind, id)
+			stm.Del(key)
+		}
+
+		for _, data := range update {
+			id := k.dataID(data)
+			buf, err := yaml.Marshal(data)
+			if err != nil {
+				return fmt.Errorf("BUG: marshal %#v to yaml failed: %v", data, err)
+			}
+			key := s.dataKey(kind, id)
+			stm.Put(key, string(buf))
+		}
+		return nil
+	})
+}
+
+// DeleteData deletes a custom data
+func (s *Store) DeleteData(kind string, id string) error {
+	data, err := s.GetData(kind, id)
+	if err != nil {
+		return err
+	}
+
+	if data == nil {
+		return fmt.Errorf("%s not found", id)
+	}
+
+	return s.cluster.Delete(s.dataKey(kind, id))
+}
+
+// DeleteAllData deletes all custom data of kind 'kind'
+func (s *Store) DeleteAllData(kind string) error {
+	prefix := s.dataPrefix(kind)
+	return s.cluster.DeletePrefix(prefix)
+}
+
+// Watch watches the data of custom data kind 'kind'
+func (s *Store) Watch(ctx context.Context, kind string, onChange func([]Data)) error {
+	syncer, err := s.cluster.Syncer(5 * time.Minute)
+	if err != nil {
+		return err
+	}
+
+	prefix := s.dataPrefix(kind)
+	ch, err := syncer.SyncRawPrefix(prefix)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			syncer.Close()
+			return nil
+		case m := <-ch:
+			data := make([]Data, 0, len(m))
+			for _, v := range m {
+				d, err := unmarshalData(v.Value)
+				if err == nil {
+					data = append(data, d)
+				}
+			}
+			onChange(data)
+		}
+	}
+}
diff --git a/pkg/cluster/layout.go b/pkg/cluster/layout.go
index 80331c75..daf5567f 100644
--- a/pkg/cluster/layout.go
+++ b/pkg/cluster/layout.go
@@ -33,9 +33,9 @@ const (
 	configObjectFormat       = "/config/objects/%s" // +objectName
 	configVersion            = "/config/version"
 	wasmCodeEvent            = "/wasm/code"
-	wasmDataPrefixFormat     = "/wasm/data/%s/%s/"  // + pipelineName + filterName
-	customDataPrefixFormat   = "/custom-data/%s/"   // + kind
-	customDataItemFormat     = "/custom-data/%s/%s" // + kind + item key
+	wasmDataPrefixFormat     = "/wasm/data/%s/%s/" // + pipelineName + filterName
+	customDataKindPrefix     = "/custom-data-kinds/"
+	customDataPrefix         = "/custom-data/"
 
 	// the cluster name of this eg group will be registered under this path in etcd
 	// any new member(primary or secondary ) will be rejected if it is configured a different cluster name
@@ -131,12 +131,12 @@ func (l *Layout) WasmDataPrefix(pipeline string, name string) string {
 	return fmt.Sprintf(wasmDataPrefixFormat, pipeline, name)
 }
 
-// CustomDataPrefix returns the prefix of a custom data kind
-func (l *Layout) CustomDataPrefix(kind string) string {
-	return fmt.Sprintf(customDataPrefixFormat, kind)
+// CustomDataPrefix returns the prefix of all custom data
+func (l *Layout) CustomDataPrefix() string {
+	return customDataPrefix
 }
 
-// CustomDataItem returns the full key of a custom data item
-func (l *Layout) CustomDataItem(kind, key string) string {
-	return fmt.Sprintf(customDataItemFormat, kind, key)
+// CustomDataKindPrefix returns the prefix of all custom data kind
+func (l *Layout) CustomDataKindPrefix() string {
+	return customDataKindPrefix
 }
diff --git a/pkg/context/httpcontext.go b/pkg/context/httpcontext.go
index efdaf6e8..dc5198a9 100644
--- a/pkg/context/httpcontext.go
+++ b/pkg/context/httpcontext.go
@@ -100,7 +100,7 @@ type (
 		AddCookie(cookie *http.Cookie)
 
 		Body() io.Reader
-		SetBody(io.Reader)
+		SetBody(io.Reader, bool)
 
 		Std() *http.Request
 
diff --git a/pkg/context/httprequest.go b/pkg/context/httprequest.go
index ea93169b..01e28554 100644
--- a/pkg/context/httprequest.go
+++ b/pkg/context/httprequest.go
@@ -155,8 +155,8 @@ func (r *httpRequest) Body() io.Reader {
 	return r.body
 }
 
-func (r *httpRequest) SetBody(reader io.Reader) {
-	r.body = callbackreader.New(reader)
+func (r *httpRequest) SetBody(reader io.Reader, closePreviousReader bool) {
+	r.body.SetReader(reader, closePreviousReader)
 }
 
 func (r *httpRequest) Size() uint64 {
diff --git a/pkg/context/httptemplate.go b/pkg/context/httptemplate.go
index a0f4c88a..5b1d6a18 100644
--- a/pkg/context/httptemplate.go
+++ b/pkg/context/httptemplate.go
@@ -332,7 +332,7 @@ func saveReqBody(e *HTTPTemplate, filterName string, ctx HTTPContext) error {
 	}
 	e.Engine.SetDict(fmt.Sprintf(filterReqBody, filterName), string(bodyBuff.Bytes()))
 	// Reset back into Request's Body
-	ctx.Request().SetBody(bodyBuff)
+	ctx.Request().SetBody(bodyBuff, true)
 
 	return nil
 }
diff --git a/pkg/filter/certextractor/certextractor.go b/pkg/filter/certextractor/certextractor.go
new file mode 100644
index 00000000..352d0e74
--- /dev/null
+++ b/pkg/filter/certextractor/certextractor.go
@@ -0,0 +1,162 @@
+/*
+ * 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 certextractor
+
+import (
+	"crypto/x509/pkix"
+	"fmt"
+
+	httpcontext "github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+)
+
+const (
+	// Kind is the kind of CertExtractor.
+	Kind = "CertExtractor"
+)
+
+var results = []string{}
+
+func init() {
+	httppipeline.Register(&CertExtractor{})
+}
+
+type (
+	// CertExtractor extracts given field from TLS certificates and sets it to request headers.
+	CertExtractor struct {
+		filterSpec *httppipeline.FilterSpec
+		spec       *Spec
+
+		headerKey string
+	}
+
+	// Spec describes the CertExtractor.
+	Spec struct {
+		CertIndex int16  `yaml:"certIndex" jsonschema:"required"`
+		Target    string `yaml:"target" jsonschema:"required,enum=subject,enum=issuer"`
+		// Different field options listed here https://pkg.go.dev/crypto/x509/pkix#Name
+		Field     string `yaml:"field" jsonschema:"required,enum=Country,enum=Organization,enum=OrganizationalUnit,enum=Locality,enum=Province,enum=StreetAddress,enum=PostalCode,enum=SerialNumber,enum=CommonName"`
+		HeaderKey string `yaml:"headerKey" jsonschema:"required"`
+	}
+)
+
+// Validate is dummy as yaml rules already validate Spec.
+func (spec *Spec) Validate() error { return nil }
+
+// Kind returns the kind of CertExtractor.
+func (ce *CertExtractor) Kind() string {
+	return Kind
+}
+
+// DefaultSpec returns the default spec of CertExtractor.
+func (ce *CertExtractor) DefaultSpec() interface{} {
+	return &Spec{}
+}
+
+// Description returns the description of CertExtractor.
+func (ce *CertExtractor) Description() string {
+	return "CertExtractor extracts given field from TLS certificates and sets it to request headers."
+}
+
+// Results returns the results of CertExtractor.
+func (ce *CertExtractor) Results() []string {
+	return results
+}
+
+// Init initializes CertExtractor.
+func (ce *CertExtractor) Init(filterSpec *httppipeline.FilterSpec) {
+	ce.filterSpec, ce.spec = filterSpec, filterSpec.FilterSpec().(*Spec)
+
+	ce.headerKey = fmt.Sprintf("tls-%s-%s", ce.spec.Target, ce.spec.Field)
+	if ce.spec.HeaderKey != "" {
+		ce.headerKey = ce.spec.HeaderKey
+	}
+}
+
+// Inherit inherits previous generation of CertExtractor.
+func (ce *CertExtractor) Inherit(filterSpec *httppipeline.FilterSpec, previousGeneration httppipeline.Filter) {
+	previousGeneration.Close()
+	ce.Init(filterSpec)
+}
+
+// Close closes CertExtractor.
+func (ce *CertExtractor) Close() {}
+
+// Handle retrieves header values and sets request headers.
+func (ce *CertExtractor) Handle(ctx httpcontext.HTTPContext) string {
+	result := ce.handle(ctx)
+	return ctx.CallNextHandler(result)
+}
+
+// CertExtractor extracts given field from TLS certificates and sets it to request headers.
+func (ce *CertExtractor) handle(ctx httpcontext.HTTPContext) string {
+	r := ctx.Request()
+	connectionState := r.Std().TLS
+	if connectionState == nil {
+		return ""
+	}
+
+	certs := connectionState.PeerCertificates
+	if certs == nil || len(certs) < 1 {
+		return ""
+	}
+
+	n := int16(len(certs))
+	// positive ce.spec.CertIndex from the beginning, negative from the end
+	relativeIndex := ce.spec.CertIndex % n
+	index := (n + relativeIndex) % n
+	cert := certs[index]
+
+	var target pkix.Name
+	if ce.spec.Target == "subject" {
+		target = cert.Subject
+	} else {
+		target = cert.Issuer
+	}
+
+	var result []string
+	switch ce.spec.Field {
+	case "Country":
+		result = target.Country
+	case "Organization":
+		result = target.Organization
+	case "OrganizationalUnit":
+		result = target.OrganizationalUnit
+	case "Locality":
+		result = target.Locality
+	case "Province":
+		result = target.Province
+	case "StreetAddress":
+		result = target.StreetAddress
+	case "PostalCode":
+		result = target.PostalCode
+	case "SerialNumber":
+		result = append(result, target.SerialNumber)
+	case "CommonName":
+		result = append(result, target.CommonName)
+	}
+	for _, res := range result {
+		if res != "" {
+			r.Header().Add(ce.headerKey, res)
+		}
+	}
+	return ""
+}
+
+// Status returns status.
+func (ce *CertExtractor) Status() interface{} { return nil }
diff --git a/pkg/filter/corsadaptor/corsadaptor.go b/pkg/filter/corsadaptor/corsadaptor.go
index 1ba69f61..f16a8d93 100644
--- a/pkg/filter/corsadaptor/corsadaptor.go
+++ b/pkg/filter/corsadaptor/corsadaptor.go
@@ -48,13 +48,17 @@ type (
 		cors *cors.Cors
 	}
 
-	// Spec is describes of CORSAdaptor.
+	// Spec describes of CORSAdaptor.
 	Spec struct {
 		AllowedOrigins   []string `yaml:"allowedOrigins" jsonschema:"omitempty"`
 		AllowedMethods   []string `yaml:"allowedMethods" jsonschema:"omitempty,uniqueItems=true,format=httpmethod-array"`
 		AllowedHeaders   []string `yaml:"allowedHeaders" jsonschema:"omitempty"`
 		AllowCredentials bool     `yaml:"allowCredentials" jsonschema:"omitempty"`
 		ExposedHeaders   []string `yaml:"exposedHeaders" jsonschema:"omitempty"`
+		MaxAge           int      `yaml:"maxAge" jsonschema:"omitempty"`
+		// If true, handle requests with 'Origin' header. https://fetch.spec.whatwg.org/#http-requests
+		// By default, only CORS-preflight requests are handled.
+		SupportCORSRequest bool `yaml:"supportCORSRequest" jsonschema:"omitempty"`
 	}
 )
 
@@ -98,11 +102,16 @@ func (a *CORSAdaptor) reload() {
 		AllowedHeaders:   a.spec.AllowedHeaders,
 		AllowCredentials: a.spec.AllowCredentials,
 		ExposedHeaders:   a.spec.ExposedHeaders,
+		MaxAge:           a.spec.MaxAge,
 	})
 }
 
 // Handle handles simple cross-origin requests or directs.
 func (a *CORSAdaptor) Handle(ctx context.HTTPContext) string {
+	if a.spec.SupportCORSRequest {
+		result := a.handleCORS(ctx)
+		return ctx.CallNextHandler(result)
+	}
 	result := a.handle(ctx)
 	return ctx.CallNextHandler(result)
 }
@@ -119,6 +128,23 @@ func (a *CORSAdaptor) handle(ctx context.HTTPContext) string {
 	return ""
 }
 
+func (a *CORSAdaptor) handleCORS(ctx context.HTTPContext) string {
+	r := ctx.Request()
+	w := ctx.Response()
+	method := r.Method()
+	isCorsRequest := r.Header().Get("Origin") != ""
+	isPreflight := method == http.MethodOptions && r.Header().Get("Access-Control-Request-Method") != ""
+	// set CORS headers to response
+	a.cors.HandlerFunc(w.Std(), r.Std())
+	if !isCorsRequest {
+		return "" // next filter
+	}
+	if isPreflight {
+		return resultPreflighted // pipeline jumpIf skips following filters
+	}
+	return "" // next filter
+}
+
 // Status return status.
 func (a *CORSAdaptor) Status() interface{} {
 	return nil
diff --git a/pkg/filter/headerlookup/headerlookup.go b/pkg/filter/headerlookup/headerlookup.go
index 91706e1b..52b8da2a 100644
--- a/pkg/filter/headerlookup/headerlookup.go
+++ b/pkg/filter/headerlookup/headerlookup.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"regexp"
 	"strings"
 	"time"
 
@@ -55,6 +56,7 @@ type (
 		spec       *Spec
 		etcdPrefix string
 		headerKey  string
+		pathRegExp *regexp.Regexp
 
 		cache   *lru.Cache
 		cluster cluster.Cluster
@@ -71,9 +73,15 @@ type (
 	// Spec defines header key and etcd prefix that form etcd key like /custom-data/{etcdPrefix}/{headerKey's value}.
 	// This /custom-data/{etcdPrefix}/{headerKey's value} is retrieved from etcd and HeaderSetters extract keys from the
 	// from the retrieved etcd item.
+	// When PathRegExp is defined, PathRegExp is used with `regexp.FindStringSubmatch` to identify a group from path.
+	// The first captured group is appended to the etcd key in following format:
+	// /custom-data/{etcdPrefix}/{headerKey's value}-{regex group} . For example, for path
+	// "/api/bananas/33" and pathRegExp: "^/api/([a-z]+)/[0-9]*", the group "bananas" is extracted and etcd key is
+	// /custom-data/{etcdPrefix}/{headerKey's value}-bananas.
 	Spec struct {
 		HeaderKey     string              `yaml:"headerKey" jsonschema:"required"`
 		EtcdPrefix    string              `yaml:"etcdPrefix" jsonschema:"required"`
+		PathRegExp    string              `yaml:"pathRegExp" jsonschema:"omitempty"`
 		HeaderSetters []*HeaderSetterSpec `yaml:"headerSetters" jsonschema:"required"`
 	}
 )
@@ -97,6 +105,10 @@ func (spec Spec) Validate() error {
 			return fmt.Errorf("headerSetters[i].headerKey is required")
 		}
 	}
+
+	if _, err := regexp.Compile(spec.PathRegExp); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -130,6 +142,7 @@ func (hl *HeaderLookup) Init(filterSpec *httppipeline.FilterSpec) {
 	hl.headerKey = http.CanonicalHeaderKey(hl.spec.HeaderKey)
 	hl.cache, _ = lru.New(cacheSize)
 	hl.stopCtx, hl.cancel = context.WithCancel(context.Background())
+	hl.pathRegExp = regexp.MustCompile(hl.spec.PathRegExp)
 	hl.watchChanges()
 }
 
@@ -249,6 +262,12 @@ func (hl *HeaderLookup) handle(ctx httpcontext.HTTPContext) string {
 		logger.Warnf("request does not have header '%s'", hl.spec.HeaderKey)
 		return ""
 	}
+	if hl.spec.PathRegExp != "" {
+		path := ctx.Request().Path()
+		if match := hl.pathRegExp.FindStringSubmatch(path); match != nil && len(match) > 1 {
+			headerVal = headerVal + "-" + match[1]
+		}
+	}
 	headersToAdd, err := hl.lookup(headerVal)
 	if err != nil {
 		logger.Errorf(err.Error())
diff --git a/pkg/filter/headertojson/headertojson.go b/pkg/filter/headertojson/headertojson.go
index 8a745b00..23c34b41 100644
--- a/pkg/filter/headertojson/headertojson.go
+++ b/pkg/filter/headertojson/headertojson.go
@@ -203,6 +203,6 @@ func (h *HeaderToJSON) handle(ctx context.HTTPContext) string {
 	if err != nil {
 		return resultJSONEncodeDecodeErr
 	}
-	ctx.Request().SetBody(bytes.NewReader(bodyBytes))
+	ctx.Request().SetBody(bytes.NewReader(bodyBytes), true)
 	return ""
 }
diff --git a/pkg/filter/proxy/proxy.go b/pkg/filter/proxy/proxy.go
index b10fbbbd..ebe525b2 100644
--- a/pkg/filter/proxy/proxy.go
+++ b/pkg/filter/proxy/proxy.go
@@ -318,7 +318,7 @@ func (b *Proxy) Handle(ctx context.HTTPContext) (result string) {
 func (b *Proxy) handle(ctx context.HTTPContext) (result string) {
 	if b.mirrorPool != nil && b.mirrorPool.filter.Filter(ctx) {
 		primaryBody, secondaryBody := newPrimarySecondaryReader(ctx.Request().Body())
-		ctx.Request().SetBody(primaryBody)
+		ctx.Request().SetBody(primaryBody, false)
 
 		wg := &sync.WaitGroup{}
 		wg.Add(1)
diff --git a/pkg/filter/remotefilter/remotefilter.go b/pkg/filter/remotefilter/remotefilter.go
index 6e8152d4..4af1e6c0 100644
--- a/pkg/filter/remotefilter/remotefilter.go
+++ b/pkg/filter/remotefilter/remotefilter.go
@@ -309,7 +309,7 @@ func (rf *RemoteFilter) unmarshalHTTPContext(buff []byte, ctx context.HTTPContex
 	r.SetPath(re.Path)
 	r.SetQuery(re.Query)
 	r.Header().Reset(re.Header)
-	r.SetBody(bytes.NewReader(re.Body))
+	r.SetBody(bytes.NewReader(re.Body), true)
 
 	if we == nil {
 		return
diff --git a/pkg/filter/requestadaptor/requestadaptor.go b/pkg/filter/requestadaptor/requestadaptor.go
index acae51d6..3dde4a75 100644
--- a/pkg/filter/requestadaptor/requestadaptor.go
+++ b/pkg/filter/requestadaptor/requestadaptor.go
@@ -19,6 +19,8 @@ package requestadaptor
 
 import (
 	"bytes"
+	"compress/gzip"
+	"io"
 
 	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
@@ -31,9 +33,12 @@ import (
 const (
 	// Kind is the kind of RequestAdaptor.
 	Kind = "RequestAdaptor"
+
+	resultDecompressFail = "decompressFail"
+	resultCompressFail   = "compressFail"
 )
 
-var results = []string{}
+var results = []string{resultDecompressFail, resultCompressFail}
 
 func init() {
 	httppipeline.Register(&RequestAdaptor{})
@@ -50,11 +55,13 @@ type (
 
 	// Spec is HTTPAdaptor Spec.
 	Spec struct {
-		Host   string                `yaml:"host" jsonschema:"omitempty"`
-		Method string                `yaml:"method" jsonschema:"omitempty,format=httpmethod"`
-		Path   *pathadaptor.Spec     `yaml:"path,omitempty" jsonschema:"omitempty"`
-		Header *httpheader.AdaptSpec `yaml:"header,omitempty" jsonschema:"omitempty"`
-		Body   string                `yaml:"body" jsonschema:"omitempty"`
+		Host       string                `yaml:"host" jsonschema:"omitempty"`
+		Method     string                `yaml:"method" jsonschema:"omitempty,format=httpmethod"`
+		Path       *pathadaptor.Spec     `yaml:"path,omitempty" jsonschema:"omitempty"`
+		Header     *httpheader.AdaptSpec `yaml:"header,omitempty" jsonschema:"omitempty"`
+		Body       string                `yaml:"body" jsonschema:"omitempty"`
+		Compress   string                `yaml:"compress" jsonschema:"omitempty"`
+		Decompress string                `yaml:"decompress" jsonschema:"omitempty"`
 	}
 )
 
@@ -81,6 +88,18 @@ func (ra *RequestAdaptor) Results() []string {
 // Init initializes RequestAdaptor.
 func (ra *RequestAdaptor) Init(filterSpec *httppipeline.FilterSpec) {
 	ra.filterSpec, ra.spec = filterSpec, filterSpec.FilterSpec().(*Spec)
+	if ra.spec.Decompress != "" && ra.spec.Decompress != "gzip" {
+		panic("RequestAdaptor only support decompress type of gzip")
+	}
+	if ra.spec.Compress != "" && ra.spec.Compress != "gzip" {
+		panic("RequestAdaptor only support decompress type of gzip")
+	}
+	if ra.spec.Compress != "" && ra.spec.Decompress != "" {
+		panic("RequestAdaptor can only do compress or decompress for given request body, not both")
+	}
+	if ra.spec.Body != "" && ra.spec.Decompress != "" {
+		panic("No need to decompress when body is specified in RequestAdaptor spec")
+	}
 	ra.reload()
 }
 
@@ -132,11 +151,12 @@ func (ra *RequestAdaptor) handle(ctx context.HTTPContext) string {
 				logger.Errorf("BUG request render body failed, template %s, err %v",
 					ra.spec.Body, err)
 			} else {
-				ctx.Request().SetBody(bytes.NewReader([]byte(body)))
+				ctx.Request().SetBody(bytes.NewReader([]byte(body)), true)
 			}
 		} else {
-			ctx.Request().SetBody(bytes.NewReader([]byte(ra.spec.Body)))
+			ctx.Request().SetBody(bytes.NewReader([]byte(ra.spec.Body)), true)
 		}
+		ctx.Request().Header().Del("Content-Encoding")
 	}
 
 	if len(ra.spec.Host) != 0 {
@@ -151,6 +171,58 @@ func (ra *RequestAdaptor) handle(ctx context.HTTPContext) string {
 			ctx.Request().SetHost(ra.spec.Host)
 		}
 	}
+
+	if ra.spec.Compress != "" {
+		res := ra.processCompress(ctx)
+		if res != "" {
+			return res
+		}
+	}
+
+	if ra.spec.Decompress != "" {
+		res := ra.processDecompress(ctx)
+		if res != "" {
+			return res
+		}
+	}
+	return ""
+}
+
+func (ra *RequestAdaptor) processCompress(ctx context.HTTPContext) string {
+	encoding := ctx.Request().Header().Get("Content-Encoding")
+	if encoding != "" {
+		return ""
+	}
+	var buf bytes.Buffer
+	gw := gzip.NewWriter(&buf)
+
+	_, err := io.Copy(gw, ctx.Request().Body())
+	if err != nil {
+		logger.Errorf("compress request body failed, %v", err)
+		return resultCompressFail
+	}
+	gw.Close()
+
+	ctx.Request().SetBody(&buf, true)
+	ctx.Request().Header().Set("Content-Encoding", "gzip")
+	return ""
+}
+
+func (ra *RequestAdaptor) processDecompress(ctx context.HTTPContext) string {
+	encoding := ctx.Request().Header().Get("Content-Encoding")
+	if ra.spec.Decompress == "gzip" && encoding == "gzip" {
+		reader, err := gzip.NewReader(ctx.Request().Body())
+		if err != nil {
+			return resultDecompressFail
+		}
+		defer reader.Close()
+		data, err := io.ReadAll(reader)
+		if err != nil {
+			return resultDecompressFail
+		}
+		ctx.Request().SetBody(bytes.NewReader(data), true)
+		ctx.Request().Header().Del("Content-Encoding")
+	}
 	return ""
 }
 
diff --git a/pkg/filter/retryer/retryer.go b/pkg/filter/retryer/retryer.go
index 988f9a86..949887ed 100644
--- a/pkg/filter/retryer/retryer.go
+++ b/pkg/filter/retryer/retryer.go
@@ -179,7 +179,7 @@ func (r *Retryer) handle(ctx context.HTTPContext, u *URLRule) string {
 	data, _ := io.ReadAll(ctx.Request().Body())
 	for {
 		attempt++
-		ctx.Request().SetBody(bytes.NewReader(data))
+		ctx.Request().SetBody(bytes.NewReader(data), true)
 
 		result := ctx.CallNextHandler("")
 
diff --git a/pkg/object/meshcontroller/api/api_customresource.go b/pkg/object/meshcontroller/api/api_customresource.go
index c04c96d3..b209cd6c 100644
--- a/pkg/object/meshcontroller/api/api_customresource.go
+++ b/pkg/object/meshcontroller/api/api_customresource.go
@@ -124,7 +124,7 @@ func (a *API) saveCustomResourceKind(w http.ResponseWriter, r *http.Request, upd
 		return err
 	}
 
-	a.service.PutCustomResourceKind(kind)
+	a.service.PutCustomResourceKind(kind, update)
 	return nil
 }
 
@@ -163,14 +163,14 @@ func (a *API) deleteCustomResourceKind(w http.ResponseWriter, r *http.Request) {
 func (a *API) listAllCustomResources(w http.ResponseWriter, r *http.Request) {
 	resources := a.service.ListCustomResources("")
 	sort.Slice(resources, func(i, j int) bool {
-		k1, k2 := resources[i].Kind(), resources[j].Kind()
+		k1, k2 := resources[i].GetString("kind"), resources[j].GetString("kind")
 		if k1 < k2 {
 			return true
 		}
 		if k1 > k2 {
 			return false
 		}
-		return resources[i].Name() < resources[j].Name()
+		return resources[i].GetString("name") < resources[j].GetString("name")
 	})
 
 	w.Header().Set("Content-Type", "application/json")
@@ -194,7 +194,7 @@ func (a *API) listCustomResources(w http.ResponseWriter, r *http.Request) {
 
 	resources := a.service.ListCustomResources(kind)
 	sort.Slice(resources, func(i, j int) bool {
-		return resources[i].Name() < resources[j].Name()
+		return resources[i].GetString("name") < resources[j].GetString("name")
 	})
 
 	w.Header().Set("Content-Type", "application/json")
@@ -235,15 +235,15 @@ func (a *API) getCustomResource(w http.ResponseWriter, r *http.Request) {
 }
 
 func (a *API) saveCustomResource(w http.ResponseWriter, r *http.Request, update bool) error {
-	resource := &spec.CustomResource{}
-	err := json.NewDecoder(r.Body).Decode(resource)
+	resource := spec.CustomResource{}
+	err := json.NewDecoder(r.Body).Decode(&resource)
 	if err != nil {
 		err = fmt.Errorf("unmarshal custom resource failed: %v", err)
 		api.HandleAPIError(w, r, http.StatusBadRequest, err)
 		return err
 	}
 
-	kind, name := resource.Kind(), resource.Name()
+	kind, name := resource.GetString("kind"), resource.GetString("name")
 	if kind == "" {
 		err = fmt.Errorf("kind cannot be empty")
 		api.HandleAPIError(w, r, http.StatusBadRequest, err)
@@ -292,7 +292,7 @@ func (a *API) saveCustomResource(w http.ResponseWriter, r *http.Request, update
 		return err
 	}
 
-	a.service.PutCustomResource(resource)
+	a.service.PutCustomResource(resource, update)
 	return nil
 }
 
@@ -342,7 +342,7 @@ func (a *API) watchCustomResources(w http.ResponseWriter, r *http.Request) {
 	logger.Infof("begin watch custom resources of kind '%s'", kind)
 
 	w.Header().Set("Content-type", "application/octet-stream")
-	a.service.WatchCustomResource(r.Context(), kind, func(resources []*spec.CustomResource) {
+	a.service.WatchCustomResource(r.Context(), kind, func(resources []spec.CustomResource) {
 		err = json.NewEncoder(w).Encode(resources)
 		if err != nil {
 			logger.Errorf("marshal custom resource failed: %v", err)
diff --git a/pkg/object/meshcontroller/layout/layout.go b/pkg/object/meshcontroller/layout/layout.go
index 05965fad..c26310c8 100644
--- a/pkg/object/meshcontroller/layout/layout.go
+++ b/pkg/object/meshcontroller/layout/layout.go
@@ -53,10 +53,10 @@ const (
 	trafficTargetPrefix  = "/mesh/traffic-targets/"
 
 	customResourceKindPrefix = "/mesh/custom-resource-kinds/"
-	customResourceKind       = "/mesh/custom-resource-kinds/%s/" // +kind
+	customResourceKind       = "/mesh/custom-resource-kinds/%s" // +kind
 	allCustomResourcePrefix  = "/mesh/custom-resources/"
-	customResourcePrefix     = "/mesh/custom-resources/%s/"    // +kind
-	customResource           = "/mesh/custom-resources/%s/%s/" // +kind +name
+	customResourcePrefix     = "/mesh/custom-resources/%s/"   // +kind
+	customResource           = "/mesh/custom-resources/%s/%s" // +kind +name
 
 	globalCanaryHeaders = "/mesh/canary-headers"
 
diff --git a/pkg/object/meshcontroller/service/service.go b/pkg/object/meshcontroller/service/service.go
index e7bc7a4d..ea052e13 100644
--- a/pkg/object/meshcontroller/service/service.go
+++ b/pkg/object/meshcontroller/service/service.go
@@ -26,6 +26,7 @@ import (
 	"gopkg.in/yaml.v2"
 
 	"github.com/megaease/easegress/pkg/api"
+	"github.com/megaease/easegress/pkg/cluster/customdata"
 	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/object/meshcontroller/layout"
 	"github.com/megaease/easegress/pkg/object/meshcontroller/spec"
@@ -41,15 +42,19 @@ type (
 		spec      *spec.Admin
 
 		store storage.Storage
+		cds   *customdata.Store
 	}
 )
 
 // New creates a service with spec
 func New(superSpec *supervisor.Spec) *Service {
+	kindPrefix := layout.CustomResourceKindPrefix()
+	dataPrefix := layout.AllCustomResourcePrefix()
 	s := &Service{
 		superSpec: superSpec,
 		spec:      superSpec.ObjectSpec().(*spec.Admin),
 		store:     storage.New(superSpec.Name(), superSpec.Super().Cluster()),
+		cds:       customdata.NewStore(superSpec.Super().Cluster(), kindPrefix, dataPrefix),
 	}
 
 	return s
@@ -657,163 +662,77 @@ func (s *Service) DeleteIngressSpec(ingressName string) {
 
 // ListCustomResourceKinds lists custom resource kinds
 func (s *Service) ListCustomResourceKinds() []*spec.CustomResourceKind {
-	kvs, err := s.store.GetRawPrefix(layout.CustomResourceKindPrefix())
+	kinds, err := s.cds.ListKinds()
 	if err != nil {
-		api.ClusterPanic(err)
-	}
-
-	kinds := []*spec.CustomResourceKind{}
-	for _, v := range kvs {
-		kind := &spec.CustomResourceKind{}
-		err := yaml.Unmarshal(v.Value, kind)
-		if err != nil {
-			logger.Errorf("BUG: unmarshal %s to yaml failed: %v", v, err)
-			continue
-		}
-		kinds = append(kinds, kind)
+		panic(err)
 	}
-
 	return kinds
 }
 
 // DeleteCustomResourceKind deletes a custom resource kind
 func (s *Service) DeleteCustomResourceKind(kind string) {
-	err := s.store.Delete(layout.CustomResourceKindKey(kind))
+	err := s.cds.DeleteKind(kind)
 	if err != nil {
-		api.ClusterPanic(err)
+		panic(err)
 	}
 }
 
 // GetCustomResourceKind gets custom resource kind with its name
 func (s *Service) GetCustomResourceKind(name string) *spec.CustomResourceKind {
-	kvs, err := s.store.GetRaw(layout.CustomResourceKindKey(name))
+	kind, err := s.cds.GetKind(name)
 	if err != nil {
-		api.ClusterPanic(err)
-	}
-
-	if kvs == nil {
-		return nil
+		panic(err)
 	}
-
-	kind := &spec.CustomResourceKind{}
-	err = yaml.Unmarshal(kvs.Value, kind)
-	if err != nil {
-		panic(fmt.Errorf("BUG: unmarshal %s to yaml failed: %v", string(kvs.Value), err))
-	}
-
 	return kind
 }
 
 // PutCustomResourceKind writes the custom resource kind to storage.
-func (s *Service) PutCustomResourceKind(kind *spec.CustomResourceKind) {
-	buff, err := yaml.Marshal(kind)
+func (s *Service) PutCustomResourceKind(kind *spec.CustomResourceKind, update bool) {
+	err := s.cds.PutKind(kind, update)
 	if err != nil {
 		panic(fmt.Errorf("BUG: marshal %#v to yaml failed: %v", kind, err))
 	}
-
-	err = s.store.Put(layout.CustomResourceKindKey(kind.Name), string(buff))
-	if err != nil {
-		api.ClusterPanic(err)
-	}
 }
 
 // ListCustomResources lists custom resources of specified kind.
 // if kind is empty, it returns custom objects of all kinds.
-func (s *Service) ListCustomResources(kind string) []*spec.CustomResource {
-	prefix := layout.AllCustomResourcePrefix()
-	if kind != "" {
-		prefix = layout.CustomResourcePrefix(kind)
-	}
-	kvs, err := s.store.GetRawPrefix(prefix)
+func (s *Service) ListCustomResources(kind string) []spec.CustomResource {
+	resources, err := s.cds.ListData(kind)
 	if err != nil {
-		api.ClusterPanic(err)
-	}
-
-	resources := []*spec.CustomResource{}
-	for _, v := range kvs {
-		resource := &spec.CustomResource{}
-		err := yaml.Unmarshal(v.Value, resource)
-		if err != nil {
-			logger.Errorf("BUG: unmarshal %s to yaml failed: %v", v, err)
-			continue
-		}
-		resources = append(resources, resource)
+		panic(err)
 	}
-
 	return resources
 }
 
 // DeleteCustomResource deletes a custom resource
 func (s *Service) DeleteCustomResource(kind, name string) {
-	err := s.store.Delete(layout.CustomResourceKey(kind, name))
+	err := s.cds.DeleteData(kind, name)
 	if err != nil {
-		api.ClusterPanic(err)
+		panic(err)
 	}
 }
 
 // GetCustomResource gets custom resource with its kind & name
-func (s *Service) GetCustomResource(kind, name string) *spec.CustomResource {
-	kvs, err := s.store.GetRaw(layout.CustomResourceKey(kind, name))
+func (s *Service) GetCustomResource(kind, name string) spec.CustomResource {
+	resource, err := s.cds.GetData(kind, name)
 	if err != nil {
-		api.ClusterPanic(err)
+		panic(err)
 	}
-
-	if kvs == nil {
-		return nil
-	}
-
-	resource := &spec.CustomResource{}
-	err = yaml.Unmarshal(kvs.Value, resource)
-	if err != nil {
-		panic(fmt.Errorf("BUG: unmarshal %s to yaml failed: %v", string(kvs.Value), err))
-	}
-
 	return resource
 }
 
 // PutCustomResource writes the custom resource kind to storage.
-func (s *Service) PutCustomResource(obj *spec.CustomResource) {
-	buff, err := yaml.Marshal(obj)
+func (s *Service) PutCustomResource(resource spec.CustomResource, update bool) {
+	kind := resource.GetString("kind")
+	_, err := s.cds.PutData(kind, resource, update)
 	if err != nil {
-		panic(fmt.Errorf("BUG: marshal %#v to yaml failed: %v", obj, err))
-	}
-
-	err = s.store.Put(layout.CustomResourceKey(obj.Kind(), obj.Name()), string(buff))
-	if err != nil {
-		api.ClusterPanic(err)
+		panic(err)
 	}
 }
 
 // WatchCustomResource watches custom resources of the specified kind
-func (s *Service) WatchCustomResource(ctx context.Context, kind string, onChange func([]*spec.CustomResource)) error {
-	syncer, err := s.store.Syncer()
-	if err != nil {
-		return err
-	}
-
-	prefix := layout.CustomResourcePrefix(kind)
-	ch, err := syncer.SyncRawPrefix(prefix)
-	if err != nil {
-		return err
-	}
-
-	for {
-		select {
-		case <-ctx.Done():
-			syncer.Close()
-			return nil
-		case m := <-ch:
-			resources := make([]*spec.CustomResource, 0, len(m))
-			for _, v := range m {
-				resource := &spec.CustomResource{}
-				err = yaml.Unmarshal(v.Value, resource)
-				if err == nil {
-					resources = append(resources, resource)
-				}
-			}
-			onChange(resources)
-		}
-	}
+func (s *Service) WatchCustomResource(ctx context.Context, kind string, onChange func([]spec.CustomResource)) error {
+	return s.cds.Watch(ctx, kind, onChange)
 }
 
 // ListHTTPRouteGroups lists HTTP route groups
diff --git a/pkg/object/meshcontroller/spec/spec.go b/pkg/object/meshcontroller/spec/spec.go
index 93cb0532..acc6d314 100644
--- a/pkg/object/meshcontroller/spec/spec.go
+++ b/pkg/object/meshcontroller/spec/spec.go
@@ -24,6 +24,7 @@ import (
 
 	"gopkg.in/yaml.v2"
 
+	"github.com/megaease/easegress/pkg/cluster/customdata"
 	"github.com/megaease/easegress/pkg/filter/circuitbreaker"
 	"github.com/megaease/easegress/pkg/filter/meshadaptor"
 	"github.com/megaease/easegress/pkg/filter/mock"
@@ -360,19 +361,11 @@ type (
 		httppipeline.Spec `yaml:",inline"`
 	}
 
-	// DynamicObject defines a dynamic object which is a map of string to interface{}.
-	// The value of this map could also be a dynamic object, but in this case, its type
-	// must be `map[string]interface{}`, and should not be `map[interface{}]interface{}`.
-	DynamicObject map[string]interface{}
-
 	// CustomResourceKind defines the spec of a custom resource kind
-	CustomResourceKind struct {
-		Name       string        `yaml:"name" jsonschema:"required"`
-		JSONSchema DynamicObject `yaml:"jsonSchema" jsonschema:"omitempty"`
-	}
+	CustomResourceKind = customdata.Kind
 
 	// CustomResource defines the spec of a custom resource
-	CustomResource DynamicObject
+	CustomResource = customdata.Data
 
 	// HTTPMatch defines an individual route for HTTP traffic
 	HTTPMatch struct {
@@ -464,64 +457,6 @@ func (tr *TrafficRules) Clone() *TrafficRules {
 	}
 }
 
-// UnmarshalYAML implements yaml.Unmarshaler
-// the type of a DynamicObject field could be `map[interface{}]interface{}` if it is
-// unmarshaled from yaml, but some packages, like the standard json package could not
-// handle this type, so it must be converted to `map[string]interface{}`.
-func (do *DynamicObject) UnmarshalYAML(unmarshal func(interface{}) error) error {
-	m := map[string]interface{}{}
-	if err := unmarshal(&m); err != nil {
-		return err
-	}
-
-	var convert func(interface{}) interface{}
-	convert = func(src interface{}) interface{} {
-		switch x := src.(type) {
-		case map[interface{}]interface{}:
-			x2 := map[string]interface{}{}
-			for k, v := range x {
-				x2[k.(string)] = convert(v)
-			}
-			return x2
-		case []interface{}:
-			x2 := make([]interface{}, len(x))
-			for i, v := range x {
-				x2[i] = convert(v)
-			}
-			return x2
-		}
-		return src
-	}
-
-	for k, v := range m {
-		m[k] = convert(v)
-	}
-	*do = m
-
-	return nil
-}
-
-// Name returns the 'name' field of the custom resource
-func (cr CustomResource) Name() string {
-	if v, ok := cr["name"].(string); ok {
-		return v
-	}
-	return ""
-}
-
-// Kind returns the 'kind' field of the custom resource
-func (cr CustomResource) Kind() string {
-	if v, ok := cr["kind"].(string); ok {
-		return v
-	}
-	return ""
-}
-
-// UnmarshalYAML implements yaml.Unmarshaler
-func (cr *CustomResource) UnmarshalYAML(unmarshal func(interface{}) error) error {
-	return (*DynamicObject)(cr).UnmarshalYAML(unmarshal)
-}
-
 // Validate validates Spec.
 func (a Admin) Validate() error {
 	switch a.RegistryType {
diff --git a/pkg/registry/registry.go b/pkg/registry/registry.go
index 094f4731..547af056 100644
--- a/pkg/registry/registry.go
+++ b/pkg/registry/registry.go
@@ -22,6 +22,7 @@ import (
 	// Filters
 	_ "github.com/megaease/easegress/pkg/filter/apiaggregator"
 	_ "github.com/megaease/easegress/pkg/filter/bridge"
+	_ "github.com/megaease/easegress/pkg/filter/certextractor"
 	_ "github.com/megaease/easegress/pkg/filter/circuitbreaker"
 	_ "github.com/megaease/easegress/pkg/filter/connectcontrol"
 	_ "github.com/megaease/easegress/pkg/filter/corsadaptor"
diff --git a/pkg/util/callbackreader/callbackreader.go b/pkg/util/callbackreader/callbackreader.go
index acc9ee5e..69d452cf 100644
--- a/pkg/util/callbackreader/callbackreader.go
+++ b/pkg/util/callbackreader/callbackreader.go
@@ -88,3 +88,15 @@ func (cr *CallbackReader) Close() error {
 
 	return nil
 }
+
+// SetReader replace previous reader with new reader. If closePreviousReader set to true, it will close
+// previous reader.
+func (cr *CallbackReader) SetReader(reader io.Reader, closePreviousReader bool) {
+	if closePreviousReader {
+		closer, ok := cr.reader.(io.Closer)
+		if ok {
+			closer.Close()
+		}
+	}
+	cr.reader = reader
+}
diff --git a/pkg/util/dynamicobject/dynamicobject.go b/pkg/util/dynamicobject/dynamicobject.go
new file mode 100644
index 00000000..8a737c85
--- /dev/null
+++ b/pkg/util/dynamicobject/dynamicobject.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 dynamicobject
+
+// DynamicObject defines a dynamic object which is a map of string to interface{}.
+// The value of this map could also be a dynamic object, but in this case, its type
+// must be `map[string]interface{}`, and should not be `map[interface{}]interface{}`.
+type DynamicObject map[string]interface{}
+
+// UnmarshalYAML implements yaml.Unmarshaler
+// the type of a DynamicObject field could be `map[interface{}]interface{}` if it is
+// unmarshaled from yaml, but some packages, like the standard json package could not
+// handle this type, so it must be converted to `map[string]interface{}`.
+func (do *DynamicObject) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	m := map[string]interface{}{}
+	if err := unmarshal(&m); err != nil {
+		return err
+	}
+
+	var convert func(interface{}) interface{}
+	convert = func(src interface{}) interface{} {
+		switch x := src.(type) {
+		case map[interface{}]interface{}:
+			x2 := map[string]interface{}{}
+			for k, v := range x {
+				x2[k.(string)] = convert(v)
+			}
+			return x2
+		case []interface{}:
+			x2 := make([]interface{}, len(x))
+			for i, v := range x {
+				x2[i] = convert(v)
+			}
+			return x2
+		}
+		return src
+	}
+
+	for k, v := range m {
+		m[k] = convert(v)
+	}
+	*do = m
+
+	return nil
+}
+
+// Get gets the value of field 'field'
+func (do DynamicObject) Get(field string) interface{} {
+	return do[field]
+}
+
+// GetString gets the value of field 'field' as string
+func (do DynamicObject) GetString(field string) string {
+	if v, ok := do[field].(string); ok {
+		return v
+	}
+	return ""
+}
+
+// Set set the value of field 'field' to 'value'
+func (do *DynamicObject) Set(field string, value interface{}) {
+	(*do)[field] = value
+}


Test patch:
diff --git a/pkg/cluster/clustertest/cluster.go b/pkg/cluster/clustertest/cluster.go
new file mode 100644
index 00000000..abb742aa
--- /dev/null
+++ b/pkg/cluster/clustertest/cluster.go
@@ -0,0 +1,264 @@
+/*
+ * 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 clustertest
+
+import (
+	"sync"
+	"time"
+
+	"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"
+)
+
+// MockedCluster defines a mocked cluster
+type MockedCluster struct {
+	MockedIsLeader               func() bool
+	MockedLayout                 func() *cluster.Layout
+	MockedGet                    func(key string) (*string, error)
+	MockedGetPrefix              func(prefix string) (map[string]string, error)
+	MockedGetRaw                 func(key string) (*mvccpb.KeyValue, error)
+	MockedGetRawPrefix           func(prefix string) (map[string]*mvccpb.KeyValue, error)
+	MockedGetWithOp              func(key string, ops ...cluster.ClientOp) (map[string]string, error)
+	MockedPut                    func(key, value string) error
+	MockedPutUnderLease          func(key, value string) error
+	MockedPutAndDelete           func(map[string]*string) error
+	MockedPutAndDeleteUnderLease func(map[string]*string) error
+	MockedDelete                 func(key string) error
+	MockedDeletePrefix           func(prefix string) error
+	MockedSTM                    func(apply func(concurrency.STM) error) error
+	MockedWatcher                func() (cluster.Watcher, error)
+	MockedSyncer                 func(pullInterval time.Duration) (*cluster.Syncer, error)
+	MockedMutex                  func(name string) (cluster.Mutex, error)
+	MockedCloseServer            func(wg *sync.WaitGroup)
+	MockedStartServer            func() (chan struct{}, chan struct{}, error)
+	MockedClose                  func(wg *sync.WaitGroup)
+	MockedPurgeMember            func(member string) error
+}
+
+// NewMockedCluster creates a new mocked cluster
+func NewMockedCluster() *MockedCluster {
+	return &MockedCluster{}
+}
+
+// IsLeader implements interface function IsLeader
+func (mc *MockedCluster) IsLeader() bool {
+	if mc.MockedIsLeader != nil {
+		return mc.MockedIsLeader()
+	}
+	return true
+}
+
+// Layout implements interface function Layout
+func (mc *MockedCluster) Layout() *cluster.Layout {
+	if mc.MockedLayout != nil {
+		return mc.MockedLayout()
+	}
+	return nil
+}
+
+// Get implements interface function Get
+func (mc *MockedCluster) Get(key string) (*string, error) {
+	if mc.MockedGet != nil {
+		return mc.MockedGet(key)
+	}
+	return nil, nil
+}
+
+// GetPrefix implements interface function GetPrefix
+func (mc *MockedCluster) GetPrefix(prefix string) (map[string]string, error) {
+	if mc.MockedGetPrefix != nil {
+		return mc.MockedGetPrefix(prefix)
+	}
+	return nil, nil
+}
+
+// GetRaw implements interface function GetRaw
+func (mc *MockedCluster) GetRaw(key string) (*mvccpb.KeyValue, error) {
+	if mc.MockedGetRaw != nil {
+		return mc.MockedGetRaw(key)
+	}
+	return nil, nil
+}
+
+// GetRawPrefix implements interface function GetRawPrefix
+func (mc *MockedCluster) GetRawPrefix(prefix string) (map[string]*mvccpb.KeyValue, error) {
+	if mc.MockedGetRawPrefix != nil {
+		return mc.MockedGetRawPrefix(prefix)
+	}
+	return nil, nil
+}
+
+// GetWithOp implements interface function GetWithOp
+func (mc *MockedCluster) GetWithOp(key string, ops ...cluster.ClientOp) (map[string]string, error) {
+	if mc.MockedGetWithOp != nil {
+		return mc.MockedGetWithOp(key, ops...)
+	}
+	return nil, nil
+}
+
+// Put implements interface function Put
+func (mc *MockedCluster) Put(key, value string) error {
+	if mc.MockedPut != nil {
+		return mc.MockedPut(key, value)
+	}
+	return nil
+}
+
+// PutUnderLease implements interface function PutUnderLease
+func (mc *MockedCluster) PutUnderLease(key, value string) error {
+	if mc.MockedPutUnderLease != nil {
+		return mc.MockedPutUnderLease(key, value)
+	}
+	return nil
+}
+
+// PutAndDelete implements interface function PutAndDelete
+func (mc *MockedCluster) PutAndDelete(m map[string]*string) error {
+	if mc.MockedPutAndDelete != nil {
+		return mc.MockedPutAndDelete(m)
+	}
+	return nil
+}
+
+// PutAndDeleteUnderLease implements interface function PutAndDeleteUnderLease
+func (mc *MockedCluster) PutAndDeleteUnderLease(m map[string]*string) error {
+	if mc.MockedPutAndDeleteUnderLease != nil {
+		return mc.MockedPutAndDeleteUnderLease(m)
+	}
+	return nil
+}
+
+// Delete implements interface function Delete
+func (mc *MockedCluster) Delete(key string) error {
+	if mc.MockedDelete != nil {
+		return mc.MockedDelete(key)
+	}
+	return nil
+}
+
+// DeletePrefix implements interface function DeletePrefix
+func (mc *MockedCluster) DeletePrefix(prefix string) error {
+	if mc.MockedDeletePrefix != nil {
+		return mc.MockedDeletePrefix(prefix)
+	}
+	return nil
+}
+
+// STM implements interface function STM
+func (mc *MockedCluster) STM(apply func(concurrency.STM) error) error {
+	if mc.MockedSTM != nil {
+		return mc.MockedSTM(apply)
+	}
+	return nil
+}
+
+// Watcher implements interface function Watcher
+func (mc *MockedCluster) Watcher() (cluster.Watcher, error) {
+	if mc.MockedWatcher != nil {
+		return mc.MockedWatcher()
+	}
+	return nil, nil
+}
+
+// Syncer implements interface function Syncer
+func (mc *MockedCluster) Syncer(pullInterval time.Duration) (*cluster.Syncer, error) {
+	if mc.MockedSyncer != nil {
+		return mc.MockedSyncer(pullInterval)
+	}
+	return nil, nil
+}
+
+// Mutex implements interface function Mutex
+func (mc *MockedCluster) Mutex(name string) (cluster.Mutex, error) {
+	if mc.MockedMutex != nil {
+		return mc.MockedMutex(name)
+	}
+	return nil, nil
+}
+
+// CloseServer implements interface function CloseServer
+func (mc *MockedCluster) CloseServer(wg *sync.WaitGroup) {
+	if mc.MockedCloseServer != nil {
+		mc.MockedCloseServer(wg)
+	}
+}
+
+// StartServer implements interface function StartServer
+func (mc *MockedCluster) StartServer() (chan struct{}, chan struct{}, error) {
+	if mc.MockedStartServer != nil {
+		return mc.MockedStartServer()
+	}
+	return nil, nil, nil
+}
+
+// Close implements interface function Close
+func (mc *MockedCluster) Close(wg *sync.WaitGroup) {
+	if mc.MockedClose != nil {
+		mc.MockedClose(wg)
+	}
+}
+
+// PurgeMember implements interface function PurgeMember
+func (mc *MockedCluster) PurgeMember(member string) error {
+	if mc.MockedPurgeMember != nil {
+		return mc.MockedPurgeMember(member)
+	}
+	return nil
+}
+
+// MockedSTM is a mocked cocurrency.STM
+type MockedSTM struct {
+	// embed concurrency.STM for commit & reset
+	concurrency.STM
+	MockedGet func(key ...string) string
+	MockedPut func(key, val string, opts ...clientv3.OpOption)
+	MockedRev func(key string) int64
+	MockedDel func(key string)
+}
+
+// Get implements STM.Get
+func (stm *MockedSTM) Get(key ...string) string {
+	if stm.MockedGet != nil {
+		return stm.MockedGet(key...)
+	}
+	return ""
+}
+
+// Put implements STM.Put
+func (stm *MockedSTM) Put(key, val string, opts ...clientv3.OpOption) {
+	if stm.MockedPut != nil {
+		stm.MockedPut(key, val, opts...)
+	}
+}
+
+// Rev implements STM.Rev
+func (stm *MockedSTM) Rev(key string) int64 {
+	if stm.MockedRev != nil {
+		return stm.MockedRev(key)
+	}
+	return 0
+}
+
+// Del implements STM.Del
+func (stm *MockedSTM) Del(key string) {
+	if stm.MockedRev != nil {
+		stm.MockedDel(key)
+	}
+}
diff --git a/pkg/cluster/customdata/customdata_test.go b/pkg/cluster/customdata/customdata_test.go
new file mode 100644
index 00000000..e5e9695f
--- /dev/null
+++ b/pkg/cluster/customdata/customdata_test.go
@@ -0,0 +1,580 @@
+/*
+ * 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 customdata
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/megaease/easegress/pkg/cluster"
+	"github.com/megaease/easegress/pkg/cluster/clustertest"
+	"github.com/megaease/easegress/pkg/util/dynamicobject"
+
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	"go.etcd.io/etcd/client/v3/concurrency"
+)
+
+func TestDataID(t *testing.T) {
+	k := Kind{}
+	data := Data{}
+	data["name"] = "abc"
+	data["key"] = "key"
+
+	if id := k.dataID(data); id != "abc" {
+		t.Errorf("data ID should be 'abc` instead of %q", id)
+	}
+
+	k.IDField = "key"
+	if id := k.dataID(data); id != "key" {
+		t.Errorf("data ID should be 'key` instead of %q", id)
+	}
+}
+
+func TestUnmarshal(t *testing.T) {
+	data := []byte(":%344")
+
+	_, err := unmarshalKind(data)
+	if err == nil {
+		t.Errorf("unmarshalKind should fail")
+	}
+
+	_, err = unmarshalData(data)
+	if err == nil {
+		t.Errorf("unmarshal should fail")
+	}
+
+	data = []byte(`name: kind1
+idField: name
+jsonSchema:
+  type: object`)
+
+	k, err := unmarshalKind(data)
+	if err != nil {
+		t.Errorf("unmarshalKind should succeeded")
+	}
+	if k.Name != "kind1" {
+		t.Errorf("kind name should be 'kind1' instead of %q", k.Name)
+	}
+	if k.IDField != "name" {
+		t.Errorf("ID field should be 'name' instead of %q", k.IDField)
+	}
+
+	data = []byte(`name: data1
+field1: 123
+`)
+	d, err := unmarshalData(data)
+	if err != nil {
+		t.Errorf("unmarshalData should succeeded")
+	}
+
+	if name := d.GetString("name"); name != "data1" {
+		t.Errorf("name should be 'data1' instead of %q", name)
+	}
+
+	if v := d.Get("field1"); v != 123 {
+		t.Errorf("field1 should be 123 instead of %v", v)
+	}
+}
+
+func TestNewStore(t *testing.T) {
+	s := NewStore(nil, "/kind/", "/data/")
+	if s.cluster != nil {
+		t.Error("cluster should be nil")
+	}
+	if s.KindPrefix != "/kind/" {
+		t.Error("KindPrefix should be '/kind'")
+	}
+	if s.DataPrefix != "/data/" {
+		t.Error("DataPrefix should be '/data'")
+	}
+
+	if key := s.kindKey("test"); key != "/kind/test" {
+		t.Errorf("kind key should be '/kind/test` instead of %q", key)
+	}
+
+	if prefix := s.dataPrefix("test"); prefix != "/data/test/" {
+		t.Errorf("data prefix should be '/data/test/` instead of %q", prefix)
+	}
+
+	if key := s.dataKey("foo", "bar"); key != "/data/foo/bar" {
+		t.Errorf("data key should be '/data/foo/bar` instead of %q", key)
+	}
+}
+
+func TestGetKind(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err := s.GetKind("kind1")
+	if err == nil {
+		t.Errorf("GetKind should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	k, err := s.GetKind("kind1")
+	if err != nil {
+		t.Errorf("GetKind should succeed")
+	}
+	if k != nil {
+		t.Errorf("GetKind should return nil")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1`),
+		}, nil
+	}
+	k, err = s.GetKind("kind1")
+	if err != nil {
+		t.Errorf("GetKind should succeed")
+	}
+	if k == nil {
+		t.Errorf("GetKind should return non-nil value")
+	}
+}
+
+func TestListKinds(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err := s.ListKinds()
+	if err == nil {
+		t.Errorf("ListKinds should fail")
+	}
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return map[string]*mvccpb.KeyValue{
+			"kind1": {Value: []byte(`name: kind1`)},
+			"kind2": {Value: []byte(`@3sj3`)},
+		}, nil
+	}
+	_, err = s.ListKinds()
+	if err == nil {
+		t.Errorf("ListKinds should fail")
+	}
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return map[string]*mvccpb.KeyValue{
+			"kind1": {Value: []byte(`name: kind1`)},
+			"kind2": {Value: []byte(`name: kind2
+idField: key`),
+			},
+		}, nil
+	}
+	kinds, err := s.ListKinds()
+	if err != nil {
+		t.Errorf("ListKinds shoud succeed")
+	}
+
+	if len(kinds) != 2 {
+		t.Errorf("ListKinds should return 2 kinds")
+	}
+}
+
+func TestPutKind(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	k := &Kind{JSONSchema: dynamicobject.DynamicObject{"@#": "2938"}}
+	err := s.PutKind(k, true)
+	if err == nil {
+		t.Errorf("PutKind should fail")
+	}
+
+	k.JSONSchema = nil
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	err = s.PutKind(k, true)
+	if err == nil {
+		t.Errorf("PutKind should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	err = s.PutKind(k, true)
+	if err == nil {
+		t.Errorf("PutKind should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1`),
+		}, nil
+	}
+	err = s.PutKind(k, false)
+	if err == nil {
+		t.Errorf("PutKind should fail")
+	}
+
+	err = s.PutKind(k, true)
+	if err != nil {
+		t.Errorf("PutKind should succeed")
+	}
+}
+
+func TestDeleteKind(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	err := s.DeleteKind("kind1")
+	if err == nil {
+		t.Errorf("DeleteKind should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	err = s.DeleteKind("kind1")
+	if err == nil {
+		t.Errorf("DeleteKind should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1`),
+		}, nil
+	}
+	err = s.DeleteKind("kind1")
+	if err != nil {
+		t.Errorf("DeleteKind should succeed")
+	}
+}
+
+func TestGetData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err := s.GetData("kind1", "data1")
+	if err == nil {
+		t.Errorf("GetData should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	d, err := s.GetData("kind1", "data1")
+	if err != nil {
+		t.Errorf("GetData should succeed")
+	}
+	if d != nil {
+		t.Errorf("GetData should return nil")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: data1`),
+		}, nil
+	}
+	d, err = s.GetData("kind1", "data1")
+	if err != nil {
+		t.Errorf("GetData should succeed")
+	}
+	if d == nil {
+		t.Errorf("GetData should return non-nil value")
+	}
+}
+
+func TestListData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err := s.ListData("")
+	if err == nil {
+		t.Errorf("ListData should fail")
+	}
+	_, err = s.ListData("kind1")
+	if err == nil {
+		t.Errorf("ListData should fail")
+	}
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return map[string]*mvccpb.KeyValue{
+			"data1": {Value: []byte(`name: data1`)},
+			"data2": {Value: []byte(`@3sj3`)},
+		}, nil
+	}
+	_, err = s.ListData("kind1")
+	if err == nil {
+		t.Errorf("ListData should fail")
+	}
+
+	cls.MockedGetRawPrefix = func(prefix string) (map[string]*mvccpb.KeyValue, error) {
+		return map[string]*mvccpb.KeyValue{
+			"data1": {Value: []byte(`name: data1`)},
+			"data2": {Value: []byte(`name: data2
+field1: 123`),
+			},
+		}, nil
+	}
+	data, err := s.ListData("kind1")
+	if err != nil {
+		t.Errorf("ListData shoud succeed")
+	}
+
+	if len(data) != 2 {
+		t.Errorf("ListData should return 2 data items")
+	}
+}
+
+func TestPutData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	// failed to get kind
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err := s.PutData("kind1", nil, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// kind not found
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	_, err = s.PutData("kind1", nil, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// empty data id
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1
+jsonSchema:
+  type: object
+  properties:
+    field1:
+      type: string
+      required: true
+`)}, nil
+	}
+	d := Data{}
+	_, err = s.PutData("kind1", d, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// validation failure
+	d["name"] = "data1"
+	_, err = s.PutData("kind1", d, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// make validation success
+	d["field1"] = "abc"
+
+	// get data fail
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	_, err = s.PutData("kind1", d, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// get nil data
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		if strings.HasSuffix(key, "data1") {
+			return nil, nil
+		}
+		return &mvccpb.KeyValue{Value: []byte(`name: kind1`)}, nil
+	}
+	_, err = s.PutData("kind1", d, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// get data success
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		if strings.HasSuffix(key, "data1") {
+			return &mvccpb.KeyValue{Value: []byte(`name: data1`)}, nil
+		}
+		return &mvccpb.KeyValue{Value: []byte(`name: kind1`)}, nil
+	}
+	_, err = s.PutData("kind1", d, false)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+
+	// update success
+	id, err := s.PutData("kind1", d, true)
+	if err != nil {
+		t.Errorf("PutData should succeed")
+	}
+	if id != "data1" {
+		t.Errorf("data ID should be 'data1' instead of %q", id)
+	}
+
+	// put data failure
+	cls.MockedPut = func(key, value string) error {
+		return fmt.Errorf("mocked error")
+	}
+	_, err = s.PutData("kind1", d, true)
+	if err == nil {
+		t.Errorf("PutData should fail")
+	}
+}
+
+func TestBatchUpdateData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	// failed to get kind
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	err := s.BatchUpdateData("kind1", nil, nil)
+	if err == nil {
+		t.Errorf("BatchUpdateData should fail")
+	}
+
+	// kind not found
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, nil
+	}
+	err = s.BatchUpdateData("kind1", nil, nil)
+	if err == nil {
+		t.Errorf("BatchUpdateData should fail")
+	}
+
+	data := []Data{
+		{"name": "data1", "field1": "foo"},
+		{"field2": "field2"},
+	}
+	// validation failure
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1
+jsonSchema:
+  type: object
+  properties:
+    field1:
+      type: string
+      required: true
+`)}, nil
+	}
+	err = s.BatchUpdateData("kind1", nil, data)
+	if err == nil {
+		t.Errorf("BatchUpdateData should fail")
+	}
+
+	// empty data id
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return &mvccpb.KeyValue{
+			Value: []byte(`name: kind1`),
+		}, nil
+	}
+	err = s.BatchUpdateData("kind1", nil, data)
+	if err == nil {
+		t.Errorf("BatchUpdateData should fail")
+	}
+
+	// make validation success
+	cls.MockedSTM = func(apply func(concurrency.STM) error) error {
+		stm := &clustertest.MockedSTM{}
+		return apply(stm)
+	}
+	data[1]["name"] = "data2"
+	err = s.BatchUpdateData("kind1", []string{"data1", "data3"}, data)
+	if err != nil {
+		t.Errorf("BatchUpdateData should succeed")
+	}
+}
+
+func TestDeleteData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+	err := s.DeleteData("kind1", "data1")
+	if err == nil {
+		t.Errorf("DeleteData should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		if strings.HasSuffix(key, "data1") {
+			return nil, nil
+		}
+		return &mvccpb.KeyValue{Value: []byte(`name: kind1`)}, nil
+	}
+	err = s.DeleteData("kind1", "data1")
+	if err == nil {
+		t.Errorf("DeleteData should fail")
+	}
+
+	cls.MockedGetRaw = func(key string) (*mvccpb.KeyValue, error) {
+		if strings.HasSuffix(key, "data1") {
+			return &mvccpb.KeyValue{Value: []byte(`name: data1`)}, nil
+		}
+		return &mvccpb.KeyValue{Value: []byte(`name: kind1`)}, nil
+	}
+	err = s.DeleteData("kind1", "data1")
+	if err != nil {
+		t.Errorf("DeleteData should succeed")
+	}
+}
+
+func TestDeleteAllData(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+	s.DeleteAllData("kind1")
+}
+
+func TestWatch(t *testing.T) {
+	cls := clustertest.NewMockedCluster()
+	s := NewStore(cls, "/kind/", "/data/")
+
+	cls.MockedSyncer = func(pullInterval time.Duration) (*cluster.Syncer, error) {
+		return nil, fmt.Errorf("mocked error")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	err := s.Watch(ctx, "kind1", func(d []Data) {})
+	if err == nil {
+		t.Errorf("Watch should fail")
+	}
+
+	cancel()
+}
diff --git a/pkg/context/contexttest/request.go b/pkg/context/contexttest/request.go
index aac1a914..0b4a2dc5 100644
--- a/pkg/context/contexttest/request.go
+++ b/pkg/context/contexttest/request.go
@@ -44,7 +44,7 @@ type MockedHTTPRequest struct {
 	MockedCookies     func() []*http.Cookie
 	MockedAddCookie   func(cookie *http.Cookie)
 	MockedBody        func() io.Reader
-	MockedSetBody     func(io.Reader)
+	MockedSetBody     func(io.Reader, bool)
 	MockedStd         func() *http.Request
 	MockedSize        func() uint64
 }
@@ -189,9 +189,9 @@ func (r *MockedHTTPRequest) Body() io.Reader {
 }
 
 // SetBody mocks the SetBody function of HTTPRequest
-func (r *MockedHTTPRequest) SetBody(body io.Reader) {
+func (r *MockedHTTPRequest) SetBody(body io.Reader, closePreviousReader bool) {
 	if r.MockedSetBody != nil {
-		r.MockedSetBody(body)
+		r.MockedSetBody(body, closePreviousReader)
 	}
 }
 
diff --git a/pkg/filter/certextractor/certextractor_test.go b/pkg/filter/certextractor/certextractor_test.go
new file mode 100644
index 00000000..b6951ee1
--- /dev/null
+++ b/pkg/filter/certextractor/certextractor_test.go
@@ -0,0 +1,215 @@
+/*
+ * 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 certextractor
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func createCertExtractor(
+	yamlSpec string, prev *CertExtractor, supervisor *supervisor.Supervisor) (*CertExtractor, error) {
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+	spec, err := httppipeline.NewFilterSpec(rawSpec, supervisor)
+	if err != nil {
+		return nil, err
+	}
+	hl := &CertExtractor{}
+	if prev == nil {
+		hl.Init(spec)
+	} else {
+		hl.Inherit(spec, prev)
+	}
+	return hl, nil
+}
+
+const yaml = `
+kind: "CertExtractor"
+name: "cn-extractor"
+certIndex: -1
+target: "subject"
+field: "CommonName"
+headerKey: "key"
+`
+
+func TestSpec(t *testing.T) {
+	assert := assert.New(t)
+
+	const yamlNoKey = `
+kind: "CertExtractor"
+name: "cn-extractor"
+certIndex: -1
+target: "subject"
+field: "CommonName"
+`
+	ce, err := createCertExtractor(yaml, nil, nil)
+	assert.Nil(err)
+	assert.NotEqual(ce.Description(), 0)
+	assert.Nil(err)
+	assert.NotEqual(ce.Description(), 0)
+	assert.Nil(ce.Status())
+	ce, err = createCertExtractor(yaml, ce, nil)
+	assert.Nil(err)
+	assert.Equal(ce.headerKey, "key")
+
+	ce, err = createCertExtractor(yamlNoKey, ce, nil)
+	assert.Nil(err)
+	assert.Equal(ce.headerKey, "tls-subject-CommonName")
+}
+
+func prepareCtxAndHeader(connState *tls.ConnectionState) (*contexttest.MockedHTTPContext, http.Header) {
+	ctx := &contexttest.MockedHTTPContext{}
+	header := http.Header{}
+	stdr := &http.Request{}
+	stdr.TLS = connState
+	ctx.MockedRequest.MockedStd = func() *http.Request {
+		return stdr
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+	return ctx, header
+}
+
+func TestHandle(t *testing.T) {
+	assert := assert.New(t)
+
+	t.Run("no TLS", func(t *testing.T) {
+		ctx := &contexttest.MockedHTTPContext{}
+		ce, _ := createCertExtractor(yaml, nil, nil)
+		assert.Equal("", ce.Handle(ctx))
+
+		peerCertificates := make([]*x509.Certificate, 0)
+		connState := &tls.ConnectionState{PeerCertificates: peerCertificates}
+		ctx, _ = prepareCtxAndHeader(connState)
+		ce, _ = createCertExtractor(yaml, nil, nil)
+		assert.Equal("", ce.Handle(ctx))
+	})
+
+	t.Run("TLS has no field", func(t *testing.T) {
+		peerCertificates := make([]*x509.Certificate, 0)
+		peerCertificates = append(peerCertificates, &x509.Certificate{
+			Subject: pkix.Name{},
+			Issuer:  pkix.Name{},
+		})
+		connState := &tls.ConnectionState{PeerCertificates: peerCertificates}
+		ctx, _ := prepareCtxAndHeader(connState)
+		ce, _ := createCertExtractor(yaml, nil, nil)
+		assert.Equal("", ce.Handle(ctx))
+	})
+
+	t.Run("TLS has field", func(t *testing.T) {
+		peerCertificates := make([]*x509.Certificate, 0)
+		peerCertificates = append(peerCertificates, &x509.Certificate{
+			Subject: pkix.Name{
+				Country:            []string{"1"},
+				Organization:       []string{"2"},
+				OrganizationalUnit: []string{"3"},
+				Locality:           []string{"4"},
+				Province:           []string{"5"},
+				StreetAddress:      []string{"6"},
+				PostalCode:         []string{"7"},
+				SerialNumber:       "8",
+				CommonName:         "INFO-1",
+			},
+			Issuer: pkix.Name{
+				CommonName: "INFO-2",
+			},
+		})
+		connState := &tls.ConnectionState{PeerCertificates: peerCertificates}
+		t.Run("subject", func(t *testing.T) {
+			ctx, header := prepareCtxAndHeader(connState)
+			subjectYaml := yaml
+			ce, _ := createCertExtractor(subjectYaml, nil, nil)
+			assert.Equal("", ce.handle(ctx))
+			assert.Equal("INFO-1", header.Get("key"))
+		})
+
+		t.Run("issuer", func(t *testing.T) {
+			ctx, header := prepareCtxAndHeader(connState)
+			issuerYaml := strings.ReplaceAll(yaml, `target: "subject"`, `target: "issuer"`)
+			ce, _ := createCertExtractor(issuerYaml, nil, nil)
+			assert.Equal("", ce.handle(ctx))
+			assert.Equal("INFO-2", header.Get("key"))
+		})
+
+		t.Run("test all fields", func(t *testing.T) {
+			ctx, header := prepareCtxAndHeader(connState)
+			fields := []string{"Country", "Organization", "OrganizationalUnit", "Locality",
+				"Province", "StreetAddress", "PostalCode", "SerialNumber",
+			}
+			for _, fieldName := range fields {
+				ce, _ := createCertExtractor(
+					strings.ReplaceAll(yaml, `field: "CommonName"`, fmt.Sprintf(`field: "%s"`, fieldName)),
+					nil,
+					nil,
+				)
+				assert.Equal("", ce.Handle(ctx))
+			}
+			assert.Equal([]string{"1", "2", "3", "4", "5", "6", "7", "8"}, header["Key"])
+		})
+		t.Run("multiple certs", func(t *testing.T) {
+			for _, val := range []string{"second", "third", "fourth"} {
+				peerCertificates = append(peerCertificates, &x509.Certificate{
+					Subject: pkix.Name{
+						Province: []string{val},
+					},
+				})
+
+			}
+			connState := &tls.ConnectionState{PeerCertificates: peerCertificates}
+			ctx, header := prepareCtxAndHeader(connState)
+			yamlConfig := `
+kind: "CertExtractor"
+name: "cn-extractor"
+certIndex: -2 # second last certificate
+target: "subject"
+field: "Province"
+`
+			ce, _ := createCertExtractor(yamlConfig, nil, nil)
+			assert.Equal("", ce.Handle(ctx))
+			assert.Equal("third", header.Get("tls-subject-province"))
+
+			ctx, header = prepareCtxAndHeader(connState)
+			yamlConfig2 := strings.ReplaceAll(yamlConfig, "certIndex: -2", "certIndex: -15")
+			ce, _ = createCertExtractor(yamlConfig2, nil, nil)
+			assert.Equal("", ce.Handle(ctx))
+			assert.Equal("second", header.Get("tls-subject-province"))
+		})
+	})
+}
diff --git a/pkg/filter/corsadaptor/corsadaptor_test.go b/pkg/filter/corsadaptor/corsadaptor_test.go
index 3790363d..abd883f6 100644
--- a/pkg/filter/corsadaptor/corsadaptor_test.go
+++ b/pkg/filter/corsadaptor/corsadaptor_test.go
@@ -28,50 +28,105 @@ import (
 )
 
 func TestCORSAdaptor(t *testing.T) {
-	const yamlSpec = `
+	t.Run("CORS preflight-request", func(t *testing.T) {
+		const yamlSpec = `
 kind: CORSAdaptor
 name: cors
 `
-	rawSpec := make(map[string]interface{})
-	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+		rawSpec := make(map[string]interface{})
+		yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
 
-	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
-	if e != nil {
-		t.Errorf("unexpected error: %v", e)
-	}
+		spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+		if e != nil {
+			t.Errorf("unexpected error: %v", e)
+		}
 
-	cors := &CORSAdaptor{}
-	cors.Init(spec)
+		cors := &CORSAdaptor{}
+		cors.Init(spec)
 
-	header := http.Header{}
-	ctx := &contexttest.MockedHTTPContext{}
-	ctx.MockedRequest.MockedMethod = func() string {
-		return http.MethodOptions
-	}
-	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
-		return httpheader.New(header)
-	}
+		header := http.Header{}
+		ctx := &contexttest.MockedHTTPContext{}
+		ctx.MockedRequest.MockedMethod = func() string {
+			return http.MethodOptions
+		}
+		ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+			return httpheader.New(header)
+		}
 
-	result := cors.Handle(ctx)
-	if result == resultPreflighted {
-		t.Error("request should not be preflighted")
-	}
+		result := cors.Handle(ctx)
+		if result == resultPreflighted {
+			t.Error("request should not be preflighted")
+		}
 
-	header.Add("Access-Control-Request-Method", "abc")
-	result = cors.Handle(ctx)
-	if result != resultPreflighted {
-		t.Error("request should be preflighted")
-	}
+		header.Add("Access-Control-Request-Method", "abc")
+		result = cors.Handle(ctx)
+		if result != resultPreflighted {
+			t.Error("request should be preflighted")
+		}
 
-	newCors := &CORSAdaptor{}
-	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
-	newCors.Inherit(spec, cors)
-	cors.Close()
-	ctx.MockedRequest.MockedMethod = func() string {
-		return http.MethodGet
-	}
-	result = newCors.Handle(ctx)
-	if result == resultPreflighted {
-		t.Error("request should not be preflighted")
-	}
+		newCors := &CORSAdaptor{}
+		spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+		newCors.Inherit(spec, cors)
+		cors.Close()
+		ctx.MockedRequest.MockedMethod = func() string {
+			return http.MethodGet
+		}
+		result = newCors.Handle(ctx)
+		if result == resultPreflighted {
+			t.Error("request should not be preflighted")
+		}
+	})
+	t.Run("CORS request", func(t *testing.T) {
+		const yamlSpec = `
+kind: CORSAdaptor
+name: cors
+supportCORSRequest: true
+allowedOrigins:
+  - test.orig.test
+`
+		rawSpec := make(map[string]interface{})
+		yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+		spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+		if e != nil {
+			t.Errorf("unexpected error: %v", e)
+		}
+
+		cors := &CORSAdaptor{}
+		cors.Init(spec)
+		cors.Description()
+		cors.Status()
+		header := http.Header{}
+		ctx := &contexttest.MockedHTTPContext{}
+		ctx.MockedRequest.MockedMethod = func() string {
+			return http.MethodOptions
+		}
+		ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+			return httpheader.New(header)
+		}
+		result := cors.Handle(ctx)
+		if result == resultPreflighted {
+			t.Error("request should not be preflighted")
+		}
+		header.Add("Origin", "test.orig.test")
+		header.Add("Access-Control-Request-Method", "get")
+		result = cors.Handle(ctx)
+		if result != resultPreflighted {
+			t.Error("request should be preflighted")
+		}
+
+		header = http.Header{}
+		header.Add("Origin", "test.orig.test")
+		ctx = &contexttest.MockedHTTPContext{}
+		ctx.MockedRequest.MockedMethod = func() string {
+			return http.MethodGet
+		}
+		ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+			return httpheader.New(header)
+		}
+		result = cors.Handle(ctx)
+		if result == resultPreflighted {
+			t.Error("request should not be preflighted")
+		}
+	})
 }
diff --git a/pkg/filter/headerlookup/headerlookup_test.go b/pkg/filter/headerlookup/headerlookup_test.go
index 2c3fce5f..0859d79a 100644
--- a/pkg/filter/headerlookup/headerlookup_test.go
+++ b/pkg/filter/headerlookup/headerlookup_test.go
@@ -106,6 +106,24 @@ headerKey: "X-AUTH-USER"
 etcdPrefix: "/credentials/"
 headerSetters:
   - etcdKey: "ext-id"
+`,
+		`
+name: headerLookup
+kind: HeaderLookup
+headerKey: "X-AUTH-USER"
+etcdPrefix: "/credentials/"
+headerSetters:
+  - headerKey: "X-ext-id"
+`,
+		`
+name: headerLookup
+kind: HeaderLookup
+headerKey: "X-AUTH-USER"
+pathRegExp: "**"
+etcdPrefix: "/credentials/"
+headerSetters:
+  - headerKey: "X-ext-id"
+    etcdKey: "ext-id"
 `,
 	}
 
@@ -236,6 +254,73 @@ extra-entry: "extra"
 		}
 	}
 
+	if hl.Status() != nil {
+		t.Errorf("status should be nil")
+	}
+	if len(hl.Description()) == 0 {
+		t.Errorf("description should not be empty")
+	}
+	hl.Close()
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	clusterInstance.CloseServer(wg)
+	wg.Wait()
+}
+
+func TestHandleWithPath(t *testing.T) {
+	etcdDirName, err := ioutil.TempDir("", "etcd-headerlookup-path-test")
+	check(err)
+	defer os.RemoveAll(etcdDirName)
+	const config = `
+name: headerLookup
+kind: HeaderLookup
+headerKey: "X-AUTH-USER"
+etcdPrefix: "credentials/"
+pathRegExp: "^/api/([a-z]+)/[0-9]*"
+headerSetters:
+  - etcdKey: "ext-id"
+    headerKey: "user-ext-id"
+`
+	clusterInstance := cluster.CreateClusterForTest(etcdDirName)
+	var mockMap sync.Map
+	supervisor := supervisor.NewMock(
+		nil, clusterInstance, mockMap, mockMap, nil, nil, false, nil, nil)
+
+	// let's put data to 'bob'
+	clusterInstance.Put("/custom-data/credentials/bob-bananas",
+		`
+ext-id: 333
+extra-entry: "extra"
+`)
+	clusterInstance.Put("/custom-data/credentials/bob-pearls",
+		`
+ext-id: 4444
+extra-entry: "extra"
+`)
+	hl, err := createHeaderLookup(config, nil, supervisor)
+	check(err)
+
+	ctx, header := prepareCtxAndHeader()
+	header.Set("X-AUTH-USER", "bob")
+	hl.Handle(ctx) // path does not match
+	if header.Get("user-ext-id") != "" {
+		t.Errorf("failed")
+	}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/api/bananas/9281"
+	}
+	hl.Handle(ctx)
+	if header.Get("user-ext-id") != "333" {
+		t.Errorf("failed")
+	}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/api/pearls/"
+	}
+	hl.Handle(ctx)
+	if header.Get("user-ext-id") != "4444" {
+		t.Errorf("failed")
+	}
+
 	hl.Close()
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
diff --git a/pkg/filter/headertojson/headertojson_test.go b/pkg/filter/headertojson/headertojson_test.go
index 136c6825..31de0694 100644
--- a/pkg/filter/headertojson/headertojson_test.go
+++ b/pkg/filter/headertojson/headertojson_test.go
@@ -80,10 +80,11 @@ func TestHandleHTTP(t *testing.T) {
 		}
 		body, err := json.Marshal(bodyMap)
 		assert.Nil(err)
+
 		req, err := http.NewRequest(http.MethodPost, "127.0.0.1", bytes.NewReader(body))
 		assert.Nil(err)
-
 		req.Header.Add("x-username", "clientA")
+
 		w := httptest.NewRecorder()
 		ctx := context.New(w, req, tracing.NoopTracing, "no trace")
 		ctx.SetHandlerCaller(func(lastResult string) string {
@@ -137,10 +138,11 @@ func TestHandleHTTP(t *testing.T) {
 		}
 		body, err := json.Marshal(bodyMap)
 		assert.Nil(err)
+
 		req, err := http.NewRequest(http.MethodPost, "127.0.0.1", bytes.NewReader(body))
 		assert.Nil(err)
-
 		req.Header.Add("x-username", "clientA")
+
 		w := httptest.NewRecorder()
 		ctx := context.New(w, req, tracing.NoopTracing, "no trace")
 		ctx.SetHandlerCaller(func(lastResult string) string {
diff --git a/pkg/filter/requestadaptor/requestadaptor_test.go b/pkg/filter/requestadaptor/requestadaptor_test.go
new file mode 100644
index 00000000..e87845ce
--- /dev/null
+++ b/pkg/filter/requestadaptor/requestadaptor_test.go
@@ -0,0 +1,336 @@
+/*
+ * 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 requestadaptor
+
+import (
+	"bytes"
+	"compress/gzip"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/tracing"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/pathadaptor"
+	"github.com/stretchr/testify/assert"
+)
+
+func init() {
+	logger.InitNop()
+}
+
+func defaultFilterSpec(spec *Spec) *httppipeline.FilterSpec {
+	meta := &httppipeline.FilterMetaSpec{
+		Name:     "request-adaptor",
+		Kind:     Kind,
+		Pipeline: "pipeline-demo",
+	}
+	filterSpec := httppipeline.MockFilterSpec(nil, nil, "", meta, spec)
+	return filterSpec
+}
+
+func getGzipEncoding(t *testing.T, data []byte) io.Reader {
+	var buf bytes.Buffer
+	gw := gzip.NewWriter(&buf)
+	defer gw.Close()
+	_, err := gw.Write(data)
+	assert.Nil(t, err)
+	return &buf
+}
+
+func getContext(t *testing.T, req *http.Request, httpTemp *context.HTTPTemplate) context.HTTPContext {
+	w := httptest.NewRecorder()
+	ctx := context.New(w, req, tracing.NoopTracing, "no trace")
+	ctx.SetHandlerCaller(func(lastResult string) string {
+		return lastResult
+	})
+	ctx.SetTemplate(httpTemp)
+	return ctx
+}
+
+func getTemplate(t *testing.T, filterSpec *httppipeline.FilterSpec) *context.HTTPTemplate {
+	filterBuffs := []context.FilterBuff{
+		{Name: filterSpec.Name(), Buff: []byte(filterSpec.YAMLConfig())},
+	}
+	httpTemp, err := context.NewHTTPTemplate(filterBuffs)
+	assert.Nil(t, err)
+	return httpTemp
+}
+
+func TestRequestAdaptor(t *testing.T) {
+	assert := assert.New(t)
+	{
+		// normal case
+		spec := &Spec{}
+		filterSpec := defaultFilterSpec(spec)
+		ra := &RequestAdaptor{}
+		ra.Init(filterSpec)
+		assert.Equal(Kind, ra.Kind())
+		assert.NotEmpty(ra.Description())
+		assert.NotNil(ra.DefaultSpec())
+		assert.NotEmpty(ra.Results())
+		assert.Nil(ra.Status())
+
+		newRA := &RequestAdaptor{}
+		newRA.Inherit(filterSpec, ra)
+		newRA.Close()
+	}
+
+	{
+		// panic when invalid compress type
+		spec := &Spec{
+			Compress: "zip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		ra := &RequestAdaptor{}
+		assert.Panics(func() { ra.Init(filterSpec) })
+	}
+
+	{
+		// panic when invalid decompress type
+		spec := &Spec{
+			Decompress: "zip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		ra := &RequestAdaptor{}
+		assert.Panics(func() { ra.Init(filterSpec) })
+	}
+
+	{
+		// panic when set both set compress and decompress
+		spec := &Spec{
+			Decompress: "gzip",
+			Compress:   "gzip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		ra := &RequestAdaptor{}
+		assert.Panics(func() { ra.Init(filterSpec) })
+	}
+
+	{
+		// panic when set body and Decompress
+		spec := &Spec{
+			Decompress: "gzip",
+			Body:       "body",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		ra := &RequestAdaptor{}
+		assert.Panics(func() { ra.Init(filterSpec) })
+	}
+}
+
+func TestDecompress(t *testing.T) {
+	assert := assert.New(t)
+
+	{
+		// decompress without body in spec
+		spec := &Spec{
+			Decompress: "gzip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		httpTemp := getTemplate(t, filterSpec)
+
+		ra := &RequestAdaptor{}
+		ra.Init(filterSpec)
+
+		{
+			// compress success
+			data := "123"
+			req, err := http.NewRequest(http.MethodPost, "127.0.0.1", getGzipEncoding(t, []byte(data)))
+			assert.Nil(err)
+			req.Header.Add("Content-Encoding", "gzip")
+
+			ctx := getContext(t, req, httpTemp)
+
+			ans := ra.Handle(ctx)
+			assert.Equal("", ans)
+			ctx.Finish()
+
+			encoding := ctx.Request().Header().Get("Content-Encoding")
+			assert.Equal("", encoding)
+
+			body, err := io.ReadAll(ctx.Request().Body())
+			assert.Nil(err)
+			assert.Equal("123", string(body))
+		}
+
+		{
+			// compress fail
+			data := "123"
+			req, err := http.NewRequest(http.MethodPost, "127.0.0.1", bytes.NewReader([]byte(data)))
+			assert.Nil(err)
+			req.Header.Add("Content-Encoding", "gzip")
+
+			ctx := getContext(t, req, httpTemp)
+
+			ans := ra.Handle(ctx)
+			assert.Equal(resultDecompressFail, ans)
+			ctx.Finish()
+		}
+	}
+}
+
+func TestCompress(t *testing.T) {
+	assert := assert.New(t)
+
+	{
+		// compress without body in spec
+		spec := &Spec{
+			Compress: "gzip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		httpTemp := getTemplate(t, filterSpec)
+
+		ra := &RequestAdaptor{}
+		ra.Init(filterSpec)
+
+		data := "123"
+		req, err := http.NewRequest(http.MethodPost, "127.0.0.1", bytes.NewReader([]byte(data)))
+		assert.Nil(err)
+
+		ctx := getContext(t, req, httpTemp)
+
+		ans := ra.Handle(ctx)
+		assert.Equal("", ans)
+		ctx.Finish()
+
+		encoding := ctx.Request().Header().Get("Content-Encoding")
+		assert.Equal("gzip", encoding)
+
+		reader, err := gzip.NewReader(ctx.Request().Body())
+		assert.Nil(err)
+		defer reader.Close()
+		body, err := io.ReadAll(reader)
+		assert.Nil(err)
+		assert.Equal("123", string(body))
+	}
+
+	{
+		// compress with body in spec
+		spec := &Spec{
+			Body:     "spec_body",
+			Compress: "gzip",
+		}
+		filterSpec := defaultFilterSpec(spec)
+		httpTemp := getTemplate(t, filterSpec)
+
+		ra := &RequestAdaptor{}
+		ra.Init(filterSpec)
+
+		{
+			// original gzip body
+			data := "123"
+			req, err := http.NewRequest(http.MethodPost, "127.0.0.1", getGzipEncoding(t, []byte(data)))
+			assert.Nil(err)
+			req.Header.Add("Content-Encoding", "gzip")
+
+			ctx := getContext(t, req, httpTemp)
+
+			ans := ra.Handle(ctx)
+			assert.Equal("", ans)
+			ctx.Finish()
+
+			encoding := ctx.Request().Header().Get("Content-Encoding")
+			assert.Equal("gzip", encoding)
+
+			reader, err := gzip.NewReader(ctx.Request().Body())
+			assert.Nil(err)
+			defer reader.Close()
+			body, err := io.ReadAll(reader)
+			assert.Nil(err)
+			assert.Equal("spec_body", string(body))
+		}
+
+		{
+			// original uncompressed body
+			data := "123"
+			req, err := http.NewRequest(http.MethodPost, "127.0.0.1", bytes.NewReader([]byte(data)))
+			assert.Nil(err)
+
+			ctx := getContext(t, req, httpTemp)
+
+			ans := ra.Handle(ctx)
+			assert.Equal("", ans)
+			ctx.Finish()
+
+			encoding := ctx.Request().Header().Get("Content-Encoding")
+			assert.Equal("gzip", encoding)
+
+			reader, err := gzip.NewReader(ctx.Request().Body())
+			assert.Nil(err)
+			defer reader.Close()
+			body, err := io.ReadAll(reader)
+			assert.Nil(err)
+			assert.Equal("spec_body", string(body))
+		}
+	}
+}
+
+func TestHandle(t *testing.T) {
+	assert := assert.New(t)
+
+	spec := &Spec{
+		Method: http.MethodDelete,
+		Host:   "127.0.0.2",
+		Body:   "123",
+		Header: &httpheader.AdaptSpec{
+			Add: map[string]string{"X-Add": "add-value"},
+			Set: map[string]string{"X-Set": "set-value"},
+		},
+		Path: &pathadaptor.Spec{
+			Replace: "/path",
+		},
+	}
+	filterSpec := defaultFilterSpec(spec)
+	httpTemp := getTemplate(t, filterSpec)
+
+	ra := &RequestAdaptor{}
+	ra.Init(filterSpec)
+
+	req, err := http.NewRequest(http.MethodPost, "127.0.0.1", nil)
+	assert.Nil(err)
+
+	ctx := getContext(t, req, httpTemp)
+
+	ans := ra.Handle(ctx)
+	assert.Equal("", ans)
+	ctx.Finish()
+
+	method := ctx.Request().Method()
+	assert.Equal(http.MethodDelete, method)
+
+	host := ctx.Request().Host()
+	assert.Equal("127.0.0.2", host)
+
+	body, err := io.ReadAll(ctx.Request().Body())
+	assert.Nil(err)
+	assert.Equal("123", string(body))
+
+	headerValue := ctx.Request().Header().Get("X-Add")
+	assert.Equal("add-value", headerValue)
+
+	headerValue = ctx.Request().Header().Get("X-Set")
+	assert.Equal("set-value", headerValue)
+
+	path := ctx.Request().Path()
+	assert.Equal("/path", path)
+}
diff --git a/pkg/object/meshcontroller/spec/spec_test.go b/pkg/object/meshcontroller/spec/spec_test.go
index 306c2438..f55bf2cc 100644
--- a/pkg/object/meshcontroller/spec/spec_test.go
+++ b/pkg/object/meshcontroller/spec/spec_test.go
@@ -1205,30 +1205,6 @@ func TestEgressName(t *testing.T) {
 
 func TestCustomResource(t *testing.T) {
 	r := CustomResource{}
-	if r.Name() != "" {
-		t.Error("name should be empty")
-	}
-	r["name"] = 1
-	if r.Name() != "" {
-		t.Error("name should be empty")
-	}
-	r["name"] = "obj1"
-	if r.Name() != "obj1" {
-		t.Error("name should be obj1")
-	}
-
-	if r.Kind() != "" {
-		t.Error("kind should be empty")
-	}
-	r["kind"] = 1
-	if r.Kind() != "" {
-		t.Error("kind should be empty")
-	}
-	r["kind"] = "kind1"
-	if r.Kind() != "kind1" {
-		t.Error("kind should be kind1")
-	}
-
 	r["field1"] = map[string]interface{}{
 		"sub1": 1,
 		"sub2": "value2",
diff --git a/pkg/util/dynamicobject/dynamicobject_test.go b/pkg/util/dynamicobject/dynamicobject_test.go
new file mode 100644
index 00000000..ed88cacc
--- /dev/null
+++ b/pkg/util/dynamicobject/dynamicobject_test.go
@@ -0,0 +1,64 @@
+/*
+ * 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 dynamicobject
+
+import (
+	"testing"
+
+	"gopkg.in/yaml.v2"
+)
+
+func TestDynamicObject(t *testing.T) {
+	do := DynamicObject{}
+	if do.GetString("name") != "" {
+		t.Error("name should be empty")
+	}
+	do.Set("name", 1)
+	if do.GetString("name") != "" {
+		t.Error("name should be empty")
+	}
+	do.Set("name", "obj1")
+	if do.GetString("name") != "obj1" {
+		t.Error("name should be obj1")
+	}
+	do.Set("value", 1)
+	if do.Get("value") != 1 {
+		t.Error("value should be 1")
+	}
+
+	do.Set("field1", map[string]interface{}{
+		"sub1": 1,
+		"sub2": "value2",
+	})
+
+	do.Set("field2", []interface{}{"sub1", "sub2"})
+
+	data, err := yaml.Marshal(do)
+	if err != nil {
+		t.Errorf("yaml.Marshal should succeed: %v", err.Error())
+	}
+
+	err = yaml.Unmarshal(data, &do)
+	if err != nil {
+		t.Errorf("yaml.Marshal should succeed: %v", err.Error())
+	}
+
+	if _, ok := do.Get("field1").(map[string]interface{}); !ok {
+		t.Errorf("the type of 'field1' should be 'map[string]interface{}'")
+	}
+}