Feat: transformer plugin support map from body to header (#892)

plugins/wasm-go/Makefile

@@ -29,7 +29,6 @@ build-image:
 	                            --build-arg BUILDER=${BUILDER}  \
 	                            --build-arg GOPROXY=$(GOPROXY) \
 	                            -t ${IMG} \
-	                            --load \
 	                            .
 	@echo ""
 	@echo "image:            ${IMG}"
@@ -72,4 +71,4 @@ local-run:
 	python3 .devcontainer/gen_config.py ${PLUGIN_NAME}
 	envoy -c extensions/${PLUGIN_NAME}/config.yaml --concurrency 0 --log-level info --component-log-level wasm:debug
 
-local-all: local-build local-run
+local-all: local-build local-run

plugins/wasm-go/extensions/transformer/README.md

@@ -502,3 +502,95 @@ $ curl -v -X POST console.higress.io/post \
   ...
 }
 ```
+# 特殊用法：实现基于Body参数路由
+
+**Note**
+
+> 需要数据面的proxy wasm版本大于0.2.100
+
+> 编译时，需要带上版本的tag，例如：`tinygo build -o main.wasm -scheduler=none -target=wasi -gc=custom -tags="custommalloc nottinygc_finalizer proxy_wasm_version_0_2_100" ./`
+
+
+配置示例：
+
+```yaml
+reqRules:
+- operate: map
+  headers:
+  - fromKey: userId
+    toKey: x-user-id
+  mapSource: body
+```
+
+此规则将请求body中的`userId`解析出后，设置到请求Header`x-user-id`中，这样就可以基于Higress请求Header匹配路由的能力来实现基于Body参数的路由了。
+
+此配置同时支持`application/json`和`application/x-www-form-urlencoded`两种类型的请求Body。
+
+举例来说：
+
+**对于application/json类型的body**
+
+```bash
+curl localhost -d '{"userId":12, "userName":"johnlanni"}' -H 'content-type:application/json'
+```
+
+将从json中提取出`userId`字段的值，设置到`x-user-id`中，后端服务收到的请求头将增加:`x-usr-id: 12`。
+
+因为在插件新增这个Header后，网关将重新计算路由，所以可以实现网关路由配置根据这个请求头来匹配路由到特定的目标服务。
+
+
+**对于application/x-www-form-urlencoded类型的body**
+
+```bash
+curl localhost -d 'userId=12&userName=johnlanni'
+```
+
+将从`k1=v1&k2=v2`这样的表单格式中提取出`userId`字段的值，设置到`x-user-id`中，后端服务收到的请求头将增加:`x-usr-id: 12`。
+
+因为在插件新增这个Header后，网关将重新计算路由，所以可以实现网关路由配置根据这个请求头来匹配路由到特定的目标服务。
+
+## json path 支持
+
+可以根据 [GJSON Path 语法](https://github.com/tidwall/gjson/blob/master/SYNTAX.md)，从复杂的 json 中提取出字段。
+
+比较常用的操作举例，对于以下 json:
+
+```json
+{
+  "name": {"first": "Tom", "last": "Anderson"},
+  "age":37,
+  "children": ["Sara","Alex","Jack"],
+  "fav.movie": "Deer Hunter",
+  "friends": [
+    {"first": "Dale", "last": "Murphy", "age": 44, "nets": ["ig", "fb", "tw"]},
+    {"first": "Roger", "last": "Craig", "age": 68, "nets": ["fb", "tw"]},
+    {"first": "Jane", "last": "Murphy", "age": 47, "nets": ["ig", "tw"]}
+  ]
+}
+```
+
+可以实现这样的提取:
+
+```text
+name.last              "Anderson"
+name.first             "Tom"
+age                    37
+children               ["Sara","Alex","Jack"]
+children.0             "Sara"
+children.1             "Alex"
+friends.1              {"first": "Roger", "last": "Craig", "age": 68}
+friends.1.first        "Roger"
+```
+
+现在如果想从上面这个 json 格式的 body 中提取出 friends 中第二项的 first 字段，来设置到 Header `x-first-name` 中，同时抽取 last 字段，来设置到 Header `x-last-name` 中，则可以使用这份插件配置:
+
+```yaml
+reqRules:
+- operate: map
+  headers:
+  - fromKey: friends.1.first
+    toKey: x-first-name
+  - fromKey: friends.1.last
+    toKey: x-last-name
+  mapSource: body
+```

plugins/wasm-go/extensions/transformer/go.mod

@@ -6,7 +6,7 @@ replace github.com/alibaba/higress/plugins/wasm-go => ../..
 
 require (
 	github.com/alibaba/higress/plugins/wasm-go v0.0.0-20230829022308-8747e1ddadf0
-	github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240226064518-b3dc4646a35a
+	github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.8.4
 	github.com/tidwall/gjson v1.17.0

plugins/wasm-go/extensions/transformer/go.sum

@@ -4,8 +4,8 @@ github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/higress-group/nottinygc v0.0.0-20231101025119-e93c4c2f8520 h1:IHDghbGQ2DTIXHBHxWfqCYQW1fKjyJ/I7W1pMyUDeEA=
 github.com/higress-group/nottinygc v0.0.0-20231101025119-e93c4c2f8520/go.mod h1:Nz8ORLaFiLWotg6GeKlJMhv8cci8mM43uEnLA5t8iew=
-github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240226064518-b3dc4646a35a h1:luYRvxLTE1xYxrXYj7nmjd1U0HHh8pUPiKfdZ0MhCGE=
-github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240226064518-b3dc4646a35a/go.mod h1:hNFjhrLUIq+kJ9bOcs8QtiplSQ61GZXtd2xHKx4BYRo=
+github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc h1:t2AT8zb6N/59Y78lyRWedVoVWHNRSCBh0oWCC+bluTQ=
+github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc/go.mod h1:hNFjhrLUIq+kJ9bOcs8QtiplSQ61GZXtd2xHKx4BYRo=
 github.com/magefile/mage v1.14.0 h1:6QDX3g6z1YvJ4olPhT1wksUcSa/V0a1B+pJb73fBjyo=
 github.com/magefile/mage v1.14.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

plugins/wasm-go/extensions/transformer/main.go

@@ -243,6 +243,8 @@ func parseConfig(json gjson.Result, config *TransformerConfig, log wrapper.Log)
 		return errors.Wrapf(err, "failed to new transformer")
 	}
 
+	log.Infof("transform config is: reqRules:%+v, respRules:%+v", config.reqRules, config.respRules)
+
 	return nil
 }
 
@@ -312,11 +314,17 @@ func onHttpRequestHeaders(ctx wrapper.HttpContext, config TransformerConfig, log
 	if hs["content-type"] != nil {
 		contentType = hs["content-type"][0]
 	}
-	if config.reqTrans.IsBodyChange() && isValidRequestContentType(contentType) {
+	ctx.SetContext("content-type", contentType)
+
+	isValidRequestContent := isValidRequestContentType(contentType)
+	isBodyChange := config.reqTrans.IsBodyChange()
+	needBodyMapSource := config.reqTrans.NeedBodyMapSource()
+
+	log.Debugf("contentType:%s, isValidRequestContent:%v, isBodyChange:%v, needBodyMapSource:%v",
+		contentType, isValidRequestContent, isBodyChange, needBodyMapSource)
+
+	if isBodyChange && isValidRequestContent {
 		delete(hs, "content-length")
-		ctx.SetContext("content-type", contentType)
-	} else {
-		ctx.DontReadRequestBody()
 	}
 
 	qs, err := parseQueryByPath(path)
@@ -328,26 +336,26 @@ func onHttpRequestHeaders(ctx wrapper.HttpContext, config TransformerConfig, log
 	ctx.SetContext("headers", hs)
 	ctx.SetContext("querys", qs)
 
-	var mapSourceData MapSourceData
-	switch config.reqTrans.GetMapSource() {
-	case "headers":
-		mapSourceData.mapSourceType = "headers"
-		mapSourceData.kvs = hs
-	case "querys":
-		mapSourceData.mapSourceType = "querys"
-		mapSourceData.kvs = qs
-	case "self":
+	if !isValidRequestContent || (!isBodyChange && !needBodyMapSource) {
+		ctx.DontReadRequestBody()
+	} else if needBodyMapSource {
+		// we need do transform during body phase
+		ctx.SetContext("need_head_trans", struct{}{})
+		log.Debug("delay header's transform to body phase")
+		return types.HeaderStopIteration
+	}
 
-	default:
-		log.Warnf("invalid mapSource in request header: %v", config.reqTrans.GetMapSource())
-		return types.ActionContinue
+	mapSourceData := make(map[string]MapSourceData)
+	mapSourceData["headers"] = MapSourceData{
+		mapSourceType: "headers",
+		kvs:           hs,
+	}
+	mapSourceData["querys"] = MapSourceData{
+		mapSourceType: "querys",
+		kvs:           qs,
 	}
 
 	if config.reqTrans.IsHeaderChange() {
-		if config.reqTrans.GetMapSource() == "self" {
-			mapSourceData.mapSourceType = "headers"
-			mapSourceData.kvs = hs
-		}
 		if err = config.reqTrans.TransformHeaders(host, path, hs, mapSourceData); err != nil {
 			log.Warnf("failed to transform request headers: %v", err)
 			return types.ActionContinue
@@ -355,10 +363,6 @@ func onHttpRequestHeaders(ctx wrapper.HttpContext, config TransformerConfig, log
 	}
 
 	if config.reqTrans.IsQueryChange() {
-		if config.reqTrans.GetMapSource() == "self" {
-			mapSourceData.mapSourceType = "querys"
-			mapSourceData.kvs = qs
-		}
 		if err = config.reqTrans.TransformQuerys(host, path, qs, mapSourceData); err != nil {
 			log.Warnf("failed to transform request query params: %v", err)
 			return types.ActionContinue
@@ -381,7 +385,7 @@ func onHttpRequestHeaders(ctx wrapper.HttpContext, config TransformerConfig, log
 }
 
 func onHttpRequestBody(ctx wrapper.HttpContext, config TransformerConfig, body []byte, log wrapper.Log) types.Action {
-	if config.reqTrans == nil || !config.reqTrans.IsBodyChange() {
+	if config.reqTrans == nil {
 		return types.ActionContinue
 	}
 
@@ -406,51 +410,80 @@ func onHttpRequestBody(ctx wrapper.HttpContext, config TransformerConfig, body [
 		return types.ActionContinue
 	}
 
-	var mapSourceData MapSourceData
+	mapSourceData := make(map[string]MapSourceData)
 	var hs map[string][]string
 	var qs map[string][]string
-	switch config.reqTrans.GetMapSource() {
-	case "headers":
-		{
-			hs = ctx.GetContext("headers").(map[string][]string)
-			if hs == nil {
-				log.Warn("failed to get request headers")
+
+	hs = ctx.GetContext("headers").(map[string][]string)
+	if hs == nil {
+		log.Warn("failed to get request headers")
+		return types.ActionContinue
+	}
+	if hs[":authority"] == nil {
+		log.Warn(errGetRequestHost.Error())
+		return types.ActionContinue
+	}
+	if hs[":path"] == nil {
+		log.Warn(errGetRequestPath.Error())
+		return types.ActionContinue
+	}
+	mapSourceData["headers"] = MapSourceData{
+		mapSourceType: "headers",
+		kvs:           hs,
+	}
+
+	qs = ctx.GetContext("querys").(map[string][]string)
+	if qs == nil {
+		log.Warn("failed to get request querys")
+		return types.ActionContinue
+	}
+	mapSourceData["querys"] = MapSourceData{
+		mapSourceType: "querys",
+		kvs:           qs,
+	}
+
+	switch structuredBody.(type) {
+	case map[string]interface{}:
+		mapSourceData["body"] = MapSourceData{
+			mapSourceType: "bodyJson",
+			json:          structuredBody.(map[string]interface{})["body"].([]byte),
+		}
+	case map[string][]string:
+		mapSourceData["body"] = MapSourceData{
+			mapSourceType: "bodyKv",
+			kvs:           structuredBody.(map[string][]string),
+		}
+	}
+
+	if ctx.GetContext("need_head_trans") != nil {
+		if config.reqTrans.IsHeaderChange() {
+			if err = config.reqTrans.TransformHeaders(host, path, hs, mapSourceData); err != nil {
+				log.Warnf("failed to transform request headers: %v", err)
 				return types.ActionContinue
 			}
-			if hs[":authority"] == nil {
-				log.Warn(errGetRequestHost.Error())
-				return types.ActionContinue
-			}
-			if hs[":path"] == nil {
-				log.Warn(errGetRequestPath.Error())
-				return types.ActionContinue
-			}
-			mapSourceData.mapSourceType = "headers"
-			mapSourceData.kvs = hs
 		}
 
-	case "querys":
-		{
-			qs = ctx.GetContext("querys").(map[string][]string)
-			if qs == nil {
-				log.Warn("failed to get request querys")
+		if config.reqTrans.IsQueryChange() {
+			if err = config.reqTrans.TransformQuerys(host, path, qs, mapSourceData); err != nil {
+				log.Warnf("failed to transform request query params: %v", err)
 				return types.ActionContinue
 			}
-			mapSourceData.mapSourceType = "querys"
-			mapSourceData.kvs = qs
+			path, err = constructPath(path, qs)
+			if err != nil {
+				log.Warnf("failed to construct path: %v", err)
+				return types.ActionContinue
+			}
+			hs[":path"] = []string{path}
 		}
 
-	case "body", "self":
-		switch structuredBody.(type) {
-		case map[string]interface{}:
-			mapSourceData.mapSourceType = "bodyJson"
-			mapSourceData.json = structuredBody.(map[string]interface{})["body"].([]byte)
-		case map[string][]string:
-			mapSourceData.mapSourceType = "bodyKv"
-			mapSourceData.kvs = structuredBody.(map[string][]string)
+		headers := reconvertHeaders(hs)
+		if err = proxywasm.ReplaceHttpRequestHeaders(headers); err != nil {
+			log.Warnf("failed to replace request headers: %v", err)
+			return types.ActionContinue
 		}
-	default:
-		log.Warnf("invalid mapSource in request body: %v", config.reqTrans.GetMapSource())
+	}
+
+	if !config.reqTrans.IsBodyChange() {
 		return types.ActionContinue
 	}
 
@@ -495,21 +528,28 @@ func onHttpResponseHeaders(ctx wrapper.HttpContext, config TransformerConfig, lo
 	if hs["content-type"] != nil {
 		contentType = hs["content-type"][0]
 	}
-	if config.respTrans.IsBodyChange() && isValidResponseContentType(contentType) {
+	ctx.SetContext("content-type", contentType)
+
+	isValidResponseContent := isValidResponseContentType(contentType)
+	isBodyChange := config.respTrans.IsBodyChange()
+	needBodyMapSource := config.respTrans.NeedBodyMapSource()
+
+	if isBodyChange && isValidResponseContent {
 		delete(hs, "content-length")
-		ctx.SetContext("content-type", contentType)
-	} else {
-		ctx.DontReadResponseBody()
 	}
 
-	var mapSourceData MapSourceData
-	switch config.respTrans.GetMapSource() {
-	case "headers", "self":
-		mapSourceData.mapSourceType = "headers"
-		mapSourceData.kvs = hs
-	default:
-		log.Warnf("invalid mapSource in response header: %v", config.respTrans.GetMapSource())
-		return types.ActionContinue
+	if !isValidResponseContent || (!isBodyChange && !needBodyMapSource) {
+		ctx.DontReadResponseBody()
+	} else if needBodyMapSource {
+		// we need do transform during body phase
+		ctx.SetContext("need_head_trans", struct{}{})
+		return types.HeaderStopIteration
+	}
+
+	mapSourceData := make(map[string]MapSourceData)
+	mapSourceData["headers"] = MapSourceData{
+		mapSourceType: "headers",
+		kvs:           hs,
 	}
 
 	if config.respTrans.IsHeaderChange() {
@@ -529,7 +569,7 @@ func onHttpResponseHeaders(ctx wrapper.HttpContext, config TransformerConfig, lo
 }
 
 func onHttpResponseBody(ctx wrapper.HttpContext, config TransformerConfig, body []byte, log wrapper.Log) types.Action {
-	if config.respTrans == nil || !config.respTrans.IsBodyChange() {
+	if config.respTrans == nil {
 		return types.ActionContinue
 	}
 
@@ -554,30 +594,48 @@ func onHttpResponseBody(ctx wrapper.HttpContext, config TransformerConfig, body
 		return types.ActionContinue
 	}
 
-	var mapSourceData MapSourceData
-	switch config.respTrans.GetMapSource() {
-	case "headers":
-		{
-			hs := ctx.GetContext("headers").(map[string][]string)
-			if hs == nil {
-				log.Warn("failed to get response headers")
+	mapSourceData := make(map[string]MapSourceData)
+	var hs map[string][]string
+
+	hs = ctx.GetContext("headers").(map[string][]string)
+	if hs == nil {
+		log.Warn("failed to get response headers")
+		return types.ActionContinue
+	}
+	mapSourceData["headers"] = MapSourceData{
+		mapSourceType: "headers",
+		kvs:           hs,
+	}
+
+	switch structuredBody.(type) {
+	case map[string]interface{}:
+		mapSourceData["body"] = MapSourceData{
+			mapSourceType: "bodyJson",
+			json:          structuredBody.(map[string]interface{})["body"].([]byte),
+		}
+	case map[string][]string:
+		mapSourceData["body"] = MapSourceData{
+			mapSourceType: "bodyKv",
+			kvs:           structuredBody.(map[string][]string),
+		}
+	}
+
+	if ctx.GetContext("need_head_trans") != nil {
+		if config.respTrans.IsHeaderChange() {
+			if err = config.respTrans.TransformHeaders(host, path, hs, mapSourceData); err != nil {
+				log.Warnf("failed to transform response headers: %v", err)
 				return types.ActionContinue
 			}
-			mapSourceData.mapSourceType = "headers"
-			mapSourceData.kvs = hs
 		}
 
-	case "body", "self":
-		switch structuredBody.(type) {
-		case map[string]interface{}:
-			mapSourceData.mapSourceType = "bodyJson"
-			mapSourceData.json = structuredBody.(map[string]interface{})["body"].([]byte)
-		case map[string][]string:
-			mapSourceData.mapSourceType = "bodyKv"
-			mapSourceData.kvs = structuredBody.(map[string][]string)
+		headers := reconvertHeaders(hs)
+		if err = proxywasm.ReplaceHttpResponseHeaders(headers); err != nil {
+			log.Warnf("failed to replace response headers: %v", err)
+			return types.ActionContinue
 		}
-	default:
-		log.Warnf("invalid mapSource in response body: %v", config.respTrans.GetMapSource())
+	}
+
+	if !config.respTrans.IsBodyChange() {
 		return types.ActionContinue
 	}
 
@@ -657,36 +715,34 @@ func newTransformRule(rules []gjson.Result) (res []TransformRule, err error) {
 }
 
 type Transformer interface {
-	TransformHeaders(host, path string, hs map[string][]string, mapSourceData MapSourceData) error
-	TransformQuerys(host, path string, qs map[string][]string, mapSourceData MapSourceData) error
-	TransformBody(host, path string, body interface{}, mapSourceData MapSourceData) error
+	TransformHeaders(host, path string, hs map[string][]string, mapSourceData map[string]MapSourceData) error
+	TransformQuerys(host, path string, qs map[string][]string, mapSourceData map[string]MapSourceData) error
+	TransformBody(host, path string, body interface{}, mapSourceData map[string]MapSourceData) error
 	IsHeaderChange() bool
 	IsQueryChange() bool
 	IsBodyChange() bool
-	GetMapSource() string
+	NeedBodyMapSource() bool
 }
 
 var _ Transformer = (*requestTransformer)(nil)
 var _ Transformer = (*responseTransformer)(nil)
 
 type requestTransformer struct {
-	headerHandler  *kvHandler
-	queryHandler   *kvHandler
-	bodyHandler    *requestBodyHandler
-	isHeaderChange bool
-	isQueryChange  bool
-	isBodyChange   bool
-	// 目前插件在对request做map转换的时候只支持最多一个映射来源
-	// 取值：headers，querys，body，self
-	mapSource string
+	headerHandler     *kvHandler
+	queryHandler      *kvHandler
+	bodyHandler       *requestBodyHandler
+	isHeaderChange    bool
+	isQueryChange     bool
+	isBodyChange      bool
+	needBodyMapSource bool
 }
 
 func newRequestTransformer(config *TransformerConfig) (Transformer, error) {
-	headerKvtGroup, isHeaderChange, withHeaderMapKvt, err := newKvtGroup(config.reqRules, "headers")
+	headerKvtGroup, isHeaderChange, _, err := newKvtGroup(config.reqRules, "headers")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to new kvt group for headers")
 	}
-	queryKvtGroup, isQueryChange, withQueryMapKvt, err := newKvtGroup(config.reqRules, "querys")
+	queryKvtGroup, isQueryChange, _, err := newKvtGroup(config.reqRules, "querys")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to new kvt group for querys")
 	}
@@ -695,12 +751,7 @@ func newRequestTransformer(config *TransformerConfig) (Transformer, error) {
 		return nil, errors.Wrap(err, "failed to new kvt group for body")
 	}
 
-	mapSource := getMapSourceFromRule(config.reqRules)
-
-	// TODO: not support mapping headers or querys from body in requestTransformer before #582 is fixed
-	if mapSource == "body" && (withHeaderMapKvt || withQueryMapKvt) {
-		return nil, errors.Wrap(err, "not support mapping headers or querys from body in requestTransformer")
-	}
+	bodyMapSource := bodyMapSourceInRule(config.reqRules)
 
 	return &requestTransformer{
 		headerHandler: &kvHandler{headerKvtGroup},
@@ -709,22 +760,22 @@ func newRequestTransformer(config *TransformerConfig) (Transformer, error) {
 			formDataHandler: &kvHandler{bodyKvtGroup},
 			jsonHandler:     &jsonHandler{bodyKvtGroup},
 		},
-		isHeaderChange: isHeaderChange,
-		isQueryChange:  isQueryChange,
-		isBodyChange:   isBodyChange,
-		mapSource:      mapSource,
+		isHeaderChange:    isHeaderChange,
+		isQueryChange:     isQueryChange,
+		isBodyChange:      isBodyChange,
+		needBodyMapSource: bodyMapSource,
 	}, nil
 }
 
-func (t requestTransformer) TransformHeaders(host, path string, hs map[string][]string, mapSourceData MapSourceData) error {
+func (t requestTransformer) TransformHeaders(host, path string, hs map[string][]string, mapSourceData map[string]MapSourceData) error {
 	return t.headerHandler.handle(host, path, hs, mapSourceData)
 }
 
-func (t requestTransformer) TransformQuerys(host, path string, qs map[string][]string, mapSourceData MapSourceData) error {
+func (t requestTransformer) TransformQuerys(host, path string, qs map[string][]string, mapSourceData map[string]MapSourceData) error {
 	return t.queryHandler.handle(host, path, qs, mapSourceData)
 }
 
-func (t requestTransformer) TransformBody(host, path string, body interface{}, mapSourceData MapSourceData) error {
+func (t requestTransformer) TransformBody(host, path string, body interface{}, mapSourceData map[string]MapSourceData) error {
 	switch body.(type) {
 	case map[string][]string:
 		return t.bodyHandler.formDataHandler.handle(host, path, body.(map[string][]string), mapSourceData)
@@ -744,23 +795,22 @@ func (t requestTransformer) TransformBody(host, path string, body interface{}, m
 	return nil
 }
 
-func (t requestTransformer) IsHeaderChange() bool { return t.isHeaderChange }
-func (t requestTransformer) IsQueryChange() bool  { return t.isQueryChange }
-func (t requestTransformer) IsBodyChange() bool   { return t.isBodyChange }
-func (t requestTransformer) GetMapSource() string { return t.mapSource }
+func (t requestTransformer) IsHeaderChange() bool    { return t.isHeaderChange }
+func (t requestTransformer) IsQueryChange() bool     { return t.isQueryChange }
+func (t requestTransformer) IsBodyChange() bool      { return t.isBodyChange }
+func (t requestTransformer) NeedBodyMapSource() bool { return t.needBodyMapSource }
 
 type responseTransformer struct {
-	headerHandler  *kvHandler
-	bodyHandler    *responseBodyHandler
-	isHeaderChange bool
-	isBodyChange   bool
-	// 目前插件在对response做map转换的时候只支持最多一个映射来源
-	mapSource string
+	headerHandler     *kvHandler
+	bodyHandler       *responseBodyHandler
+	isHeaderChange    bool
+	isBodyChange      bool
+	needBodyMapSource bool
 }
 
 func newResponseTransformer(config *TransformerConfig) (Transformer, error) {
 
-	headerKvtGroup, isHeaderChange, withHeaderMapKvt, err := newKvtGroup(config.respRules, "headers")
+	headerKvtGroup, isHeaderChange, _, err := newKvtGroup(config.respRules, "headers")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to new kvt group for headers")
 	}
@@ -768,30 +818,27 @@ func newResponseTransformer(config *TransformerConfig) (Transformer, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to new kvt group for body")
 	}
-	mapSource := getMapSourceFromRule(config.respRules)
-	// TODO: not support mapping headers from body in responseTransformer before #582 is fixed
-	if mapSource == "body" && withHeaderMapKvt {
-		return nil, errors.Wrap(err, "not support mapping headers from body in responseTransformer")
-	}
+	bodyMapSource := bodyMapSourceInRule(config.respRules)
+
 	return &responseTransformer{
-		headerHandler:  &kvHandler{headerKvtGroup},
-		bodyHandler:    &responseBodyHandler{&jsonHandler{bodyKvtGroup}},
-		isHeaderChange: isHeaderChange,
-		isBodyChange:   isBodyChange,
-		mapSource:      mapSource,
+		headerHandler:     &kvHandler{headerKvtGroup},
+		bodyHandler:       &responseBodyHandler{&jsonHandler{bodyKvtGroup}},
+		isHeaderChange:    isHeaderChange,
+		isBodyChange:      isBodyChange,
+		needBodyMapSource: bodyMapSource,
 	}, nil
 }
 
-func (t responseTransformer) TransformHeaders(host, path string, hs map[string][]string, mapSourceData MapSourceData) error {
+func (t responseTransformer) TransformHeaders(host, path string, hs map[string][]string, mapSourceData map[string]MapSourceData) error {
 	return t.headerHandler.handle(host, path, hs, mapSourceData)
 }
 
-func (t responseTransformer) TransformQuerys(host, path string, qs map[string][]string, mapSourceData MapSourceData) error {
+func (t responseTransformer) TransformQuerys(host, path string, qs map[string][]string, mapSourceData map[string]MapSourceData) error {
 	// the response does not need to transform the query params, always returns nil
 	return nil
 }
 
-func (t responseTransformer) TransformBody(host, path string, body interface{}, mapSourceData MapSourceData) error {
+func (t responseTransformer) TransformBody(host, path string, body interface{}, mapSourceData map[string]MapSourceData) error {
 	switch body.(type) {
 	case map[string]interface{}:
 		m := body.(map[string]interface{})
@@ -808,10 +855,10 @@ func (t responseTransformer) TransformBody(host, path string, body interface{},
 	return nil
 }
 
-func (t responseTransformer) IsHeaderChange() bool { return t.isHeaderChange }
-func (t responseTransformer) IsQueryChange() bool  { return false } // the response does not need to transform the query params, always returns false
-func (t responseTransformer) IsBodyChange() bool   { return t.isBodyChange }
-func (t responseTransformer) GetMapSource() string { return t.mapSource }
+func (t responseTransformer) IsHeaderChange() bool    { return t.isHeaderChange }
+func (t responseTransformer) IsQueryChange() bool     { return false } // the response does not need to transform the query params, always returns false
+func (t responseTransformer) IsBodyChange() bool      { return t.isBodyChange }
+func (t responseTransformer) NeedBodyMapSource() bool { return t.needBodyMapSource }
 
 type requestBodyHandler struct {
 	formDataHandler *kvHandler
@@ -830,7 +877,7 @@ type jsonHandler struct {
 	kvtOps []kvtOperation
 }
 
-func (h kvHandler) handle(host, path string, kvs map[string][]string, mapSourceData MapSourceData) error {
+func (h kvHandler) handle(host, path string, kvs map[string][]string, mapSourceData map[string]MapSourceData) error {
 	// arbitary order. for example: remove → rename → replace → add → append → map → dedupe
 
 	for _, kvtOp := range h.kvtOps {
@@ -887,17 +934,29 @@ func (h kvHandler) handle(host, path string, kvs map[string][]string, mapSourceD
 			// map: 若指定 fromKey 不存在则无操作；否则将 fromKey 的值映射给 toKey 的值
 			for _, map_ := range kvtOp.mapKvtGroup {
 				fromKey, toKey := map_.fromKey, map_.toKey
-				if mapSourceData.mapSourceType == "headers" {
+				if kvtOp.mapSource == "headers" {
 					fromKey = strings.ToLower(fromKey)
 				}
-				if fromValue, ok := mapSourceData.search(fromKey); ok {
-					switch mapSourceData.mapSourceType {
+				source, exist := mapSourceData[kvtOp.mapSource]
+				if !exist {
+					proxywasm.LogWarnf("map key failed, source:%s not exists", kvtOp.mapSource)
+					continue
+				}
+				proxywasm.LogDebugf("search key:%s in source:%s", fromKey, kvtOp.mapSource)
+				if fromValue, ok := source.search(fromKey); ok {
+					switch source.mapSourceType {
 					case "headers", "querys", "bodyKv":
 						kvs[toKey] = fromValue.([]string)
-						// TODO: not support mapping headers or querys from body  before #582 is fixed
-						// case "bodyJson":
-						// 	kvs[toKey] = fromValue
-						// }
+						proxywasm.LogDebugf("map key:%s to key:%s success, value is: %v", fromKey, toKey, fromValue)
+
+					case "bodyJson":
+						if valueJson, ok := fromValue.(gjson.Result); ok {
+							valueStr := valueJson.String()
+							if valueStr != "" {
+								kvs[toKey] = []string{valueStr}
+								proxywasm.LogDebugf("map key:%s to key:%s success, values is:%s", fromKey, toKey, valueStr)
+							}
+						}
 					}
 				}
 			}
@@ -938,7 +997,7 @@ func (h kvHandler) handle(host, path string, kvs map[string][]string, mapSourceD
 }
 
 // only for body
-func (h jsonHandler) handle(host, path string, oriData []byte, mapSourceData MapSourceData) (data []byte, err error) {
+func (h jsonHandler) handle(host, path string, oriData []byte, mapSourceData map[string]MapSourceData) (data []byte, err error) {
 	// arbitary order. for example: remove → rename → replace → add → append → map → dedupe
 	if !gjson.ValidBytes(oriData) {
 		return nil, errors.New("invalid json body")
@@ -1054,14 +1113,30 @@ func (h jsonHandler) handle(host, path string, oriData []byte, mapSourceData Map
 			// map: 若指定 fromKey 不存在则无操作；否则将 fromKey 的值映射给 toKey 的值
 			for _, map_ := range kvtOp.mapKvtGroup {
 				fromKey, toKey := map_.fromKey, map_.toKey
-				if mapSourceData.mapSourceType == "headers" {
+				if kvtOp.mapSource == "headers" {
 					fromKey = strings.ToLower(fromKey)
 				}
-				if fromValue, ok := mapSourceData.search(fromKey); ok {
-					// search返回的类型为[]string或者gjson.Result.Value()
-					// sjson.SetBytes()能够直接处理[]byte，其他更复杂的数据类型均会json.Marshall化
-					if data, err = sjson.SetBytes(data, toKey, fromValue); err != nil {
-						return nil, errors.Wrap(err, errMap.Error())
+				source, exist := mapSourceData[kvtOp.mapSource]
+				if !exist {
+					proxywasm.LogWarnf("map key failed, source:%s not exists", kvtOp.mapSource)
+					continue
+				}
+
+				proxywasm.LogDebugf("search key:%s in source:%s", fromKey, kvtOp.mapSource)
+				if fromValue, ok := source.search(fromKey); ok {
+					switch source.mapSourceType {
+					case "headers", "querys", "bodyKv":
+						if data, err = sjson.SetBytes(data, toKey, fromValue); err != nil {
+							return nil, errors.Wrap(err, errMap.Error())
+						}
+						proxywasm.LogDebugf("map key:%s to key:%s success, value is: %v", fromKey, toKey, fromValue)
+					case "bodyJson":
+						if valueJson, ok := fromValue.(gjson.Result); ok {
+							if data, err = sjson.SetBytes(data, toKey, valueJson.Value()); err != nil {
+								return nil, errors.Wrap(err, errMap.Error())
+							}
+							proxywasm.LogDebugf("map key:%s to key:%s success, value is: %v", fromKey, toKey, fromValue)
+						}
 					}
 				}
 			}
@@ -1226,10 +1301,17 @@ func newKvtGroup(rules []TransformRule, typ string) (g []kvtOperation, isChange
 				kvtOp.renameKvtGroup = append(kvtOp.renameKvtGroup, renameKvt{p.renameParam.oldKey, p.renameParam.newKey, p.valueType})
 			case "map":
 				if typ == "headers" {
-					p.mapParam.fromKey = strings.ToLower(p.mapParam.fromKey)
 					p.mapParam.toKey = strings.ToLower(p.mapParam.toKey)
 				}
 				kvtOp.mapSource = r.mapSource
+				if kvtOp.mapSource == "self" {
+					kvtOp.mapSource = typ
+					r.mapSource = typ
+				}
+				if kvtOp.mapSource == "headers" {
+					p.mapParam.fromKey = strings.ToLower(p.mapParam.fromKey)
+				}
+
 				kvtOp.mapKvtGroup = append(kvtOp.mapKvtGroup, mapKvt{p.mapParam.fromKey, p.mapParam.toKey})
 			case "dedupe":
 				if typ == "headers" {
@@ -1301,20 +1383,19 @@ func (msdata MapSourceData) search(fromKey string) (interface{}, bool) {
 		if !fromValue.Exists() {
 			return nil, false
 		}
-		return fromValue.Value(), true
+		return fromValue, true
 	default:
 		return "", false
 	}
 }
 
-func getMapSourceFromRule(rules []TransformRule) string {
-	// 如果rules中不含map转换要求，则返回空字符串
+func bodyMapSourceInRule(rules []TransformRule) bool {
 	for _, r := range rules {
-		if r.operate == "map" {
-			return r.mapSource
+		if r.operate == "map" && r.mapSource == "body" {
+			return true
 		}
 	}
-	return ""
+	return false
 }
 
 type kvtReg struct {

plugins/wasm-go/extensions/transformer/utils.go

@@ -21,6 +21,7 @@ import (
 	"mime"
 	"mime/multipart"
 	"net/url"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -263,5 +264,8 @@ func reconvertHeaders(hs map[string][]string) [][2]string {
 			ret = append(ret, [2]string{k, v})
 		}
 	}
+	sort.SliceStable(ret, func(i, j int) bool {
+		return ret[i][0] < ret[j][0]
+	})
 	return ret
 }