support redis call

Author: johnlanni

.bazelrc

@@ -1,3 +1,4 @@
+build:compdb --build_tag_filters=-nocompdb
 # Pass CC, CXX and PATH from the environment.
 build --action_env=CC
 build --action_env=CXX

WORKSPACE

@@ -8,14 +8,14 @@ load("@proxy_wasm_cpp_host//bazel:dependencies.bzl", "proxy_wasm_cpp_host_depend
 
 proxy_wasm_cpp_host_dependencies()
 
-load("@proxy_wasm_cpp_sdk//bazel:repositories.bzl", "proxy_wasm_cpp_sdk_repositories")
+load("@proxy_wasm_cpp_sdk//bazel:repositories.bzl", "proxy_wasm_cpp_host_repositories")
 
-proxy_wasm_cpp_sdk_repositories()
+proxy_wasm_cpp_host_repositories()
 
-load("@proxy_wasm_cpp_sdk//bazel:dependencies.bzl", "proxy_wasm_cpp_sdk_dependencies")
+load("@proxy_wasm_cpp_sdk//bazel:dependencies.bzl", "proxy_wasm_cpp_host_dependencies")
 
-proxy_wasm_cpp_sdk_dependencies()
+proxy_wasm_cpp_host_dependencies()
 
-load("@proxy_wasm_cpp_sdk//bazel:dependencies_extra.bzl", "proxy_wasm_cpp_sdk_dependencies_extra")
+load("@proxy_wasm_cpp_sdk//bazel:dependencies_extra.bzl", "proxy_wasm_cpp_host_dependencies_extra")
 
-proxy_wasm_cpp_sdk_dependencies_extra()
+proxy_wasm_cpp_host_dependencies_extra()

bazel/repositories.bzl

@@ -19,6 +19,13 @@ load("@bazel_tools//tools/build_defs/repo:utils.bzl", "maybe")
 def proxy_wasm_cpp_host_repositories():
     # Bazel extensions.
 
+    maybe(
+        http_archive,
+        name = "bazel_compdb",
+        strip_prefix = "bazel-compilation-database-0.5.2",
+        url = "https://github.com/grailbio/bazel-compilation-database/archive/0.5.2.tar.gz",
+    )
+
     maybe(
         http_archive,
         name = "bazel_skylib",
@@ -94,11 +101,10 @@ def proxy_wasm_cpp_host_repositories():
     )
 
     maybe(
-        http_archive,
+        git_repository,
         name = "proxy_wasm_cpp_sdk",
-        sha256 = "89792fc1abca331f29f99870476a04146de5e82ff903bdffca90e6729c1f2470",
-        strip_prefix = "proxy-wasm-cpp-sdk-95bb82ce45c41d9100fd1ec15d2ffc67f7f3ceee",
-        urls = ["https://github.com/proxy-wasm/proxy-wasm-cpp-sdk/archive/95bb82ce45c41d9100fd1ec15d2ffc67f7f3ceee.tar.gz"],
+        commit = "0a2b9e1928878e1662ad671be4ee9a8d0853243a",
+        remote = "[email protected]:Ingress/proxy-wasm-cpp-sdk.git",
     )
 
     # Test dependencies.

include/proxy-wasm/context.h

@@ -136,6 +136,7 @@ class ContextBase : public RootInterface,
                     public StreamInterface,
                     public HeaderInterface,
                     public HttpCallInterface,
+                    public RedisCallInterface,
                     public GrpcCallInterface,
                     public GrpcStreamInterface,
                     public MetricsInterface,
@@ -151,6 +152,7 @@ class ContextBase : public RootInterface,
   virtual ~ContextBase();
 
   WasmBase *wasm() const { return wasm_; }
+  void clearWasm() { wasm_ = nullptr; }
   uint32_t id() const { return id_; }
   // The VM Context used for calling "malloc" has an id_ == 0.
   bool isVmContext() const { return id_ == 0; }
@@ -214,6 +216,10 @@ class ContextBase : public RootInterface,
   // Async call response.
   void onHttpCallResponse(HttpCallToken token, uint32_t headers, uint32_t body_size,
                           uint32_t trailers) override;
+
+  // Redis
+  void onRedisCallResponse(RedisCallToken token, uint32_t status, uint32_t response_size) override;
+
   // Grpc
   void onGrpcReceiveInitialMetadata(GrpcToken token, uint32_t elements) override;
   void onGrpcReceive(GrpcToken token, uint32_t response_size) override;
@@ -276,6 +282,16 @@ class ContextBase : public RootInterface,
     return unimplemented();
   }
 
+  // Redis
+  WasmResult redisInit(std::string_view /* target */, std::string_view /* username */,
+                       std::string_view /* password */, int /* timeout_millisconds */) override {
+    return unimplemented();
+  }
+  WasmResult redisCall(std::string_view /* target */, std::string_view /* query */,
+                       uint32_t * /* token_ptr */) override {
+    return unimplemented();
+  }
+
   // gRPC
   WasmResult grpcCall(std::string_view /* grpc_service */, std::string_view /* service_name */,
                       std::string_view /* method_name */, const Pairs & /* initial_metadata */,

include/proxy-wasm/context_interface.h

@@ -37,6 +37,7 @@ using GrpcToken = uint32_t;
 using GrpcStatusCode = uint32_t;
 using SharedQueueDequeueToken = uint32_t;
 using SharedQueueEnqueueToken = uint32_t;
+using RedisCallToken = uint32_t;
 
 // TODO: update SDK and use this.
 enum class ProxyAction : uint32_t {
@@ -161,6 +162,12 @@ struct RootInterface : public RootGrpcInterface {
    */
   virtual void onHttpCallResponse(HttpCallToken token, uint32_t headers, uint32_t body_size,
                                   uint32_t trailers) = 0;
+  /**
+   * Called on a Root Context when a response arrives for an outstanding redisCall().
+   * @param token is the token returned by the corresponding redisCall().
+   */
+  virtual void onRedisCallResponse(RedisCallToken token, uint32_t status,
+                                   uint32_t response_size) = 0;
 
   /**
    * Called on a Root Context when an Inter-VM shared queue message has arrived.
@@ -416,6 +423,14 @@ struct HttpCallInterface {
                               int timeout_milliseconds, HttpCallToken *token_ptr) = 0;
 };
 
+struct RedisCallInterface {
+  virtual ~RedisCallInterface() = default;
+  virtual WasmResult redisInit(std::string_view target, std::string_view username,
+                               std::string_view password, int timeout_milliseconds) = 0;
+  virtual WasmResult redisCall(std::string_view target, std::string_view query,
+                               RedisCallToken *token_ptr) = 0;
+};
+
 struct GrpcCallInterface {
   virtual ~GrpcCallInterface() = default;
   /**

include/proxy-wasm/exports.h

@@ -104,6 +104,10 @@ Word get_response_body_buffer_bytes(Word start, Word length, Word ptr_ptr, Word
 Word http_call(Word uri_ptr, Word uri_size, Word header_pairs_ptr, Word header_pairs_size,
                Word body_ptr, Word body_size, Word trailer_pairs_ptr, Word trailer_pairs_size,
                Word timeout_milliseconds, Word token_ptr);
+Word redis_init(Word service_ptr, Word service_size, Word username_ptr, Word username_size,
+                Word passowrd_ptr, Word password_size, Word timeout_milliseconds);
+Word redis_call(Word service_ptr, Word service_size, Word query_ptr, Word query_size,
+                Word token_ptr);
 Word define_metric(Word metric_type, Word name_ptr, Word name_size, Word metric_id_ptr);
 Word increment_metric(Word metric_id, int64_t offset);
 Word record_metric(Word metric_id, uint64_t value);
@@ -163,7 +167,8 @@ void emscripten_notify_memory_growth(Word);
                                   _f(get_current_time_nanoseconds) _f(define_metric)               \
                                       _f(increment_metric) _f(record_metric) _f(get_metric)        \
                                           _f(set_effective_context) _f(done)                       \
-                                              _f(call_foreign_function)
+                                              _f(call_foreign_function) _f(redis_init)             \
+                                                  _f(redis_call)
 
 #define FOR_ALL_HOST_FUNCTIONS_ABI_SPECIFIC(_f)                                                    \
   _f(get_configuration) _f(continue_request) _f(continue_response) _f(clear_route_cache)           \

include/proxy-wasm/null_plugin.h

@@ -92,6 +92,8 @@ class NullPlugin : public NullVmPlugin {
   void onHttpCallResponse(uint64_t context_id, uint64_t token, uint64_t headers, uint64_t body_size,
                           uint64_t trailers);
 
+  void onRedisCallResponse(uint64_t context_id, uint64_t token, uint64_t status, uint64_t response);
+
   void onGrpcReceive(uint64_t context_id, uint64_t token, size_t body_size);
   void onGrpcClose(uint64_t context_id, uint64_t token, uint64_t status_code);
   void onGrpcReceiveInitialMetadata(uint64_t context_id, uint64_t token, uint64_t headers);

include/proxy-wasm/wasm.h

@@ -17,6 +17,7 @@
 
 #include <string.h>
 
+#include <cassert>
 #include <atomic>
 #include <deque>
 #include <map>
@@ -72,6 +73,11 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
       return it->second;
     return nullptr;
   }
+  void clearWasmInContext() {
+    for (auto &item : contexts_) {
+      item.second->clearWasm();
+    }
+  }
   uint32_t allocContextId();
   bool isFailed() { return failed_ != FailState::Ok; }
   FailState fail_state() { return failed_; }
@@ -174,6 +180,7 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
     HttpCall = 0,
     GrpcCall = 1,
     GrpcStream = 2,
+    RedisCall = 3,
   };
   static const uint32_t kCalloutTypeMask = 0x3;  // Enough to cover the 3 types.
   static const uint32_t kCalloutIncrement = 0x4; // Enough to cover the 3 types.
@@ -186,6 +193,9 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
   bool isGrpcStreamId(uint32_t callout_id) {
     return (callout_id & kCalloutTypeMask) == static_cast<uint32_t>(CalloutType::GrpcStream);
   }
+  bool isRedisCallId(uint32_t callout_id) {
+    return (callout_id & kCalloutTypeMask) == static_cast<uint32_t>(CalloutType::RedisCall);
+  }
   uint32_t nextHttpCallId() {
     // TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
     return next_http_call_id_ += kCalloutIncrement;
@@ -198,6 +208,10 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
     // TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
     return next_grpc_stream_id_ += kCalloutIncrement;
   }
+  uint32_t nextRedisCallId() {
+    // TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
+    return next_redis_call_id_ += kCalloutIncrement;
+  }
 
 protected:
   friend class ContextBase;
@@ -257,6 +271,8 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
 
   WasmCallVoid<5> on_http_call_response_;
 
+  WasmCallVoid<4> on_redis_call_response_;
+
   WasmCallVoid<3> on_grpc_receive_;
   WasmCallVoid<3> on_grpc_close_;
   WasmCallVoid<3> on_grpc_create_initial_metadata_;
@@ -276,9 +292,10 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
           _f(on_downstream_connection_close) _f(on_upstream_connection_close) _f(on_request_body)  \
               _f(on_request_trailers) _f(on_request_metadata) _f(on_response_body)                 \
                   _f(on_response_trailers) _f(on_response_metadata) _f(on_http_call_response)      \
-                      _f(on_grpc_receive) _f(on_grpc_close) _f(on_grpc_receive_initial_metadata)   \
-                          _f(on_grpc_receive_trailing_metadata) _f(on_queue_ready) _f(on_done)     \
-                              _f(on_log) _f(on_delete)
+                      _f(on_redis_call_response) _f(on_grpc_receive) _f(on_grpc_close)             \
+                          _f(on_grpc_receive_initial_metadata)                                     \
+                              _f(on_grpc_receive_trailing_metadata) _f(on_queue_ready) _f(on_done) \
+                                  _f(on_log) _f(on_delete)
 
   // Capabilities which are allowed to be linked to the module. If this is empty, restriction
   // is not enforced.
@@ -307,6 +324,7 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
   uint32_t next_http_call_id_ = static_cast<uint32_t>(CalloutType::HttpCall);
   uint32_t next_grpc_call_id_ = static_cast<uint32_t>(CalloutType::GrpcCall);
   uint32_t next_grpc_stream_id_ = static_cast<uint32_t>(CalloutType::GrpcStream);
+  uint32_t next_redis_call_id_ = static_cast<uint32_t>(CalloutType::RedisCall);
 
   // Actions to be done after the call into the VM returns.
   std::deque<std::function<void()>> after_vm_call_actions_;
@@ -335,8 +353,26 @@ class WasmHandleBase : public std::enable_shared_from_this<WasmHandleBase> {
 
   std::shared_ptr<WasmBase> &wasm() { return wasm_base_; }
 
+  void setRecoverVmCallback(std::function<std::shared_ptr<WasmHandleBase>()> &&f) {
+    recover_vm_callback_ = std::move(f);
+  }
+
+  // Recover the wasm vm and generate a new wasm handle
+  bool doRecover(std::shared_ptr<WasmHandleBase> &new_handle) {
+    assert(new_handle == nullptr);
+    if (recover_vm_callback_ == nullptr) {
+      return true;
+    }
+    new_handle = recover_vm_callback_();
+    if (!new_handle) {
+      return false;
+    }
+    return true;
+  }
+
 protected:
   std::shared_ptr<WasmBase> wasm_base_;
+  std::function<std::shared_ptr<WasmHandleBase>()> recover_vm_callback_;
   std::unordered_map<std::string, bool> plugin_canary_cache_;
 };
 
@@ -365,10 +401,35 @@ class PluginHandleBase : public std::enable_shared_from_this<PluginHandleBase> {
 
   std::shared_ptr<PluginBase> &plugin() { return plugin_; }
   std::shared_ptr<WasmBase> &wasm() { return wasm_handle_->wasm(); }
+  std::shared_ptr<WasmHandleBase> &wasmHandle() { return wasm_handle_; }
+
+  void setRecoverPluginCallback(
+      std::function<std::shared_ptr<PluginHandleBase>(std::shared_ptr<WasmHandleBase> &)> &&f) {
+    recover_plugin_callback_ = std::move(f);
+  }
+
+  // Recover the wasm plugin and generate a new plugin handle
+  bool doRecover(std::shared_ptr<PluginHandleBase> &new_handle) {
+    assert(new_handle == nullptr);
+    if (recover_plugin_callback_ == nullptr) {
+      return true;
+    }
+    std::shared_ptr<WasmHandleBase> new_wasm_handle;
+    if (!wasm_handle_->doRecover(new_wasm_handle)) {
+      return false;
+    }
+    new_handle = recover_plugin_callback_(new_wasm_handle);
+    if (!new_handle) {
+      return false;
+    }
+    return true;
+  }
 
 protected:
   std::shared_ptr<PluginBase> plugin_;
   std::shared_ptr<WasmHandleBase> wasm_handle_;
+  std::function<std::shared_ptr<PluginHandleBase>(std::shared_ptr<WasmHandleBase> &)>
+      recover_plugin_callback_;
 };
 
 using PluginHandleFactory = std::function<std::shared_ptr<PluginHandleBase>(

include/proxy-wasm/wasm_api_impl.h

@@ -211,6 +211,23 @@ inline WasmResult proxy_http_call(const char *uri_ptr, size_t uri_size, void *he
                                              WR(trailer_pairs_ptr), WS(trailer_pairs_size),
                                              WS(timeout_milliseconds), WR(token_ptr)));
 }
+
+// Redis
+// Returns token, used in callback onRedisCallResponse
+inline WasmResult proxy_redis_call(const char *service_ptr, size_t service_size,
+                                   const char *query_ptr, size_t query_size, uint32_t *token_ptr) {
+  return wordToWasmResult(exports::redis_call(WR(service_ptr), WS(service_size), WR(query_ptr),
+                                              WS(query_size), WR(token_ptr)));
+}
+inline WasmResult proxy_redis_init(const char *service_ptr, size_t service_size,
+                                   const char *username_ptr, size_t username_size,
+                                   const char *password_ptr, size_t password_size,
+                                   uint32_t timeout_milliseconds) {
+  return wordToWasmResult(exports::redis_init(WR(service_ptr), WS(service_size), WR(username_ptr),
+                                              WS(username_size), WR(password_ptr),
+                                              WS(password_size), WS(timeout_milliseconds)));
+}
+
 // gRPC
 // Returns token, used in gRPC callbacks (onGrpc...)
 inline WasmResult proxy_grpc_call(const char *service_ptr, size_t service_size,

include/proxy-wasm/wasm_vm.h

@@ -70,9 +70,9 @@ using WasmCallWord = std::function<WasmCallInFuncType<N, Word, ContextBase *, Wo
 
 #define FOR_ALL_WASM_VM_EXPORTS(_f)                                                                \
   _f(proxy_wasm::WasmCallVoid<0>) _f(proxy_wasm::WasmCallVoid<1>) _f(proxy_wasm::WasmCallVoid<2>)  \
-      _f(proxy_wasm::WasmCallVoid<3>) _f(proxy_wasm::WasmCallVoid<5>)                              \
-          _f(proxy_wasm::WasmCallWord<1>) _f(proxy_wasm::WasmCallWord<2>)                          \
-              _f(proxy_wasm::WasmCallWord<3>)
+      _f(proxy_wasm::WasmCallVoid<3>) _f(proxy_wasm::WasmCallVoid<4>)                              \
+          _f(proxy_wasm::WasmCallVoid<5>) _f(proxy_wasm::WasmCallWord<1>)                          \
+              _f(proxy_wasm::WasmCallWord<2>) _f(proxy_wasm::WasmCallWord<3>)
 
 // These are templates and its helper for constructing signatures of functions callbacks from Wasm
 // VMs.
@@ -139,7 +139,13 @@ enum class Cloneable {
   InstantiatedModule // VMs can be cloned from an instantiated module.
 };
 
-enum class AbiVersion { ProxyWasm_0_1_0, ProxyWasm_0_2_0, ProxyWasm_0_2_1, ProxyWasm_0_2_100, Unknown };
+enum class AbiVersion {
+  ProxyWasm_0_1_0,
+  ProxyWasm_0_2_0,
+  ProxyWasm_0_2_1,
+  ProxyWasm_0_2_100,
+  Unknown
+};
 
 class NullPlugin;
 
@@ -174,6 +180,7 @@ enum class FailState : int {
   StartFailed = 5,
   ConfigureFailed = 6,
   RuntimeError = 7,
+  RecoverError = 8,
 };
 
 // Wasm VM instance. Provides the low level WASM interface.