Modify shared_queue example: use define multiple plugins in singleton. (tetratelabs#191)

Signed-off-by: Takeshi Yoneda <[email protected]>

Author: mathetake

e2e/e2e_test.go

@@ -243,6 +243,12 @@ func Test_shared_queue(t *testing.T) {
 	defer kill()
 	require.Eventually(t, func() bool {
 		res, err := http.Get("http://localhost:18000")
+		if err != nil || res.StatusCode != http.StatusOK {
+			return false
+		}
+		defer res.Body.Close()
+
+		res, err = http.Get("http://localhost:18001")
 		if err != nil {
 			return false
 		}
@@ -252,7 +258,9 @@ func Test_shared_queue(t *testing.T) {
 	require.Eventually(t, func() bool {
 		return checkMessage(stdErr.String(), []string{
 			`enqueued data: {"key": ":method","value": "GET"}`,
-			`dequeued data: {"key": ":method","value": "GET"}`,
+			`dequeued data from http_request_headers`,
+			`dequeued data from http_response_headers`,
+			`dequeued data from tcp_data_hashes`,
 		}, nil)
 	}, 5*time.Second, time.Millisecond, stdErr.String())
 }

examples/dispatch_call_on_tick/main.go

@@ -44,7 +44,7 @@ type pluginContext struct {
 }
 
 // Override types.DefaultPluginContext.
-func (ctx *pluginContext) OnPluginStart(vmConfigurationSize int) types.OnPluginStartStatus {
+func (ctx *pluginContext) OnPluginStart(pluginConfigurationSize int) types.OnPluginStartStatus {
 	if err := proxywasm.SetTickPeriodMilliSeconds(tickMilliseconds); err != nil {
 		proxywasm.LogCriticalf("failed to set tick period: %v", err)
 		return types.OnPluginStartStatusFailed

examples/foreign_call_on_tick/main.go

@@ -47,7 +47,7 @@ type pluginContext struct {
 }
 
 // Override types.DefaultPluginContext.
-func (ctx *pluginContext) OnPluginStart(vmConfigurationSize int) types.OnPluginStartStatus {
+func (ctx *pluginContext) OnPluginStart(pluginConfigurationSize int) types.OnPluginStartStatus {
 	if err := proxywasm.SetTickPeriodMilliSeconds(tickMilliseconds); err != nil {
 		proxywasm.LogCriticalf("failed to set tick period: %v", err)
 		return types.OnPluginStartStatusFailed

examples/helloworld/main.go

@@ -47,7 +47,7 @@ type helloWorld struct {
 }
 
 // Override types.DefaultPluginContext.
-func (ctx *helloWorld) OnPluginStart(vmConfigurationSize int) types.OnPluginStartStatus {
+func (ctx *helloWorld) OnPluginStart(pluginConfigurationSize int) types.OnPluginStartStatus {
 	rand.Seed(time.Now().UnixNano())
 
 	proxywasm.LogInfo("proxy_on_vm_start from Go!")

examples/shared_queue/README.md

@@ -7,28 +7,57 @@ There are two Wasm VMs are configured (See `envoy.yaml` for detail):
 1. The one with `vm_id="receiver"` and the binary of `receiver/main.go`.
 2. Another one with `vm_id="sender"` and the binary of `sender/main.go`.
 
-`receiver` VM runs as a singleton [Wasm Service](https://www.envoyproxy.io/docs/envoy/latest/configuration/other_features/wasm_service.html) which runs in the main thread, and registered a shared queue named `http_headers` by calling the `RegisterSharedQueue` host call.
+`receiver` VM runs as a singleton [Wasm Service](https://www.envoyproxy.io/docs/envoy/latest/configuration/other_features/wasm_service.html) which runs in the main thread, and there are **three** plugin configurations are given. These configuration values are `http_response_headers`, `http_request_headers` and `tcp_data_hashes`.
+Each of these corresponding PluginContext registers a shared queue whose name equals that configuration respectively.
 
-`sender` VM runs in the http filter chain on worker threads, and enqueue request headers to the shared queue resolved by the `ResolveSharedQueue` with the args of `vm_id=receiver` and `name=http_headers`.
+`sender` VM runs in a http filter and a network filter chain on worker threads, and 
+- enqueue request headers to the shared queue resolved by the `ResolveSharedQueue` with the args of (`vm_id=receiver`,`name=http_request_headers`) and (`vm_id=receiver`,`name=http_response_headers`).
+- enqueue hash values of tcp data frames to the shared queue resolved by the `ResolveSharedQueue` with the args of (`vm_id=receiver`,`name=tcp_data_hashes`).
 
 See [this talk](https://www.youtube.com/watch?v=XdWmm_mtVXI&t=1171s) for detail.
 
 
 ```bash
+wasm log receiver: queue "http_request_headers" registered as queueID=1 by contextID=1
+wasm log receiver: queue "http_request_headers" registered as queueID=1 by contextID=1
+wasm log receiver: queue "http_response_headers" registered as queueID=2 by contextID=2
+wasm log receiver: queue "tcp_data_hashes" registered as queueID=3 by contextID=3
+all clusters initialized. initializing init manager
+all dependencies initialized. starting workers
+wasm log sender: contextID=1 is configured for http
+wasm log sender: contextID=2 is configured for tcp
+wasm log sender: contextID=1 is configured for http
+wasm log sender: contextID=2 is configured for tcp
+
+....
+
+# curl localhost:18000
+
 wasm log sender: enqueued data: {"key": ":authority","value": "localhost:18000"}
-wasm log receiver: dequeued data: {"key": ":authority","value": "localhost:18000"}
 wasm log sender: enqueued data: {"key": ":path","value": "/"}
-wasm log receiver: dequeued data: {"key": ":path","value": "/"}
 wasm log sender: enqueued data: {"key": ":method","value": "GET"}
-wasm log receiver: dequeued data: {"key": ":method","value": "GET"}
 wasm log sender: enqueued data: {"key": ":scheme","value": "http"}
-wasm log receiver: dequeued data: {"key": ":scheme","value": "http"}
 wasm log sender: enqueued data: {"key": "user-agent","value": "curl/7.68.0"}
-wasm log receiver: dequeued data: {"key": "user-agent","value": "curl/7.68.0"}
 wasm log sender: enqueued data: {"key": "accept","value": "*/*"}
-wasm log receiver: dequeued data: {"key": "accept","value": "*/*"}
 wasm log sender: enqueued data: {"key": "x-forwarded-proto","value": "http"}
-wasm log receiver: dequeued data: {"key": "x-forwarded-proto","value": "http"}
-wasm log sender: enqueued data: {"key": "x-request-id","value": "73a13840-6ca2-4f9f-a639-91b014c8d485"}
-wasm log receiver: dequeued data: {"key": "x-request-id","value": "73a13840-6ca2-4f9f-a639-91b014c8d485"}
+wasm log sender: enqueued data: {"key": "x-request-id","value": "57d77551-02e8-455c-bf86-45a0d7308a0e"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": ":authority","value": "localhost:18000"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": ":path","value": "/"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": ":method","value": "GET"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": ":scheme","value": "http"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": "user-agent","value": "curl/7.68.0"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": "accept","value": "*/*"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": "x-forwarded-proto","value": "http"}
+wasm log receiver: (contextID=1) dequeued data from http_request_headers(queueID=1): {"key": "x-request-id","value": "57d77551-02e8-455c-bf86-45a0d7308a0e"}
+wasm log sender: (contextID=3) enqueued data: {"key": ":status","value": "200"}
+wasm log receiver: (contextID=2) dequeued data from http_response_headers(queueID=2): {"key": ":status","value": "200"}
+wasm log sender: (contextID=3) enqueued data: {"key": "content-length","value": "13"}
+wasm log receiver: (contextID=2) dequeued data from http_response_headers(queueID=2): {"key": "content-length","value": "13"}
+wasm log sender: (contextID=3) enqueued data: {"key": "content-type","value": "text/plain"}
+wasm log receiver: (contextID=2) dequeued data from http_response_headers(queueID=2): {"key": "content-type","value": "text/plain"}
+
+# curl localhost:18001
+
+wasm log sender: (contextID=4) enqueued data: 7d1a184bc958cdb9f1fee6591a3f2ae2
+wasm log receiver: (contextID=3) dequeued data from tcp_data_hashes(queueID=3): 7d1a184bc958cdb9f1fee6591a3f2ae2
 ```

examples/shared_queue/envoy.yaml

@@ -4,6 +4,45 @@ bootstrap_extensions:
     "@type": type.googleapis.com/envoy.extensions.wasm.v3.WasmService
     singleton: true
     config:
+      # Used as a queue name
+      configuration:
+        "@type": type.googleapis.com/google.protobuf.StringValue
+        value: "http_request_headers"
+      # Use the same vm_config as below, so we can reuse the same VM for multiple queues.
+      vm_config:
+        vm_id: "receiver"
+        runtime: "envoy.wasm.runtime.v8"
+        code:
+          local:
+            filename: "./examples/shared_queue/receiver/main.go.wasm"
+
+- name: envoy.bootstrap.wasm
+  typed_config:
+    "@type": type.googleapis.com/envoy.extensions.wasm.v3.WasmService
+    singleton: true
+    config:
+      # Used as a queue name
+      configuration:
+        "@type": type.googleapis.com/google.protobuf.StringValue
+        value: "http_response_headers"
+      # Use the same vm_config as above, so we can reuse the same VM for multiple queues.
+      vm_config:
+        vm_id: "receiver"
+        runtime: "envoy.wasm.runtime.v8"
+        code:
+          local:
+            filename: "./examples/shared_queue/receiver/main.go.wasm"
+
+- name: envoy.bootstrap.wasm
+  typed_config:
+    "@type": type.googleapis.com/envoy.extensions.wasm.v3.WasmService
+    singleton: true
+    config:
+      configuration:
+        # Used as a queue name
+        "@type": type.googleapis.com/google.protobuf.StringValue
+        value: "tcp_data_hashes"
+      # Use the same vm_config as above, so we can reuse the same VM for multiple queues.
       vm_config:
         vm_id: "receiver"
         runtime: "envoy.wasm.runtime.v8"
@@ -13,7 +52,7 @@ bootstrap_extensions:
 
 static_resources:
   listeners:
-    - name: main
+    - name: http
       address:
         socket_address:
           address: 0.0.0.0
@@ -43,13 +82,60 @@ static_resources:
                     typed_config:
                       "@type": type.googleapis.com/envoy.extensions.filters.http.wasm.v3.Wasm
                       config:
+                        # Instruct that the PluginContext should behave as a Http filter.
+                        configuration:
+                          "@type": type.googleapis.com/google.protobuf.StringValue
+                          value: "http"
+                        # Use the same vm_config as below, so we can reuse the same VM for multiple queues.
                         vm_config:
                           vm_id: "sender"
                           runtime: "envoy.wasm.runtime.v8"
                           code:
                             local:
                               filename: "./examples/shared_queue/sender/main.go.wasm"
                   - name: envoy.filters.http.router
+    - name: tcp
+      address:
+        socket_address:
+          address: 0.0.0.0
+          port_value: 18001
+      filter_chains:
+        - filters:
+            - name: envoy.filters.network.wasm
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.filters.network.wasm.v3.Wasm
+                config:
+                  # Instruct that the PluginContext should behave as a Tcp filter.
+                  configuration:
+                    "@type": type.googleapis.com/google.protobuf.StringValue
+                    value: "tcp"
+                  # Use the same vm_config as above, so we can reuse the same VM for multiple queues.
+                  vm_config:
+                    vm_id: "sender"
+                    runtime: "envoy.wasm.runtime.v8"
+                    code:
+                      local:
+                        filename: "./examples/shared_queue/sender/main.go.wasm"
+
+            - name: envoy.tcp_proxy
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+                stat_prefix: ingress
+                cluster: admin
+  clusters:
+    - name: admin
+      connect_timeout: 5000s
+      type: strict_dns
+      lb_policy: round_robin
+      load_assignment:
+        cluster_name: admin
+        endpoints:
+          - lb_endpoints:
+              - endpoint:
+                  address:
+                    socket_address:
+                      address: 0.0.0.0
+                      port_value: 8001
 
 admin:
   access_log_path: "/dev/null"

examples/shared_queue/receiver/main.go

@@ -15,12 +15,12 @@
 package main
 
 import (
+	"fmt"
+
 	"github.com/tetratelabs/proxy-wasm-go-sdk/proxywasm"
 	"github.com/tetratelabs/proxy-wasm-go-sdk/proxywasm/types"
 )
 
-const queueName = "http_headers"
-
 func main() {
 	proxywasm.SetVMContext(&vmContext{})
 }
@@ -32,23 +32,34 @@ type vmContext struct {
 }
 
 // Override types.DefaultVMContext.
-func (*vmContext) NewPluginContext(uint32) types.PluginContext {
-	return &receiverPluginContext{}
+func (*vmContext) NewPluginContext(contextID uint32) types.PluginContext {
+	return &receiverPluginContext{contextID: contextID}
 }
 
 type receiverPluginContext struct {
 	// Embed the default plugin context here,
 	// so that we don't need to reimplement all the methods.
+	contextID uint32
 	types.DefaultPluginContext
+	queueName string
 }
 
 // Override types.DefaultPluginContext.
-func (ctx *receiverPluginContext) OnPluginStart(vmConfigurationSize int) types.OnPluginStartStatus {
-	queueID, err := proxywasm.RegisterSharedQueue(queueName)
+func (ctx *receiverPluginContext) OnPluginStart(pluginConfigurationSize int) types.OnPluginStartStatus {
+	// Get Plugin configuration.
+	config, err := proxywasm.GetPluginConfiguration(pluginConfigurationSize)
+	if err != nil {
+		panic(fmt.Sprintf("failed to get plugin config: %v", err))
+	}
+
+	// Treat the config as the queue name for receiving.
+	ctx.queueName = string(config)
+
+	queueID, err := proxywasm.RegisterSharedQueue(ctx.queueName)
 	if err != nil {
 		panic("failed register queue")
 	}
-	proxywasm.LogInfof("queue \"%s\" registered as id=%d", queueName, queueID)
+	proxywasm.LogInfof("queue \"%s\" registered as queueID=%d by contextID=%d", ctx.queueName, queueID, ctx.contextID)
 	return types.OnPluginStartStatusOK
 }
 
@@ -59,7 +70,7 @@ func (ctx *receiverPluginContext) OnQueueReady(queueID uint32) {
 	case types.ErrorStatusEmpty:
 		return
 	case nil:
-		proxywasm.LogInfof("dequeued data: %s", string(data))
+		proxywasm.LogInfof("(contextID=%d) dequeued data from %s(queueID=%d): %s", ctx.contextID, ctx.queueName, queueID, string(data))
 	default:
 		proxywasm.LogCriticalf("error retrieving data from queue %d: %v", queueID, err)
 	}