extend wasm abi

include/proxy-wasm/context.h

@@ -353,6 +353,22 @@ class ContextBase : public RootInterface,
                                std::string_view /* details */) override {
     return unimplemented();
   }
+
+  // Inject Data
+  WasmResult injectEncodedDataToFilterChain(std::string_view /* body_text */, bool /* end_stream */) override {
+    return unimplemented();
+  }
+
+  // Get all enpoint metrics of upstream cluster
+  WasmResult getUpstreamHosts(StringPairs * /* result */) override {
+    return unimplemented();
+  }
+
+  // Override upstream host and bypass lb policy
+  WasmResult setUpstreamOverrideHost(std::string_view /* address */) override {
+    return unimplemented();
+  }
+
   void clearRouteCache() override { unimplemented(); }
   void failStream(WasmStreamType stream_type) override { closeStream(stream_type); }
 

include/proxy-wasm/context_interface.h

@@ -31,6 +31,7 @@ namespace proxy_wasm {
 
 using Pairs = std::vector<std::pair<std::string_view, std::string_view>>;
 using PairsWithStringValues = std::vector<std::pair<std::string_view, std::string>>;
+using StringPairs = std::vector<std::pair<std::string, std::string>>;
 using TimerToken = uint32_t;
 using HttpCallToken = uint32_t;
 using GrpcToken = uint32_t;
@@ -255,6 +256,14 @@ struct HttpInterface {
 
   // Call just before the Context is deleted. See RootInterface.
   virtual void onDelete() = 0;
+
+  // Inject encoded data to filter chain
+  virtual WasmResult injectEncodedDataToFilterChain(std::string_view /* body_text */, bool /* end_stream */) = 0;
+
+  // Get all enpoint metrics of upstream cluster
+  virtual WasmResult getUpstreamHosts(StringPairs * /* result */) = 0;
+  // Override upstream host and bypass lb policy
+  virtual WasmResult setUpstreamOverrideHost(std::string_view /* address */) = 0;
 };
 
 /**

include/proxy-wasm/exports.h

@@ -78,6 +78,9 @@ Word send_local_response(Word response_code, Word response_code_details_ptr,
                          Word response_code_details_size, Word body_ptr, Word body_size,
                          Word additional_response_header_pairs_ptr,
                          Word additional_response_header_pairs_size, Word grpc_status);
+Word inject_encoded_data_to_filter_chain(Word body_ptr, Word body_size, Word end_stream);
+Word set_upstream_override_host(Word address_ptr, Word address_size);
+Word get_upstream_hosts(Word ptr, Word size);
 Word clear_route_cache();
 Word get_shared_data(Word key_ptr, Word key_size, Word value_ptr_ptr, Word value_size_ptr,
                      Word cas_ptr);
@@ -198,7 +201,9 @@ Word wasi_unstable_path_filestat_get(Word fd, Word flags, Word path, Word path_l
                                       _f(increment_metric) _f(record_metric) _f(get_metric)        \
                                           _f(set_effective_context) _f(done)                       \
                                               _f(call_foreign_function) _f(redis_init)             \
-                                                  _f(redis_call)
+                                                _f(redis_call) _f(get_upstream_hosts)              \
+                                                  _f(inject_encoded_data_to_filter_chain)          \
+                                                    _f(set_upstream_override_host)
 
 #define FOR_ALL_HOST_FUNCTIONS_ABI_SPECIFIC(_f)                                                    \
   _f(get_configuration) _f(continue_request) _f(continue_response) _f(clear_route_cache)           \

include/proxy-wasm/wasm_api_impl.h

@@ -102,6 +102,19 @@ proxy_send_local_response(uint32_t response_code, const char *response_code_deta
       WR(body_ptr), WS(body_size), WR(additional_response_header_pairs_ptr),
       WS(additional_response_header_pairs_size), WS(grpc_status)));
 }
+inline WasmResult
+proxy_inject_encoded_data_to_filter_chain(const char *body_ptr, size_t body_size, bool end_stream) {
+  return wordToWasmResult(exports::inject_encoded_data_to_filter_chain(
+      WR(body_ptr), WS(body_size), WS(end_stream)));
+}
+inline WasmResult
+proxy_set_upstream_override_host(const char *address_ptr, size_t address_size) {
+  return wordToWasmResult(exports::set_upstream_override_host(WR(address_ptr), WS(address_size)));
+}
+inline WasmResult
+proxy_get_upstream_hosts(const char **ptr, size_t *size) {
+  return wordToWasmResult(exports::get_upstream_hosts(WR(ptr), WR(size)));
+}
 
 inline WasmResult proxy_clear_route_cache() {
   return wordToWasmResult(exports::clear_route_cache());

src/exports.cc

@@ -159,6 +159,58 @@ Word send_local_response(Word response_code, Word response_code_details_ptr,
   return WasmResult::Ok;
 }
 
+Word inject_encoded_data_to_filter_chain(Word body_ptr, Word body_size, Word end_stream) {
+  auto *context = contextOrEffectiveContext();
+  auto body = context->wasmVm()->getMemory(body_ptr, body_size);
+  if (!body) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  context->injectEncodedDataToFilterChain(body.value(), end_stream != 0U);
+  return WasmResult::Ok;
+}
+
+Word get_upstream_hosts(Word ptr_ptr, Word size_ptr) {
+  auto *context = contextOrEffectiveContext();
+  StringPairs pairs;
+  auto result = context->getUpstreamHosts(&pairs);
+  if (result != WasmResult::Ok) {
+    return result;
+  }
+  if (pairs.empty()) {
+    if (!context->wasm()->copyToPointerSize("", ptr_ptr, size_ptr)) {
+      return WasmResult::InvalidMemoryAccess;
+    }
+    return WasmResult::Ok;
+  }
+  uint64_t size = PairsUtil::pairsSize(pairs);
+  uint64_t ptr = 0;
+  char *buffer = static_cast<char *>(context->wasm()->allocMemory(size, &ptr));
+  if (buffer == nullptr) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  if (!PairsUtil::marshalPairs(pairs, buffer, size)) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  if (!context->wasmVm()->setWord(ptr_ptr, Word(ptr))) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  if (!context->wasmVm()->setWord(size_ptr, Word(size))) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  return WasmResult::Ok;
+}
+
+Word set_upstream_override_host(Word address_ptr, Word address_size) {
+  auto *context = contextOrEffectiveContext();
+  auto address = context->wasmVm()->getMemory(address_ptr, address_size);
+  if (!address) {
+    return WasmResult::InvalidMemoryAccess;
+  }
+  context->setUpstreamOverrideHost(address.value());
+  return WasmResult::Ok;
+}
+
+
 Word clear_route_cache() {
   auto *context = contextOrEffectiveContext();
   context->clearRouteCache();

src/wasm.cc

@@ -153,6 +153,9 @@ void WasmBase::registerCallbacks() {
     _REGISTER_PROXY(get_log_level);
     _REGISTER_PROXY(redis_init);
     _REGISTER_PROXY(redis_call);
+    _REGISTER_PROXY(inject_encoded_data_to_filter_chain);
+    _REGISTER_PROXY(set_upstream_override_host);
+    _REGISTER_PROXY(get_upstream_hosts);
   }
 #undef _REGISTER_PROXY