add filter WasmHost (close #1) (#116)

* add wasm filter (close #1)

* [wasm] update according to comment

* update according to review comments

* add status

* fix a bug and make the case of 'wasm` consistent

* rename WasmFilter to WasmHost

Co-authored-by: Bomin Zhang <bomin.zhang@megaease.com>

Makefile

@@ -22,6 +22,18 @@ endif
 # Build Flags
 GO_LD_FLAGS= "-s -w -X github.com/megaease/easegress/pkg/version.RELEASE=${RELEASE} -X github.com/megaease/easegress/pkg/version.COMMIT=${GIT_COMMIT} -X github.com/megaease/easegress/pkg/version.REPO=${GIT_REPO_INFO}"
 
+# Cgo is disabled by default
+ENABLE_CGO= CGO_ENABLED=0
+
+# Check Go build tags, the tags are from command line of make
+ifdef GOTAGS
+  GO_BUILD_TAGS= -tags ${GOTAGS}
+  # Must enable Cgo when wasmhost is included
+  ifeq ($(findstring wasmhost,${GOTAGS}), wasmhost)
+	ENABLE_CGO= CGO_ENABLED=1
+  endif
+endif
+
 # Targets
 TARGET_SERVER=${RELEASE_DIR}/easegress-server
 TARGET_CLIENT=${RELEASE_DIR}/egctl
@@ -38,7 +50,7 @@ build_client:
 build_server:
 	@echo "build server"
 	cd ${MKFILE_DIR} && \
-	CGO_ENABLED=0 go build -v -trimpath -ldflags ${GO_LD_FLAGS} \
+	${ENABLE_CGO} go build ${GO_BUILD_TAGS} -v -trimpath -ldflags ${GO_LD_FLAGS} \
 	-o ${TARGET_SERVER} ${MKFILE_DIR}cmd/server
 
 dev_build: dev_build_client dev_build_server

README.md

@@ -69,6 +69,8 @@ The architecture of Easegress:
 		- **FaaS** integrates with the serverless platform Knative.
 		- **Service Discovery** integrates with Eureka, Consul, Etcd, and Zookeeper.
 		- **Ingress Controller** integrates with Kubernetes as an ingress controller.
+- **Extensibility**
+    - **WebAssembly** executes user developed [WebAssembly](https://webassembly.org/) code.
 - **High Performance and Availability**
 	- **Adaption**: adapts request, response in the handling chain.
 	- **Validation**: headers validation, OAuth2, JWT, and HMAC verification.

doc/filters.md

@@ -43,6 +43,9 @@
   - [Validator](#validator)
     - [Configuration](#configuration-13)
     - [Results](#results-13)
+  - [WasmHost](#wasmhost)
+    - [Configuration](#configuration-14)
+    - [Results](#results-14)
   - [Common Types](#common-types)
     - [apiaggregator.APIProxy](#apiaggregatorapiproxy)
     - [pathadaptor.Spec](#pathadaptorspec)
@@ -598,6 +601,48 @@ oauth2:
 | ------- | ----------------------------------- |
 | invalid | The request doesn't pass validation |
 
+## WasmHost
+
+The WasmHost filter implements a host environment for user-developed [WebAssembly](https://webassembly.org/) code. Below is an example configuration that loads wasm code from a file.
+
+```yaml
+name: wasm-host-example
+kind: WasmHost
+maxConcurrency: 2
+code: /home/megaease/wasm/hello.wasm
+timeout: 200ms
+```
+
+Note: this filter is disabled in the default build of `Easegress`, it can be enabled by:
+
+```bash
+$ make GOTAGS=wasmhost
+```
+
+or
+
+```bash
+$ go build -tags=wasmhost
+```
+
+### Configuration
+
+| Name           | Type   | Description    | Required |
+| -------------- | ------ |--------------- | -------- |
+| maxConcurrency | int32  | The maximum requests the filter can process concurrently. Default is 10 and minimum value is 1. | Yes       |
+| code           | string | The wasm code, can be the base64 encoded code, or path/url of the file which contains the code. | Yes    |
+| timeout        | string | Timeout for wasm execution, default is 100ms. | Yes     |
+
+### Results
+
+| Value       | Description                         |
+| ----------- | ----------------------------------- |
+| outOfVM     | Can not found an available wasm VM. |
+| wasmError   | An error occurs during the execution of wasm code. |
+| wasmResult1 <td rowspan="3">Results defined and returned by wasm code.</td>
+|     ...      
+| wasmResult9 
+
 ## Common Types
 
 ### apiaggregator.APIProxy

doc/reference.md

@@ -12,4 +12,5 @@
   * [TimeLimiter](./filters.md#TimeLimiter)
   * [Retryer](./filters.md#Retryer)
   * [ResponseAdaptor](./filters.md#ResponseAdaptor)
-  * [Validator](./filters.md#Validator)
+  * [Validator](./filters.md#Validator)
+  * [WasmHost](./filters.md#WasmHost)

example/backend-service/mirror/mirror.go

@@ -3,38 +3,55 @@ package main
 import (
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net/http"
+	"os"
 	"time"
 )
 
+type TeeWriter struct {
+	writers []io.Writer
+}
+
+func NewTeeWriter(writers ...io.Writer) *TeeWriter {
+	return &TeeWriter{writers: writers}
+}
+
+func (tw *TeeWriter) Write(p []byte) (n int, err error) {
+	for _, w := range tw.writers {
+		w.Write(p)
+	}
+	return len(p), nil
+}
+
 func main() {
 	helloHandler := func(w http.ResponseWriter, req *http.Request) {
 		io.WriteString(w, "hello")
 	}
 	mirrorHandler := func(w http.ResponseWriter, req *http.Request) {
 		time.Sleep(10 * time.Millisecond)
-		body, err := ioutil.ReadAll(req.Body)
+		body, err := io.ReadAll(req.Body)
 		if err != nil {
 			body = []byte(fmt.Sprintf("<read failed: %v>", err))
 		}
 
+		tw := NewTeeWriter(w, os.Stdout)
+
 		url := req.URL.Path
 		if req.URL.Query().Encode() != "" {
 			url += "?" + req.URL.Query().Encode()
 		}
 
-		content := fmt.Sprintf(`Your Request
-===============
-Method: %s
-URL   : %s
-Header: %v
-Body  : %s`, req.Method, url, req.Header, body)
+		fmt.Fprintln(tw, "Your Request")
+		fmt.Fprintln(tw, "==============")
+		fmt.Fprintln(tw, "Method:", req.Method)
+		fmt.Fprintln(tw, "URL   :", url)
 
-		// fmt.Printf("%s: %d bytes body received, %d bytes body sent\n",
-		// 	req.Host, len(body), len(content))
+		fmt.Fprintln(tw, "Header:")
+		for k, v := range req.Header {
+			fmt.Fprintf(tw, "    %s: %v\n", k, v)
+		}
 
-		io.WriteString(w, content)
+		fmt.Fprintln(tw, "Body  :", string(body))
 	}
 
 	http.HandleFunc("/", mirrorHandler)

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/ArthurHlt/go-eureka-client v1.1.0
 	github.com/Shopify/sarama v1.29.1
 	github.com/alecthomas/jsonschema v0.0.0-20210526225647-edb03dcab7bc
+	github.com/bytecodealliance/wasmtime-go v0.28.0
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/facebookgo/ensure v0.0.0-20200202191622-63f1cf65ac4c // indirect
 	github.com/facebookgo/freeport v0.0.0-20150612182905-d4adf43b75b9 // indirect

go.sum

@@ -203,6 +203,8 @@ github.com/c2h5oh/datasize v0.0.0-20171227191756-4eba002a5eae/go.mod h1:S/7n9cop
 github.com/c2h5oh/datasize v0.0.0-20200112174442-28bbd4740fee/go.mod h1:S/7n9copUssQ56c7aAgHqftWO4LTf4xY6CGWt8Bc+3M=
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
+github.com/bytecodealliance/wasmtime-go v0.28.0 h1:JTWP482wkmR79O9T0JiIAllPqmNW5oP0v56v/FwCpaQ=
+github.com/bytecodealliance/wasmtime-go v0.28.0/go.mod h1:q320gUxqyI8yB+ZqRuaJOEnGkAnHh6WtJjMaT2CW4wI=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.3.0 h1:t/LhUZLVitR1Ow2YOnduCsavhwFUklBMoGVYUCqmCqk=
 github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=

pkg/api/api.go

@@ -48,6 +48,8 @@ var (
 	apisMutex      = sync.Mutex{}
 	apis           = make(map[string]*APIGroup)
 	apisChangeChan = make(chan struct{}, 10)
+
+	appendAddonAPIs []func(s *Server, group *APIGroup)
 )
 
 type apisbyOrder []*APIGroup
@@ -98,6 +100,10 @@ func (s *Server) registerAPIs() {
 	group.Entries = append(group.Entries, s.healthAPIEntries()...)
 	group.Entries = append(group.Entries, s.aboutAPIEntries()...)
 
+	for _, fn := range appendAddonAPIs {
+		fn(s, group)
+	}
+
 	RegisterAPIs(group)
 }
 

pkg/api/wasm.go

@@ -0,0 +1,53 @@
+// +build wasmhost
+
+/*
+ * 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 api
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+)
+
+func (s *Server) postWasmCodeUpdateEvent() string {
+	key := s.cluster.Layout().WasmCodeEvent()
+	value := time.Now().Format(time.RFC3339Nano)
+	if e := s.cluster.Put(key, value); e != nil {
+		ClusterPanic(e)
+	}
+	return value
+}
+
+func appendWasmAPI(s *Server, group *APIGroup) {
+	entry := &APIEntry{
+		Path:   "/wasm/code",
+		Method: "POST",
+		Handler: func(w http.ResponseWriter, r *http.Request) {
+			v := s.postWasmCodeUpdateEvent()
+			w.Header().Set("Content-Type", "text/plain")
+			fmt.Fprintf(w, "wasm code update event posted at: %s\n", v)
+		},
+	}
+
+	group.Entries = append(group.Entries, entry)
+}
+
+func init() {
+	appendAddonAPIs = append(appendAddonAPIs, appendWasmAPI)
+}

pkg/cluster/layout.go

@@ -32,6 +32,7 @@ const (
 	configObjectPrefix       = "/config/objects/"
 	configObjectFormat       = "/config/objects/%s" // +objectName
 	configVersion            = "/config/version"
+	wasmCodeEvent            = "/wasm/code"
 
 	// the cluster name of this eg group will be registered under this path in etcd
 	// any new member(reader or writer ) will be rejected if it is configured a different cluster name
@@ -116,3 +117,8 @@ func (l *Layout) ConfigObjectKey(name string) string {
 func (l *Layout) ConfigVersion() string {
 	return configVersion
 }
+
+// WasmCodeEvent returns the key of wasm code event
+func (l *Layout) WasmCodeEvent() string {
+	return wasmCodeEvent
+}

pkg/cluster/op.go

@@ -167,8 +167,8 @@ func (c *cluster) GetRawPrefix(prefix string) (map[string]*mvccpb.KeyValue, erro
 		return kvs, err
 	}
 
-	for idx, kv := range resp.Kvs {
-		kvs[string(kv.Key)] = resp.Kvs[idx]
+	for _, kv := range resp.Kvs {
+		kvs[string(kv.Key)] = kv
 	}
 
 	return kvs, nil

pkg/filter/wasmhost/doc.go

@@ -0,0 +1,18 @@
+/*
+ * 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 wasmhost