Skip to content

Commit bbef2aa

Browse files
PiotrSikoraPiotrSikora
authored
wasm: make HTTP/gRPC callout IDs unique per WasmVM instance. (#17692)
Fixes proxy-wasm/proxy-wasm-cpp-host#185. Signed-off-by: Piotr Sikora <piotrsikora@google.com>
1 parent 39cc756 commit bbef2aa

File tree

5 files changed

+47
-96
lines changed

5 files changed

+47
-96
lines changed

‎bazel/repository_locations.bzl

+3-3
Original file line numberDiff line numberDiff line change
@@ -975,8 +975,8 @@ REPOSITORY_LOCATIONS_SPEC = dict(
975975
project_name = "WebAssembly for Proxies (C++ host implementation)",
976976
project_desc = "WebAssembly for Proxies (C++ host implementation)",
977977
project_url = "https://github.com/proxy-wasm/proxy-wasm-cpp-host",
978-
version = "d32cb05cb666216db5e1a383c1ffa2c1aeec19bb",
979-
sha256 = "c81b948fde4c0bebc3e1d32d3507052bc788153179e2229686aa28e817ee6913",
978+
version = "03185974ef574233a5f6383311eb74a380146fe2",
979+
sha256 = "34948e3ba239cc721af8d0a0a5b678325f363cbd542bddecf2267d24780d5b4d",
980980
strip_prefix = "proxy-wasm-cpp-host-{version}",
981981
urls = ["https://github.com/proxy-wasm/proxy-wasm-cpp-host/archive/{version}.tar.gz"],
982982
use_category = ["dataplane_ext"],
@@ -992,7 +992,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
992992
"envoy.wasm.runtime.wavm",
993993
"envoy.wasm.runtime.wasmtime",
994994
],
995-
release_date = "2021-08-02",
995+
release_date = "2021-08-12",
996996
cpe = "N/A",
997997
),
998998
proxy_wasm_rust_sdk = dict(

‎source/extensions/common/wasm/context.cc

+17-64
Original file line numberDiff line numberDiff line change
@@ -910,21 +910,6 @@ void Context::onUpstreamConnectionClose(CloseType close_type) {
910910
}
911911
}
912912

913-
uint32_t Context::nextHttpCallToken() {
914-
uint32_t token = next_http_call_token_++;
915-
// Handle rollover.
916-
for (;;) {
917-
if (token == 0) {
918-
token = next_http_call_token_++;
919-
}
920-
if (!http_request_.count(token)) {
921-
break;
922-
}
923-
token = next_http_call_token_++;
924-
}
925-
return token;
926-
}
927-
928913
// Async call via HTTP
929914
WasmResult Context::httpCall(std::string_view cluster, const Pairs& request_headers,
930915
std::string_view request_body, const Pairs& request_trailers,
@@ -961,7 +946,7 @@ WasmResult Context::httpCall(std::string_view cluster, const Pairs& request_head
961946
timeout = std::chrono::milliseconds(timeout_milliseconds);
962947
}
963948

964-
uint32_t token = nextHttpCallToken();
949+
uint32_t token = wasm()->nextHttpCallId();
965950
auto& handler = http_request_[token];
966951
handler.context_ = this;
967952
handler.token_ = token;
@@ -983,22 +968,6 @@ WasmResult Context::httpCall(std::string_view cluster, const Pairs& request_head
983968
return WasmResult::Ok;
984969
}
985970

986-
uint32_t Context::nextGrpcCallToken() {
987-
uint32_t token = next_grpc_token_++;
988-
if (isGrpcStreamToken(token)) {
989-
token = next_grpc_token_++;
990-
}
991-
// Handle rollover. Note: token is always odd.
992-
for (;;) {
993-
if (!grpc_call_request_.count(token)) {
994-
break;
995-
}
996-
next_grpc_token_++; // Skip stream token.
997-
token = next_grpc_token_++;
998-
}
999-
return token;
1000-
}
1001-
1002971
WasmResult Context::grpcCall(std::string_view grpc_service, std::string_view service_name,
1003972
std::string_view method_name, const Pairs& initial_metadata,
1004973
std::string_view request, std::chrono::milliseconds timeout,
@@ -1019,7 +988,7 @@ WasmResult Context::grpcCall(std::string_view grpc_service, std::string_view ser
1019988
return WasmResult::ParseFailure;
1020989
}
1021990
}
1022-
uint32_t token = nextGrpcCallToken();
991+
uint32_t token = wasm()->nextGrpcCallId();
1023992
auto& handler = grpc_call_request_[token];
1024993
handler.context_ = this;
1025994
handler.token_ = token;
@@ -1049,26 +1018,6 @@ WasmResult Context::grpcCall(std::string_view grpc_service, std::string_view ser
10491018
return WasmResult::Ok;
10501019
}
10511020

1052-
uint32_t Context::nextGrpcStreamToken() {
1053-
uint32_t token = next_grpc_token_++;
1054-
if (isGrpcCallToken(token)) {
1055-
token = next_grpc_token_++;
1056-
}
1057-
// Handle rollover. Note: token is always even.
1058-
for (;;) {
1059-
if (token == 0) {
1060-
next_grpc_token_++; // Skip call token.
1061-
token = next_grpc_token_++;
1062-
}
1063-
if (!grpc_stream_.count(token)) {
1064-
break;
1065-
}
1066-
next_grpc_token_++; // Skip call token.
1067-
token = next_grpc_token_++;
1068-
}
1069-
return token;
1070-
}
1071-
10721021
WasmResult Context::grpcStream(std::string_view grpc_service, std::string_view service_name,
10731022
std::string_view method_name, const Pairs& initial_metadata,
10741023
uint32_t* token_ptr) {
@@ -1088,7 +1037,7 @@ WasmResult Context::grpcStream(std::string_view grpc_service, std::string_view s
10881037
return WasmResult::ParseFailure;
10891038
}
10901039
}
1091-
uint32_t token = nextGrpcStreamToken();
1040+
uint32_t token = wasm()->nextGrpcStreamId();
10921041
auto& handler = grpc_stream_[token];
10931042
handler.context_ = this;
10941043
handler.token_ = token;
@@ -1879,7 +1828,7 @@ void Context::onGrpcReceiveWrapper(uint32_t token, ::Envoy::Buffer::InstancePtr
18791828
ContextBase::onGrpcReceive(token, response_size);
18801829
grpc_receive_buffer_.reset();
18811830
}
1882-
if (isGrpcCallToken(token)) {
1831+
if (wasm()->isGrpcCallId(token)) {
18831832
grpc_call_request_.erase(token);
18841833
}
18851834
}
@@ -1899,9 +1848,9 @@ void Context::onGrpcCloseWrapper(uint32_t token, const Grpc::Status::GrpcStatus&
18991848
onGrpcClose(token, status_code_);
19001849
status_message_ = "";
19011850
}
1902-
if (isGrpcCallToken(token)) {
1851+
if (wasm()->isGrpcCallId(token)) {
19031852
grpc_call_request_.erase(token);
1904-
} else {
1853+
} else if (wasm()->isGrpcStreamId(token)) {
19051854
auto it = grpc_stream_.find(token);
19061855
if (it != grpc_stream_.end()) {
19071856
if (it->second.local_closed_) {
@@ -1912,7 +1861,7 @@ void Context::onGrpcCloseWrapper(uint32_t token, const Grpc::Status::GrpcStatus&
19121861
}
19131862

19141863
WasmResult Context::grpcSend(uint32_t token, std::string_view message, bool end_stream) {
1915-
if (isGrpcCallToken(token)) {
1864+
if (!wasm()->isGrpcStreamId(token)) {
19161865
return WasmResult::BadArgument;
19171866
}
19181867
auto it = grpc_stream_.find(token);
@@ -1928,7 +1877,7 @@ WasmResult Context::grpcSend(uint32_t token, std::string_view message, bool end_
19281877
}
19291878

19301879
WasmResult Context::grpcClose(uint32_t token) {
1931-
if (isGrpcCallToken(token)) {
1880+
if (wasm()->isGrpcCallId(token)) {
19321881
auto it = grpc_call_request_.find(token);
19331882
if (it == grpc_call_request_.end()) {
19341883
return WasmResult::NotFound;
@@ -1937,7 +1886,8 @@ WasmResult Context::grpcClose(uint32_t token) {
19371886
it->second.request_->cancel();
19381887
}
19391888
grpc_call_request_.erase(token);
1940-
} else {
1889+
return WasmResult::Ok;
1890+
} else if (wasm()->isGrpcStreamId(token)) {
19411891
auto it = grpc_stream_.find(token);
19421892
if (it == grpc_stream_.end()) {
19431893
return WasmResult::NotFound;
@@ -1950,12 +1900,13 @@ WasmResult Context::grpcClose(uint32_t token) {
19501900
} else {
19511901
it->second.local_closed_ = true;
19521902
}
1903+
return WasmResult::Ok;
19531904
}
1954-
return WasmResult::Ok;
1905+
return WasmResult::BadArgument;
19551906
}
19561907

19571908
WasmResult Context::grpcCancel(uint32_t token) {
1958-
if (isGrpcCallToken(token)) {
1909+
if (wasm()->isGrpcCallId(token)) {
19591910
auto it = grpc_call_request_.find(token);
19601911
if (it == grpc_call_request_.end()) {
19611912
return WasmResult::NotFound;
@@ -1964,7 +1915,8 @@ WasmResult Context::grpcCancel(uint32_t token) {
19641915
it->second.request_->cancel();
19651916
}
19661917
grpc_call_request_.erase(token);
1967-
} else {
1918+
return WasmResult::Ok;
1919+
} else if (wasm()->isGrpcStreamId(token)) {
19681920
auto it = grpc_stream_.find(token);
19691921
if (it == grpc_stream_.end()) {
19701922
return WasmResult::NotFound;
@@ -1973,8 +1925,9 @@ WasmResult Context::grpcCancel(uint32_t token) {
19731925
it->second.stream_->resetStream();
19741926
}
19751927
grpc_stream_.erase(token);
1928+
return WasmResult::Ok;
19761929
}
1977-
return WasmResult::Ok;
1930+
return WasmResult::BadArgument;
19781931
}
19791932

19801933
} // namespace Wasm

‎source/extensions/common/wasm/context.h

-9
Original file line numberDiff line numberDiff line change
@@ -304,12 +304,6 @@ class Context : public proxy_wasm::ContextBase,
304304
return dynamic_cast<T*>(it->second.get());
305305
}
306306

307-
uint32_t nextGrpcCallToken();
308-
uint32_t nextGrpcStreamToken();
309-
uint32_t nextHttpCallToken();
310-
void setNextGrpcTokenForTesting(uint32_t token) { next_grpc_token_ = token; }
311-
void setNextHttpCallTokenForTesting(uint32_t token) { next_http_call_token_ = token; }
312-
313307
protected:
314308
friend class Wasm;
315309

@@ -392,9 +386,6 @@ class Context : public proxy_wasm::ContextBase,
392386
void onGrpcCloseWrapper(uint32_t token, const Grpc::Status::GrpcStatus& status,
393387
const std::string_view message);
394388

395-
bool isGrpcStreamToken(uint32_t token) { return (token & 1) == 0; }
396-
bool isGrpcCallToken(uint32_t token) { return (token & 1) == 1; }
397-
398389
Http::HeaderMap* getMap(WasmHeaderMapType type);
399390
const Http::HeaderMap* getConstMap(WasmHeaderMapType type);
400391

‎test/extensions/common/wasm/wasm_test.cc

+13-14
Original file line numberDiff line numberDiff line change
@@ -119,28 +119,27 @@ TEST_P(WasmCommonTest, WasmFailState) {
119119
wasm->wasm()->setFailStateForTesting(proxy_wasm::FailState::RuntimeError);
120120
EXPECT_EQ(toWasmEvent(wasm_base), WasmEvent::RuntimeError);
121121

122-
auto root_context = static_cast<Context*>(wasm->wasm()->createRootContext(plugin));
123-
uint32_t grpc_call_token1 = root_context->nextGrpcCallToken();
124-
uint32_t grpc_call_token2 = root_context->nextGrpcCallToken();
122+
uint32_t grpc_call_token1 = wasm->wasm()->nextGrpcCallId();
123+
EXPECT_TRUE(wasm->wasm()->isGrpcCallId(grpc_call_token1));
124+
uint32_t grpc_call_token2 = wasm->wasm()->nextGrpcCallId();
125+
EXPECT_TRUE(wasm->wasm()->isGrpcCallId(grpc_call_token2));
125126
EXPECT_NE(grpc_call_token1, grpc_call_token2);
126-
root_context->setNextGrpcTokenForTesting(0); // Rollover.
127-
EXPECT_EQ(root_context->nextGrpcCallToken(), 1);
128127

129-
uint32_t grpc_stream_token1 = root_context->nextGrpcStreamToken();
130-
uint32_t grpc_stream_token2 = root_context->nextGrpcStreamToken();
128+
uint32_t grpc_stream_token1 = wasm->wasm()->nextGrpcStreamId();
129+
EXPECT_TRUE(wasm->wasm()->isGrpcStreamId(grpc_stream_token1));
130+
uint32_t grpc_stream_token2 = wasm->wasm()->nextGrpcStreamId();
131+
EXPECT_TRUE(wasm->wasm()->isGrpcStreamId(grpc_stream_token2));
131132
EXPECT_NE(grpc_stream_token1, grpc_stream_token2);
132-
root_context->setNextGrpcTokenForTesting(0xFFFFFFFF); // Rollover.
133-
EXPECT_EQ(root_context->nextGrpcStreamToken(), 2);
134133

135-
uint32_t http_call_token1 = root_context->nextHttpCallToken();
136-
uint32_t http_call_token2 = root_context->nextHttpCallToken();
134+
uint32_t http_call_token1 = wasm->wasm()->nextHttpCallId();
135+
EXPECT_TRUE(wasm->wasm()->isHttpCallId(http_call_token1));
136+
uint32_t http_call_token2 = wasm->wasm()->nextHttpCallId();
137+
EXPECT_TRUE(wasm->wasm()->isHttpCallId(http_call_token2));
137138
EXPECT_NE(http_call_token1, http_call_token2);
138-
root_context->setNextHttpCallTokenForTesting(0); // Rollover.
139-
EXPECT_EQ(root_context->nextHttpCallToken(), 1);
140139

140+
auto root_context = static_cast<Context*>(wasm->wasm()->createRootContext(plugin));
141141
EXPECT_EQ(root_context->getBuffer(WasmBufferType::HttpCallResponseBody), nullptr);
142142
EXPECT_EQ(root_context->getBuffer(WasmBufferType::PluginConfiguration), nullptr);
143-
144143
delete root_context;
145144

146145
Filters::Common::Expr::CelStatePrototype wasm_state_prototype(

‎test/extensions/filters/http/wasm/wasm_filter_test.cc

+14-6
Original file line numberDiff line numberDiff line change
@@ -1124,12 +1124,20 @@ TEST_P(WasmHttpFilterTest, GrpcCallFailure) {
11241124
filter().decodeHeaders(request_headers, false));
11251125

11261126
// Test some additional error paths.
1127-
EXPECT_EQ(filter().grpcSend(99999, "", false), proxy_wasm::WasmResult::BadArgument);
1128-
EXPECT_EQ(filter().grpcSend(10000, "", false), proxy_wasm::WasmResult::NotFound);
1129-
EXPECT_EQ(filter().grpcCancel(9999), proxy_wasm::WasmResult::NotFound);
1130-
EXPECT_EQ(filter().grpcCancel(10000), proxy_wasm::WasmResult::NotFound);
1131-
EXPECT_EQ(filter().grpcClose(9999), proxy_wasm::WasmResult::NotFound);
1132-
EXPECT_EQ(filter().grpcClose(10000), proxy_wasm::WasmResult::NotFound);
1127+
// 0xFF00 (HTTP call).
1128+
EXPECT_EQ(filter().grpcSend(0xFF00, "", false), proxy_wasm::WasmResult::BadArgument);
1129+
EXPECT_EQ(filter().grpcCancel(0xFF00), proxy_wasm::WasmResult::BadArgument);
1130+
EXPECT_EQ(filter().grpcClose(0xFF00), proxy_wasm::WasmResult::BadArgument);
1131+
1132+
// 0xFF01 (gRPC call).
1133+
EXPECT_EQ(filter().grpcSend(0xFF01, "", false), proxy_wasm::WasmResult::BadArgument);
1134+
EXPECT_EQ(filter().grpcCancel(0xFF01), proxy_wasm::WasmResult::NotFound);
1135+
EXPECT_EQ(filter().grpcClose(0xFF01), proxy_wasm::WasmResult::NotFound);
1136+
1137+
// 0xFF02 (gRPC stream).
1138+
EXPECT_EQ(filter().grpcSend(0xFF02, "", false), proxy_wasm::WasmResult::NotFound);
1139+
EXPECT_EQ(filter().grpcCancel(0xFF02), proxy_wasm::WasmResult::NotFound);
1140+
EXPECT_EQ(filter().grpcClose(0xFF02), proxy_wasm::WasmResult::NotFound);
11331141

11341142
ProtobufWkt::Value value;
11351143
value.set_string_value("response");

0 commit comments

Comments
 (0)