gRPC wasm stress test (happy path)

source/extensions/filters/http/wasm/config.cc

@@ -28,9 +28,7 @@ Http::FilterFactoryCb WasmFilterConfig::createFilterFactoryFromProtoTyped(
 /**
  * Static registration for the Wasm filter. @see RegisterFactory.
  */
-static Registry::RegisterFactory<WasmFilterConfig,
-                                 Server::Configuration::NamedHttpFilterConfigFactory>
-    register_;
+REGISTER_FACTORY(WasmFilterConfig, Server::Configuration::NamedHttpFilterConfigFactory);
 
 } // namespace Wasm
 } // namespace HttpFilters

test/extensions/filters/http/wasm/BUILD

@@ -51,3 +51,21 @@ envoy_extension_cc_test(
         "//test/test_common:environment_lib",
     ],
 )
+
+envoy_extension_cc_test(
+    name = "wasm_filter_stress_test",
+    srcs = envoy_select_wasm(["wasm_filter_stress_test.cc"]),
+    data = [
+        "//test/extensions/filters/http/wasm/test_data:modules",
+    ],
+    extension_name = "envoy.filters.http.wasm",
+    external_deps = ["abseil_optional"],
+    deps = [
+        "//source/extensions/common/wasm:wasm_lib",
+        "//source/extensions/filters/http/wasm:config",
+        "//source/extensions/wasm:config",
+        "//test/proto:helloworld_proto_cc",
+        "//test/stress:stress_test_lib",
+        "//test/test_common:environment_lib",
+    ],
+)

test/extensions/filters/http/wasm/test_data/Makefile

@@ -1,3 +1,3 @@
-all: headers_cpp.wasm async_call_cpp.wasm metadata_cpp.wasm grpc_call_cpp.wasm shared_cpp.wasm queue_cpp.wasm
+all: headers_cpp.wasm async_call_cpp.wasm metadata_cpp.wasm grpc_call_cpp.wasm shared_cpp.wasm queue_cpp.wasm http_callout_cpp.wasm grpc_callout_cpp.wasm
 
 include ../../../../../../api/wasm/cpp/Makefile.base_lite

test/extensions/filters/http/wasm/test_data/grpc_callout_cpp.cc

@@ -0,0 +1,82 @@
+// NOLINT(namespace-envoy)
+#include <string>
+#include <unordered_map>
+#include "proxy_wasm_intrinsics.h"
+#include "proxy_wasm_intrinsics_lite.pb.h"
+
+class RequestContext : public Context {
+ public:
+  explicit RequestContext(uint32_t id, RootContext *root) : Context(id, root) {}
+
+  FilterHeadersStatus onRequestHeaders() override;
+};
+
+class ServiceContext : public RootContext {
+ public:
+  explicit ServiceContext(uint32_t id, StringView root_id) : RootContext(id, root_id) {}
+
+  void onStart() override {
+    callout_success_counter_ = defineMetric(MetricType::Counter, "test_callout_successes");
+    callout_failure_counter_ = defineMetric(MetricType::Counter, "test_callout_failures");
+  }
+
+  void incrementCalloutSuccesses(uint32_t inc_amount = 1U) {
+    incrementMetric(callout_success_counter_, inc_amount);
+  }
+
+  void incrementCalloutFailures(uint32_t inc_amount = 1U) {
+    incrementMetric(callout_failure_counter_, inc_amount);
+  }
+
+ private:
+  uint32_t callout_success_counter_;
+  uint32_t callout_failure_counter_;
+};
+
+static RegisterContextFactory
+    register_ExampleContext(CONTEXT_FACTORY(RequestContext), ROOT_FACTORY(ServiceContext));
+
+class CalloutResponseHandler : public GrpcCallHandler<google::protobuf::Value> {
+ public:
+  CalloutResponseHandler(RequestContext *context)
+      : GrpcCallHandler<google::protobuf::Value>(context), context_(context) {}
+  void onCreateInitialMetadata() override {}
+  void onSuccess(google::protobuf::Value &&response) override {
+    logDebug(response.string_value());
+
+    serviceContext()->incrementCalloutSuccesses();
+
+    continueRequest();
+  }
+  void onFailure(GrpcStatus status,
+                 std::unique_ptr<WasmData> error_message) override {
+    logInfo(std::string("failure ") + std::to_string(static_cast<int>(status)) +
+            std::string(error_message->view()));
+
+    serviceContext()->incrementCalloutFailures();
+
+    // TODO wasm engine must support fail closed: expose abortRequest() or similar
+    continueRequest();
+  }
+
+ private:
+  ServiceContext *serviceContext() {
+    return static_cast<ServiceContext *>(context_->root());
+  }
+
+  RequestContext *context_;
+};
+
+FilterHeadersStatus RequestContext::onRequestHeaders() {
+  GrpcService grpc_service;
+  grpc_service.mutable_envoy_grpc()->set_cluster_name("callout_cluster");
+  std::string grpc_service_string;
+  grpc_service.SerializeToString(&grpc_service_string);
+
+  google::protobuf::Value value;
+  value.set_string_value("request");
+
+  grpcCallHandler(grpc_service_string, "service", "method", value, 1000,
+                  std::unique_ptr<GrpcCallHandlerBase>(new CalloutResponseHandler(this)));
+  return FilterHeadersStatus::StopIteration;
+}

test/extensions/filters/http/wasm/test_data/http_callout_cpp.cc

@@ -0,0 +1,42 @@
+// NOLINT(namespace-envoy)
+#include <string>
+#include <unordered_map>
+
+#include "proxy_wasm_intrinsics.h"
+
+class ExampleContext : public Context {
+ public:
+  explicit ExampleContext(uint32_t id, RootContext *root) : Context(id, root) {}
+
+  FilterHeadersStatus onRequestHeaders() override;
+};
+static RegisterContextFactory
+    register_ExampleContext(CONTEXT_FACTORY(ExampleContext));
+
+FilterHeadersStatus ExampleContext::onRequestHeaders() {
+    WasmDataPtr data = getRequestHeader("x-callout-url");
+
+    if (!data) {
+      logWarn("Missing x-callout-url header, cannot forward");
+      return FilterHeadersStatus::Continue;
+    }
+
+    StringView callout_url{data->view()};
+
+    logInfo("Forwarding to: " + std::string(callout_url));
+
+    auto callback = [](std::unique_ptr<WasmData> response_headers,
+                       std::unique_ptr<WasmData> body,
+                       std::unique_ptr<WasmData> response_trailers) {
+      logInfo("Got response");
+    };
+    httpCall(callout_url,
+             {{":method", "POST"}, {":path", "/"}, {":authority", "foo"}},
+             "hello world",
+             {{"trail", "cow"}},
+             1000,
+             callback);
+
+  return FilterHeadersStatus::StopIteration;
+}
+