add kafka backend and header to json filter to http pipeline

doc/reference/filters.md

@@ -46,6 +46,12 @@
   - [WasmHost](#wasmhost)
     - [Configuration](#configuration-14)
     - [Results](#results-14)
+  - [Kafka](#kafka)
+    - [Configuration](#configuration-15)
+    - [Results](#results-15)
+  - [HeaderToJSON](#headertojson)
+    - [Configuration](#configuration-16)
+    - [Results](#results-16)
   - [Common Types](#common-types)
     - [apiaggregator.Pipeline](#apiaggregatorpipeline)
     - [pathadaptor.Spec](#pathadaptorspec)
@@ -76,6 +82,8 @@
     - [validator.OAuth2ValidatorSpec](#validatoroauth2validatorspec)
     - [validator.OAuth2TokenIntrospect](#validatoroauth2tokenintrospect)
     - [validator.OAuth2JWT](#validatoroauth2jwt)
+    - [kafka.Topic](#kafkatopic)
+    - [headertojson.HeaderMap](#headertojsonheadermap)
 
 A Filter is a request/response processor. Multiple filters can be orchestrated together to form a pipeline, each filter returns a string result after it finishes processing the input request/response. An empty result means the input was successfully processed by the current filter and can go forward to the next filter in the pipeline, while a non-empty result means the pipeline or preceding filter need to take extra action.
 
@@ -663,6 +671,68 @@ $ make wasm
 | ...                                                                         |
 | wasmResult9                                                                 |
 
+
+
+## Kafka
+
+The Kafka filter converts HTTP Requests to Kafka messages and sends them to the Kafka backend. The topic of the Kafka message comes from the HTTP header, if not found, then the default topic will be used. The payload of the Kafka message comes from the body of the HTTP Request.
+
+Below is an example configuration. 
+
+```yaml
+kind: Kafka
+name: kafka-example
+backend: [":9093"] 
+topic:
+  default: kafka-topic
+  dynamic:
+    header: X-Kafka-Topic
+```
+
+### Configuration
+
+| Name         | Type     | Description                      | Required |
+| ------------ | -------- | -------------------------------- | -------- |
+| backend | []string | Addresses of Kafka backend | Yes      |
+| topic | [Kafka.Topic](#kafkatopic) | the topic is Spec used to get Kafka topic used to send message to the backend | Yes      |
+
+
+### Results
+
+| Value                   | Description                          |
+| ----------------------- | ------------------------------------ |
+| parseErr     | Failed to get Kafka message from the HTTP request |
+
+## HeaderToJSON
+
+The HeaderToJSON converts HTTP headers to JSON and combines it with the HTTP request body. To use this filter, make sure your HTTP Request body is empty or JSON schema.
+
+Below is an example configuration. 
+
+```yaml
+kind: HeaderToJSON
+name: headertojson-example
+headerMap:
+  - header: X-User-Name
+    json: username
+  - header: X-Type
+    json: type
+```
+
+### Configuration
+
+| Name         | Type     | Description                      | Required |
+| ------------ | -------- | -------------------------------- | -------- |
+| headerMap | [][HeaderToJSON.HeaderMap](#headertojsonheadermap) | headerMap defines a map between HTTP header name and corresponding JSON filed name
+  | Yes      |
+
+
+### Results
+
+| Value                   | Description                          |
+| ----------------------- | ------------------------------------ |
+| jsonEncodeDecodeErr     | Failed to convert HTTP headers to JSON. |
+
 ## Common Types
 
 ### apiaggregator.Pipeline
@@ -935,3 +1005,17 @@ The relationship between `methods` and `url` is `AND`.
 | --------- | ------ | ------------------------------------------------------------------------ | -------- |
 | algorithm | string | The algorithm for validation, `HS256`, `HS384` and `HS512` are supported | Yes      |
 | secret    | string | The secret for validation, in hex encoding                               | Yes      |
+
+### kafka.Topic
+
+| Name      | Type   | Description                                                              | Required |
+| --------- | ------ | ------------------------------------------------------------------------ | -------- |
+| default | string | Default topic for Kafka backend | Yes      |
+| dynamic.header | string | The HTTP header that contains Kafka topic | Yes      |
+
+### headertojson.HeaderMap
+
+| Name      | Type   | Description                                                              | Required |
+| --------- | ------ | ------------------------------------------------------------------------ | -------- |
+| header | string | The HTTP header that contains JSON value   | Yes      |
+| json    | string | The field name to put JSON value into HTTP body | Yes      |

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-chi/chi/v5 v5.0.3
 	github.com/go-zookeeper/zk v1.0.2
+	github.com/goccy/go-json v0.9.1
 	github.com/golang-jwt/jwt v3.2.1+incompatible
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/websocket v1.4.2

go.sum

@@ -542,6 +542,8 @@ github.com/go-zookeeper/zk v1.0.2 h1:4mx0EYENAdX/B/rbunjlt5+4RTA/a9SMHBRuSKdGxPM
 github.com/go-zookeeper/zk v1.0.2/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/gobuffalo/flect v0.2.3/go.mod h1:vmkQwuZYhN5Pc4ljYQZzP+1sq+NEkK+lh20jmEmX3jc=
 github.com/gobuffalo/here v0.6.0/go.mod h1:wAG085dHOYqUpf+Ap+WOdrPTp5IYcDAs/x7PLa8Y5fM=
+github.com/goccy/go-json v0.9.1 h1:xurvfvj3gq6SWUkkZ0opoUDTms7jif41uZ9z9s+hVlY=
+github.com/goccy/go-json v0.9.1/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
 github.com/godbus/dbus v0.0.0-20151105175453-c7fdd8b5cd55/go.mod h1:/YcGZj5zSblfDWMMoOzV4fas9FZnQYTkDnsGvmh2Grw=
 github.com/godbus/dbus v0.0.0-20180201030542-885f9cc04c9c/go.mod h1:/YcGZj5zSblfDWMMoOzV4fas9FZnQYTkDnsGvmh2Grw=
 github.com/godbus/dbus v0.0.0-20190422162347-ade71ed3457e/go.mod h1:bBOAhwG1umN6/6ZUMtDFBMQR8jRg9O75tm9K00oMsK4=

pkg/filter/headertojson/headertojson.go

@@ -0,0 +1,145 @@
+/*
+ * 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 headertojson
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+
+	json "github.com/goccy/go-json"
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+)
+
+const (
+	// Kind is the kind of Kafka
+	Kind = "HeaderToJSON"
+
+	resultJSONEncodeDecodeErr = "jsonEncodeDecodeErr"
+)
+
+func init() {
+	httppipeline.Register(&HeaderToJSON{})
+}
+
+type (
+	// HeaderToJSON put http request headers into body as JSON fields.
+	HeaderToJSON struct {
+		filterSpec *httppipeline.FilterSpec
+		spec       *Spec
+		headerMap  map[string]string
+	}
+)
+
+var _ httppipeline.Filter = (*HeaderToJSON)(nil)
+
+// Kind return kind of HeaderToJSON
+func (h *HeaderToJSON) Kind() string {
+	return Kind
+}
+
+// DefaultSpec return default spec of HeaderToJSON
+func (h *HeaderToJSON) DefaultSpec() interface{} {
+	return &Spec{}
+}
+
+// Description return description of HeaderToJSON
+func (h *HeaderToJSON) Description() string {
+	return "HeaderToJSON convert http request header to json"
+}
+
+// Results return possible results of HeaderToJSON
+func (h *HeaderToJSON) Results() []string {
+	return []string{resultJSONEncodeDecodeErr}
+}
+
+func (h *HeaderToJSON) init() {
+	h.headerMap = make(map[string]string)
+	for _, header := range h.spec.HeaderMap {
+		h.headerMap[http.CanonicalHeaderKey(header.Header)] = header.JSON
+	}
+}
+
+// Init init HeaderToJSON
+func (h *HeaderToJSON) Init(filterSpec *httppipeline.FilterSpec) {
+	h.filterSpec, h.spec = filterSpec, filterSpec.FilterSpec().(*Spec)
+	h.init()
+}
+
+// Inherit init HeaderToJSON based on previous generation
+func (h *HeaderToJSON) Inherit(filterSpec *httppipeline.FilterSpec, previousGeneration httppipeline.Filter) {
+	previousGeneration.Close()
+	h.Init(filterSpec)
+}
+
+// Close close HeaderToJSON
+func (h *HeaderToJSON) Close() {
+}
+
+// Status return status of HeaderToJSON
+func (h *HeaderToJSON) Status() interface{} {
+	return nil
+}
+
+func (h *HeaderToJSON) encodeJSON(input map[string]interface{}) ([]byte, error) {
+	return json.Marshal(input)
+}
+
+func (h *HeaderToJSON) decodeJSON(ctx context.HTTPContext) (map[string]interface{}, error) {
+	res := make(map[string]interface{})
+	decoder := json.NewDecoder(ctx.Request().Body())
+	err := decoder.Decode(&res)
+	if err != nil && err != io.EOF {
+		return nil, err
+	}
+	return res, nil
+}
+
+// Handle handle HTTPContext
+func (h *HeaderToJSON) Handle(ctx context.HTTPContext) string {
+	result := h.handle(ctx)
+	return ctx.CallNextHandler(result)
+}
+
+func (h *HeaderToJSON) handle(ctx context.HTTPContext) string {
+	headerMap := make(map[string]interface{})
+	for header, json := range h.headerMap {
+		value := ctx.Request().Header().Get(header)
+		if value != "" {
+			headerMap[json] = value
+		}
+	}
+	if len(headerMap) == 0 {
+		return ""
+	}
+
+	bodyMap, err := h.decodeJSON(ctx)
+	if err != nil {
+		return resultJSONEncodeDecodeErr
+	}
+	for k, v := range headerMap {
+		bodyMap[k] = v
+	}
+	body, err := h.encodeJSON(bodyMap)
+	if err != nil {
+		return resultJSONEncodeDecodeErr
+	}
+	ctx.Request().SetBody(bytes.NewReader(body))
+	return ""
+}