Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion include/envoy/server/wasm.h
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ class Wasm {
virtual ~Wasm() {}
};

typedef std::unique_ptr<Wasm> WasmPtr;
typedef std::shared_ptr<Wasm> WasmSharedPtr;

} // namespace Server
} // namespace Envoy
12 changes: 9 additions & 3 deletions include/envoy/server/wasm_config.h
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand All @@ -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.
Expand Down Expand Up @@ -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 servive. May be be nullptr if per silo.
Comment thread
jplevyak marked this conversation as resolved.
Outdated
* @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
Expand Down
31 changes: 15 additions & 16 deletions source/extensions/access_loggers/wasm/config.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,30 +20,29 @@ 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());
Comment thread
jplevyak marked this conversation as resolved.
auto tls_slot = context.threadLocal().allocateSlot();
Comment thread
jplevyak marked this conversation as resolved.
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()) {
auto base_wasm = Common::Wasm::createWasm(id, config.vm_config(), context.clusterManager(),
Comment thread
jplevyak marked this conversation as resolved.
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) {
Comment thread
jplevyak marked this conversation as resolved.
Comment thread
jplevyak marked this conversation as resolved.
auto result = Common::Wasm::createThreadLocalWasm(*base_wasm, *configuration, dispatcher);
Comment thread
jplevyak marked this conversation as resolved.
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&) {
Comment thread
jplevyak marked this conversation as resolved.
Comment thread
jplevyak marked this conversation as resolved.
auto result = Common::Wasm::getThreadLocalWasm(id, *configuration);
Comment thread
jplevyak marked this conversation as resolved.
return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
Comment thread
jplevyak marked this conversation as resolved.
});
}
Expand Down
49 changes: 31 additions & 18 deletions source/extensions/common/wasm/wasm.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment thread
jplevyak marked this conversation as resolved.
return 0;

Http::MessagePtr message(new Http::RequestMessageImpl(buildHeaderMapFromPairs(request_headers)));
Expand All @@ -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) {
Expand Down Expand Up @@ -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_) {
Expand Down Expand Up @@ -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();
Expand All @@ -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;
}
Expand All @@ -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_);
}
}
Expand Down Expand Up @@ -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);
Expand All @@ -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())
Expand Down
50 changes: 30 additions & 20 deletions source/extensions/common/wasm/wasm.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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();
Expand All @@ -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_; }
Comment thread
jplevyak marked this conversation as resolved.

std::shared_ptr<Context> createContext() { return std::make_shared<Context>(this); }

Expand All @@ -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
//
Expand All @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);

Expand Down
30 changes: 15 additions & 15 deletions source/extensions/filters/http/wasm/wasm_filter.cc
Original file line number Diff line number Diff line change
Expand Up @@ -14,26 +14,26 @@ 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()) {
auto base_wasm = Common::Wasm::createWasm(id, config.vm_config(), context.clusterManager(),
Comment thread
jplevyak marked this conversation as resolved.
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) {
Comment thread
jplevyak marked this conversation as resolved.
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&) {
Comment thread
jplevyak marked this conversation as resolved.
auto result = Extensions::Common::Wasm::getThreadLocalWasm(id, *configuration);
return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(result);
});
}
Expand Down
27 changes: 14 additions & 13 deletions source/extensions/wasm/config.cc
Original file line number Diff line number Diff line change
Expand Up @@ -16,22 +16,23 @@ 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) {
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());
Comment thread
jplevyak marked this conversation as resolved.
// 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) {
Comment thread
jplevyak marked this conversation as resolved.
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;
Expand Down
4 changes: 2 additions & 2 deletions source/extensions/wasm/config.h
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ class WasmFactory : public Server::Configuration::WasmFactory {
public:
~WasmFactory() override {}
std::string name() override { return "envoy.wasm"; }
Server::WasmPtr createWasm(const envoy::config::wasm::v2::WasmConfig& config,
Server::Configuration::WasmFactoryContext& context) override;
Server::WasmSharedPtr createWasm(const envoy::config::wasm::v2::WasmConfig& config,
Server::Configuration::WasmFactoryContext& context) override;
};

} // namespace Wasm
Expand Down
Loading