Fix lifetime issue which was causing non-clonable (i.e. V8)

include/envoy/server/wasm.h

@@ -16,7 +16,7 @@ class Wasm {
   virtual ~Wasm() {}
 };
 
-typedef std::unique_ptr<Wasm> WasmPtr;
+typedef std::shared_ptr<Wasm> WasmSharedPtr;
 
 } // namespace Server
 } // namespace Envoy

include/envoy/server/wasm_config.h

@@ -5,6 +5,7 @@
 #include "envoy/event/dispatcher.h"
 #include "envoy/server/wasm.h"
 #include "envoy/thread_local/thread_local.h"
+#include "envoy/upstream/cluster_manager.h"
 
 #include "common/protobuf/protobuf.h"
 
@@ -16,6 +17,11 @@ class WasmFactoryContext {
 public:
   virtual ~WasmFactoryContext() {}
 
+  /**
+   * @return Upstream::ClusterManager& singleton for use by the entire server.
+   */
+  virtual Upstream::ClusterManager& clusterManager() PURE;
+
   /**
    * @return Event::Dispatcher& the main thread's dispatcher. This dispatcher should be used
    *         for all singleton processing.
@@ -48,12 +54,12 @@ class WasmFactory {
    * @param config const ProtoBuf::Message& supplies the config for the resource monitor
    *        implementation.
    * @param context WasmFactoryContext& supplies the resource monitor's context.
-   * @return WasmPtr a singleton Wasm servive. May be be nullptr if per silo.
+   * @return WasmSharedPtr a singleton Wasm service. May be be nullptr if per silo.
    * @throw EnvoyException if the implementation is unable to produce an instance with
    *        the provided parameters.
    */
-  virtual WasmPtr createWasm(const envoy::config::wasm::v2::WasmConfig& config,
-                             WasmFactoryContext& context) PURE;
+  virtual WasmSharedPtr createWasm(const envoy::config::wasm::v2::WasmConfig& config,
+                                   WasmFactoryContext& context) PURE;
 };
 
 } // namespace Configuration

source/extensions/access_loggers/wasm/config.cc

@@ -20,30 +20,30 @@ namespace AccessLoggers {
 namespace Wasm {
 
 AccessLog::InstanceSharedPtr
-WasmAccessLogFactory::createAccessLogInstance(const Protobuf::Message& config,
+WasmAccessLogFactory::createAccessLogInstance(const Protobuf::Message& proto_config,
                                               AccessLog::FilterPtr&& filter,
                                               Server::Configuration::FactoryContext& context) {
-  const auto& wal_config =
-      MessageUtil::downcastAndValidate<const envoy::config::accesslog::v2::WasmAccessLog&>(config);
+  const auto& config =
+      MessageUtil::downcastAndValidate<const envoy::config::accesslog::v2::WasmAccessLog&>(
+          proto_config);
+  auto id = config.id();
+  auto configuration = std::make_shared<std::string>(config.configuration());
   auto tls_slot = context.threadLocal().allocateSlot();
-  if (wal_config.has_vm_config()) {
-    auto base_wasm =
-        Common::Wasm::createWasm(wal_config.id(), wal_config.vm_config(), context.api());
-    tls_slot->set([&wal_config, &base_wasm, &context](Event::Dispatcher& dispatcher) {
-      auto result =
-          Common::Wasm::createThreadLocalWasm(*base_wasm, wal_config.vm_config(), dispatcher,
-                                              wal_config.configuration(), context.api());
-      result->setClusterManager(context.clusterManager());
+  if (config.has_vm_config()) {
+    // Create a base WASM to verify that the code loads before setting/cloning the for the individual threads.
+    auto base_wasm = Common::Wasm::createWasm(id, config.vm_config(), context.clusterManager(),
+                                              context.dispatcher(), context.api());
+    // NB: the Slot set() call doesn't complete inline, so all arguments must outlive this call.
+    tls_slot->set([base_wasm, configuration](Event::Dispatcher& dispatcher) {
+      auto result = Common::Wasm::createThreadLocalWasm(*base_wasm, *configuration, dispatcher);
       return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
     });
   } else {
-    if (wal_config.id().empty()) {
+    if (id.empty()) {
       throw Common::Wasm::WasmVmException("No WASM VM Id or vm_config specified");
     }
-    tls_slot->set([&wal_config, &context](Event::Dispatcher& dispatcher) {
-      auto result = Common::Wasm::getThreadLocalWasm(wal_config.id(), wal_config.configuration());
-      result->setDispatcher(dispatcher);
-      result->setClusterManager(context.clusterManager());
+    tls_slot->set([id, configuration](Event::Dispatcher&) {
+      auto result = Common::Wasm::getThreadLocalWasm(id, *configuration);
       return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
     });
   }

source/extensions/common/wasm/wasm.cc

@@ -640,10 +640,7 @@ uint32_t Context::httpCall(absl::string_view cluster, const Pairs& request_heade
   if (timeout_milliseconds < 0)
     return 0;
   auto cluster_string = std::string(cluster);
-  auto cluster_manager = clusterManager();
-  if (!cluster_manager)
-    return 0;
-  if (cluster_manager->get(cluster_string) == nullptr)
+  if (clusterManager().get(cluster_string) == nullptr)
     return 0;
 
   Http::MessagePtr message(new Http::RequestMessageImpl(buildHeaderMapFromPairs(request_headers)));
@@ -670,7 +667,8 @@ uint32_t Context::httpCall(absl::string_view cluster, const Pairs& request_heade
 
   auto token = next_async_token_++;
   auto& handler = http_request_[token];
-  auto http_request = cluster_manager->httpAsyncClientForCluster(cluster_string)
+  auto http_request = clusterManager()
+                          .httpAsyncClientForCluster(cluster_string)
                           .send(std::move(message), handler,
                                 Http::AsyncClient::RequestOptions().setTimeout(timeout));
   if (!http_request) {
@@ -932,7 +930,10 @@ void Context::onHttpCallResponse(uint32_t token, const Pairs& response_headers,
                              trailers_ptr, trailers_size);
 }
 
-Wasm::Wasm(absl::string_view vm, absl::string_view id) {
+Wasm::Wasm(absl::string_view vm, absl::string_view id, absl::string_view initial_configuration,
+           Upstream::ClusterManager& cluster_manager, Event::Dispatcher& dispatcher)
+    : cluster_manager_(cluster_manager), dispatcher_(dispatcher),
+      initial_configuration_(initial_configuration) {
   wasm_vm_ = Common::Wasm::createWasmVm(vm);
   id_ = std::string(id);
   if (wasm_vm_) {
@@ -1018,7 +1019,8 @@ void Wasm::getFunctions() {
 #undef _GET
 }
 
-Wasm::Wasm(const Wasm& wasm) {
+Wasm::Wasm(const Wasm& wasm)
+    : cluster_manager_(wasm.cluster_manager_), dispatcher_(wasm.dispatcher_) {
   wasm_vm_ = wasm.wasmVm()->clone();
   general_context_ = createContext();
   getFunctions();
@@ -1030,6 +1032,8 @@ bool Wasm::initialize(const std::string& code, absl::string_view name, bool allo
   auto ok = wasm_vm_->initialize(code, name, allow_precompiled);
   if (!ok)
     return false;
+  code_ = code;
+  allow_precompiled_ = allow_precompiled;
   getFunctions();
   return true;
 }
@@ -1046,8 +1050,12 @@ void Wasm::start() { general_context_->onStart(); }
 void Wasm::setTickPeriod(std::chrono::milliseconds tick_period) {
   bool was_running = timer_ && tick_period_.count() > 0;
   tick_period_ = tick_period;
-  if (dispatcher_ && tick_ && tick_period_.count() > 0 && !was_running) {
-    timer_ = dispatcher_->createTimer([this]() { this->tickHandler(); });
+  if (tick_ && tick_period_.count() > 0 && !was_running) {
+    timer_ = dispatcher_.createTimer([weak = std::weak_ptr<Wasm>(shared_from_this())]() {
+      auto shared = weak.lock();
+      if (shared)
+        shared->tickHandler();
+    });
     timer_->enableTimer(tick_period_);
   }
 }
@@ -1213,10 +1221,12 @@ std::unique_ptr<WasmVm> createWasmVm(absl::string_view wasm_vm) {
   }
 }
 
-std::unique_ptr<Wasm> createWasm(absl::string_view id,
+std::shared_ptr<Wasm> createWasm(absl::string_view id,
                                  const envoy::config::wasm::v2::VmConfig& vm_config,
-                                 Api::Api& api) {
-  auto wasm = std::make_unique<Wasm>(vm_config.vm(), id);
+                                 Upstream::ClusterManager& cluster_manager,
+                                 Event::Dispatcher& dispatcher, Api::Api& api) {
+  auto wasm = std::make_shared<Wasm>(vm_config.vm(), id, vm_config.initial_configuration(),
+                                     cluster_manager, dispatcher);
   const auto& code = Config::DataSource::read(vm_config.code(), true, api);
   const auto& path = Config::DataSource::getPath(vm_config.code())
                          .value_or(code.empty() ? EMPTY_STRING : INLINE_STRING);
@@ -1230,16 +1240,19 @@ std::unique_ptr<Wasm> createWasm(absl::string_view id,
   return wasm;
 }
 
-std::shared_ptr<Wasm> createThreadLocalWasm(Wasm& base_wasm,
-                                            const envoy::config::wasm::v2::VmConfig& vm_config,
-                                            Event::Dispatcher& dispatcher,
-                                            absl::string_view configuration, Api::Api& api) {
+std::shared_ptr<Wasm> createThreadLocalWasm(Wasm& base_wasm, absl::string_view configuration,
+                                            Event::Dispatcher& dispatcher) {
   std::shared_ptr<Wasm> wasm;
   if (base_wasm.wasmVm()->clonable()) {
     wasm = std::make_shared<Wasm>(base_wasm);
   } else {
-    wasm = createWasm(base_wasm.id(), vm_config, api);
-    wasm->setDispatcher(dispatcher);
+    wasm = std::make_shared<Wasm>(base_wasm.wasmVm()->vm(), base_wasm.id(),
+                                  base_wasm.initial_configuration(), base_wasm.clusterManager(),
+                                  dispatcher);
+    if (!wasm->initialize(base_wasm.code(), base_wasm.id(), base_wasm.allow_precompiled())) {
+      throw WasmException("Failed to initialize WASM code");
+    }
+    wasm->configure(base_wasm.initial_configuration());
   }
   wasm->configure(configuration);
   if (!wasm->id().empty())

source/extensions/common/wasm/wasm.h

@@ -53,7 +53,7 @@ class Context : public Http::StreamFilter,
 
   Wasm* wasm() const { return wasm_; }
   WasmVm* wasmVm() const;
-  Upstream::ClusterManager* clusterManager() const;
+  Upstream::ClusterManager& clusterManager() const;
   uint32_t id() const { return id_; }
   const StreamInfo::StreamInfo& streamInfo() const;
 
@@ -252,15 +252,11 @@ class Wasm : public Envoy::Server::Wasm,
              public Logger::Loggable<Logger::Id::wasm>,
              public std::enable_shared_from_this<Wasm> {
 public:
-  Wasm(absl::string_view vm, absl::string_view id);
+  Wasm(absl::string_view vm, absl::string_view id, absl::string_view initial_configuration,
+       Upstream::ClusterManager& cluster_manager, Event::Dispatcher& dispatcher);
   Wasm(const Wasm& other);
   ~Wasm() {}
 
-  void setDispatcher(Event::Dispatcher& dispatcher) { dispatcher_ = &dispatcher; }
-  void setClusterManager(Upstream::ClusterManager& clusterManager) {
-    clusterManager_ = &clusterManager;
-  }
-
   bool initialize(const std::string& code, absl::string_view name, bool allow_precompiled);
   void configure(absl::string_view configuration);
   void start();
@@ -271,7 +267,7 @@ class Wasm : public Envoy::Server::Wasm,
   absl::string_view id() const { return id_; }
   WasmVm* wasmVm() const { return wasm_vm_.get(); }
   Context* generalContext() const { return general_context_.get(); }
-  Upstream::ClusterManager* clusterManager() const { return clusterManager_; }
+  Upstream::ClusterManager& clusterManager() const { return cluster_manager_; }
 
   std::shared_ptr<Context> createContext() { return std::make_shared<Context>(this); }
 
@@ -280,6 +276,13 @@ class Wasm : public Envoy::Server::Wasm,
 
   uint32_t allocContextId();
 
+  const std::string& code() const { return code_; }
+  const std::string& initial_configuration() const { return initial_configuration_; }
+  bool allow_precompiled() const { return allow_precompiled_; }
+  void setInitialConfiguration(const std::string& initial_configuration) {
+    initial_configuration_ = initial_configuration;
+  }
+
   //
   // AccessLog::Instance
   //
@@ -296,8 +299,8 @@ class Wasm : public Envoy::Server::Wasm,
 
   void getFunctions();
 
-  Event::Dispatcher* dispatcher_ = nullptr;
-  Upstream::ClusterManager* clusterManager_ = nullptr;
+  Upstream::ClusterManager& cluster_manager_;
+  Event::Dispatcher& dispatcher_;
   std::string id_;
   std::string context_id_filter_state_data_name_;
   uint32_t next_context_id_ = 0;
@@ -330,10 +333,15 @@ class Wasm : public Envoy::Server::Wasm,
   WasmContextCall0Void onDone_;
   WasmContextCall0Void onLog_;
   WasmContextCall0Void onDelete_;
+
+  // Used by the base_wasm to enable non-clonable thread local Wasm(s) to be constructed.
+  std::string code_;
+  std::string initial_configuration_;
+  bool allow_precompiled_ = false;
 };
 
 inline WasmVm* Context::wasmVm() const { return wasm_->wasmVm(); }
-inline Upstream::ClusterManager* Context::clusterManager() const { return wasm_->clusterManager(); }
+inline Upstream::ClusterManager& Context::clusterManager() const { return wasm_->clusterManager(); }
 
 inline const ProtobufWkt::Struct& getMetadata(Http::StreamFilterCallbacks* callbacks) {
   if (callbacks->route() == nullptr || callbacks->route()->routeEntry() == nullptr) {
@@ -440,18 +448,20 @@ class WasmVm : public Logger::Loggable<Logger::Id::wasm> {
   }
 };
 
-// Create a new WASM VM of the give type (e.g. "envoy.wasm.vm.wavm").
+// Create a new low-level WASM VM of the give type (e.g. "envoy.wasm.vm.wavm").
 std::unique_ptr<WasmVm> createWasmVm(absl::string_view vm);
 
-// Create a new Wasm VM not attached to any thread. Note: 'id' may be empty if this VM will not be
-// shared by APIs (e.g. HTTP Filter + AccessLog).
-std::unique_ptr<Wasm> createWasm(absl::string_view id,
-                                 const envoy::config::wasm::v2::VmConfig& vm_config, Api::Api& api);
+// Create a high level Wasm VM with Envoy API support. Note: 'id' may be empty if this VM will not
+// be shared by APIs (e.g. HTTP Filter + AccessLog).
+std::shared_ptr<Wasm> createWasm(absl::string_view id,
+                                 const envoy::config::wasm::v2::VmConfig& vm_config,
+                                 Upstream::ClusterManager& cluster_manager,
+                                 Event::Dispatcher& dispatcher, Api::Api& api);
+
 // Create a ThreadLocal VM from an existing VM (e.g. from createWasm() above).
-std::shared_ptr<Wasm> createThreadLocalWasm(Wasm& base_wasm,
-                                            const envoy::config::wasm::v2::VmConfig& vm_config,
-                                            Event::Dispatcher& dispatcher,
-                                            absl::string_view configuration, Api::Api& api);
+std::shared_ptr<Wasm> createThreadLocalWasm(Wasm& base_wasm, absl::string_view configuration,
+                                            Event::Dispatcher& dispatcher);
+
 // Get an existing ThreadLocal VM matching 'id'.
 std::shared_ptr<Wasm> getThreadLocalWasm(absl::string_view id, absl::string_view configuration);
 

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

@@ -14,26 +14,27 @@ namespace Extensions {
 namespace HttpFilters {
 namespace Wasm {
 
-FilterConfig::FilterConfig(const envoy::config::filter::http::wasm::v2::Wasm& proto_config,
+FilterConfig::FilterConfig(const envoy::config::filter::http::wasm::v2::Wasm& config,
                            Server::Configuration::FactoryContext& context)
     : tls_slot_(context.threadLocal().allocateSlot()) {
-  if (proto_config.has_vm_config()) {
-    auto base_wasm = std::shared_ptr<Common::Wasm::Wasm>(
-        Common::Wasm::createWasm(proto_config.id(), proto_config.vm_config(), context.api())
-            .release());
-    tls_slot_->set([&proto_config, base_wasm, &context](Event::Dispatcher& dispatcher) {
-      auto result = Extensions::Common::Wasm::createThreadLocalWasm(
-          *base_wasm, proto_config.vm_config(), dispatcher, proto_config.configuration(),
-          context.api());
-      result->setClusterManager(context.clusterManager());
+  auto id = config.id();
+  auto configuration = std::make_shared<std::string>(config.configuration());
+  if (config.has_vm_config()) {
+    // Create a base WASM to verify that the code loads before setting/cloning the for the individual threads.
+    auto base_wasm = Common::Wasm::createWasm(id, config.vm_config(), context.clusterManager(),
+                                              context.dispatcher(), context.api());
+    // NB: the Slot set() call doesn't complete inline, so all arguments must outlive this call.
+    tls_slot_->set([base_wasm, configuration](Event::Dispatcher& dispatcher) {
+      auto result =
+          Extensions::Common::Wasm::createThreadLocalWasm(*base_wasm, *configuration, dispatcher);
       return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
     });
   } else {
-    tls_slot_->set([&proto_config, &context](Event::Dispatcher& dispatcher) {
-      auto result = Extensions::Common::Wasm::getThreadLocalWasm(proto_config.id(),
-                                                                 proto_config.configuration());
-      result->setDispatcher(dispatcher);
-      result->setClusterManager(context.clusterManager());
+    if (id.empty()) {
+      throw Common::Wasm::WasmVmException("No WASM VM Id or vm_config specified");
+    }
+    tls_slot_->set([id, configuration](Event::Dispatcher&) {
+      auto result = Extensions::Common::Wasm::getThreadLocalWasm(id, *configuration);
       return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
     });
   }

source/extensions/wasm/config.cc

@@ -16,22 +16,24 @@ namespace Wasm {
 
 static const std::string INLINE_STRING = "<inline>";
 
-Server::WasmPtr WasmFactory::createWasm(const envoy::config::wasm::v2::WasmConfig& config,
-                                        Server::Configuration::WasmFactoryContext& context) {
+Server::WasmSharedPtr WasmFactory::createWasm(const envoy::config::wasm::v2::WasmConfig& config,
+                                              Server::Configuration::WasmFactoryContext& context) {
+  // Create a base WASM to verify that the code loads before setting/cloning the for the individual threads.
+  auto base_wasm =
+      Common::Wasm::createWasm(config.id(), config.vm_config(), context.clusterManager(),
+                               context.dispatcher(), context.api());
   if (config.singleton()) {
-    auto wasm = Common::Wasm::createWasm(config.id(), config.vm_config(), context.api());
-    wasm->setDispatcher(context.dispatcher());
-    return Server::WasmPtr(wasm.release());
+    // Return the WASM VM which will be stored as a singleton by the Server.
+    return base_wasm;
   }
+  auto configuration = std::make_shared<std::string>(config.configuration());
   // Per-thread WASM VM.
-  auto base_wasm = Common::Wasm::createWasm(config.id(), config.vm_config(), context.api());
-  context.threadLocal().allocateSlot()->set(
-      [&config, &context, &base_wasm](Event::Dispatcher& dispatcher) {
-        // NB: no need to store the result as it is cached on each thread.
-        Extensions::Common::Wasm::createThreadLocalWasm(*base_wasm, config.vm_config(), dispatcher,
-                                                        config.configuration(), context.api());
-        return nullptr;
-      });
+  // NB: the Slot set() call doesn't complete inline, so all arguments must outlive this call.
+  // NB: no need to keep the resulting slot as Wasm is cached on each thread.
+  context.threadLocal().allocateSlot()->set([base_wasm,
+                                             configuration](Event::Dispatcher& dispatcher) {
+    return Extensions::Common::Wasm::createThreadLocalWasm(*base_wasm, *configuration, dispatcher);
+  });
   // Do not return this WASM VM since this is per-thread. Returning it would indicate that this is a
   // singleton.
   return nullptr;