-
Notifications
You must be signed in to change notification settings - Fork 5.5k
lua: add fire-and-forget functionality to http call #10145
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 31 commits
871ee1b
df9df09
541bc4b
8b5b84d
323d728
e892eaa
29464ba
b566eab
4cb0ecc
27a0f3c
11c1301
21aa51e
bf9998d
3e0f6c4
38a8a99
67aee6e
e5bee45
ac562dd
aa04eb8
d838492
7b234f3
f4cd746
d6e452b
d022924
5ad50d1
55d1682
1ebaac1
f279ad2
1594207
9945ac5
1d4ba85
6dcdcf5
8459416
d190e36
9a1dbd7
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,6 +15,79 @@ namespace Extensions { | |
| namespace HttpFilters { | ||
| namespace Lua { | ||
|
|
||
| namespace { | ||
| // Okay to return non-const reference because this doesn't ever get changed. | ||
| NoopCallbacks& noopCallbacks() { | ||
| static NoopCallbacks* callbacks = new NoopCallbacks(); | ||
| return *callbacks; | ||
| } | ||
|
|
||
| void buildHeadersFromTable(Http::HeaderMap& headers, lua_State* state, int table_index) { | ||
| // Build a header map to make the request. We iterate through the provided table to do this and | ||
| // check that we are getting strings. | ||
| lua_pushnil(state); | ||
| while (lua_next(state, table_index) != 0) { | ||
| // Uses 'key' (at index -2) and 'value' (at index -1). | ||
| const char* key = luaL_checkstring(state, -2); | ||
| // Check if the current value is a table, we iterate through the table and add each element of | ||
| // it as a header entry value for the current key. | ||
| if (lua_istable(state, -1)) { | ||
| lua_pushnil(state); | ||
| while (lua_next(state, -2) != 0) { | ||
| const char* value = luaL_checkstring(state, -1); | ||
| headers.addCopy(Http::LowerCaseString(key), value); | ||
| lua_pop(state, 1); | ||
| } | ||
| } else { | ||
| const char* value = luaL_checkstring(state, -1); | ||
| headers.addCopy(Http::LowerCaseString(key), value); | ||
| } | ||
| // Removes 'value'; keeps 'key' for next iteration. This is the input for lua_next() so that | ||
| // it can push the next key/value pair onto the stack. | ||
| lua_pop(state, 1); | ||
| } | ||
| } | ||
|
|
||
| Http::AsyncClient::Request* makeHttpCall(lua_State* state, Filter& filter, | ||
| Http::AsyncClient::Callbacks& callbacks) { | ||
| const std::string cluster = luaL_checkstring(state, 2); | ||
| luaL_checktype(state, 3, LUA_TTABLE); | ||
| size_t body_size; | ||
| const char* body = luaL_optlstring(state, 4, nullptr, &body_size); | ||
| int timeout_ms = luaL_checkint(state, 5); | ||
| if (timeout_ms < 0) { | ||
| luaL_error(state, "http call timeout must be >= 0"); | ||
| } | ||
|
|
||
| if (filter.clusterManager().get(cluster) == nullptr) { | ||
| luaL_error(state, "http call cluster invalid. Must be configured"); | ||
| } | ||
|
|
||
| auto headers = std::make_unique<Http::RequestHeaderMapImpl>(); | ||
| buildHeadersFromTable(*headers, state, 3); | ||
| Http::RequestMessagePtr message(new Http::RequestMessageImpl(std::move(headers))); | ||
|
|
||
| // Check that we were provided certain headers. | ||
| if (message->headers().Path() == nullptr || message->headers().Method() == nullptr || | ||
| message->headers().Host() == nullptr) { | ||
| luaL_error(state, "http call headers must include ':path', ':method', and ':authority'"); | ||
| } | ||
|
|
||
| if (body != nullptr) { | ||
| message->body() = std::make_unique<Buffer::OwnedImpl>(body, body_size); | ||
| message->headers().setContentLength(body_size); | ||
| } | ||
|
|
||
| absl::optional<std::chrono::milliseconds> timeout; | ||
| if (timeout_ms > 0) { | ||
| timeout = std::chrono::milliseconds(timeout_ms); | ||
| } | ||
|
|
||
| return filter.clusterManager().httpAsyncClientForCluster(cluster).send( | ||
| std::move(message), callbacks, Http::AsyncClient::RequestOptions().setTimeout(timeout)); | ||
| } | ||
| } // namespace | ||
|
|
||
| StreamHandleWrapper::StreamHandleWrapper(Filters::Common::Lua::Coroutine& coroutine, | ||
| Http::HeaderMap& headers, bool end_stream, Filter& filter, | ||
| FilterCallbacks& callbacks) | ||
|
|
@@ -138,71 +211,18 @@ int StreamHandleWrapper::luaRespond(lua_State* state) { | |
| return lua_yield(state, 0); | ||
| } | ||
|
|
||
| void StreamHandleWrapper::buildHeadersFromTable(Http::HeaderMap& headers, lua_State* state, | ||
| int table_index) { | ||
| // Build a header map to make the request. We iterate through the provided table to do this and | ||
| // check that we are getting strings. | ||
| lua_pushnil(state); | ||
| while (lua_next(state, table_index) != 0) { | ||
| // Uses 'key' (at index -2) and 'value' (at index -1). | ||
| const char* key = luaL_checkstring(state, -2); | ||
| // Check if the current value is a table, we iterate through the table and add each element of | ||
| // it as a header entry value for the current key. | ||
| if (lua_istable(state, -1)) { | ||
| lua_pushnil(state); | ||
| while (lua_next(state, -2) != 0) { | ||
| const char* value = luaL_checkstring(state, -1); | ||
| headers.addCopy(Http::LowerCaseString(key), value); | ||
| lua_pop(state, 1); | ||
| } | ||
| } else { | ||
| const char* value = luaL_checkstring(state, -1); | ||
| headers.addCopy(Http::LowerCaseString(key), value); | ||
| } | ||
| // Removes 'value'; keeps 'key' for next iteration. This is the input for lua_next() so that | ||
| // it can push the next key/value pair onto the stack. | ||
| lua_pop(state, 1); | ||
| } | ||
| } | ||
|
|
||
| int StreamHandleWrapper::luaHttpCall(lua_State* state) { | ||
| ASSERT(state_ == State::Running); | ||
|
|
||
| const std::string cluster = luaL_checkstring(state, 2); | ||
| luaL_checktype(state, 3, LUA_TTABLE); | ||
| size_t body_size; | ||
| const char* body = luaL_optlstring(state, 4, nullptr, &body_size); | ||
| int timeout_ms = luaL_checkint(state, 5); | ||
| if (timeout_ms < 0) { | ||
| return luaL_error(state, "http call timeout must be >= 0"); | ||
| } | ||
|
|
||
| if (filter_.clusterManager().get(cluster) == nullptr) { | ||
| return luaL_error(state, "http call cluster invalid. Must be configured"); | ||
| } | ||
|
|
||
| auto headers = std::make_unique<Http::RequestHeaderMapImpl>(); | ||
| buildHeadersFromTable(*headers, state, 3); | ||
| Http::RequestMessagePtr message(new Http::RequestMessageImpl(std::move(headers))); | ||
|
|
||
| // Check that we were provided certain headers. | ||
| if (message->headers().Path() == nullptr || message->headers().Method() == nullptr || | ||
| message->headers().Host() == nullptr) { | ||
| return luaL_error(state, "http call headers must include ':path', ':method', and ':authority'"); | ||
| } | ||
|
|
||
| if (body != nullptr) { | ||
| message->body() = std::make_unique<Buffer::OwnedImpl>(body, body_size); | ||
| message->headers().setContentLength(body_size); | ||
| } | ||
|
|
||
| absl::optional<std::chrono::milliseconds> timeout; | ||
| if (timeout_ms > 0) { | ||
| timeout = std::chrono::milliseconds(timeout_ms); | ||
| bool asynchronous = lua_toboolean(state, 6); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: |
||
| if (asynchronous) { | ||
| return luaHttpCallAsynchronous(state); | ||
| } else { | ||
| return luaHttpCallSynchronous(state); | ||
| } | ||
| } | ||
|
|
||
| http_request_ = filter_.clusterManager().httpAsyncClientForCluster(cluster).send( | ||
| std::move(message), *this, Http::AsyncClient::RequestOptions().setTimeout(timeout)); | ||
| int StreamHandleWrapper::luaHttpCallSynchronous(lua_State* state) { | ||
| http_request_ = makeHttpCall(state, filter_, *this); | ||
| if (http_request_) { | ||
| state_ = State::HttpCall; | ||
| return lua_yield(state, 0); | ||
|
|
@@ -213,6 +233,11 @@ int StreamHandleWrapper::luaHttpCall(lua_State* state) { | |
| } | ||
| } | ||
|
|
||
| int StreamHandleWrapper::luaHttpCallAsynchronous(lua_State* state) { | ||
| makeHttpCall(state, filter_, noopCallbacks()); | ||
| return 0; | ||
| } | ||
|
|
||
| void StreamHandleWrapper::onSuccess(Http::ResponseMessagePtr&& response) { | ||
| ASSERT(state_ == State::HttpCall || state_ == State::Running); | ||
| ENVOY_LOG(debug, "async HTTP response complete"); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -86,6 +86,8 @@ class FilterCallbacks { | |
|
|
||
| class Filter; | ||
|
|
||
| class NoopCallbacks; | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't think this needs forward declaration? |
||
|
|
||
| /** | ||
| * A wrapper for a currently running request/response. This is the primary handle passed to Lua. | ||
| * The script interacts with Envoy entirely through this handle. | ||
|
|
@@ -156,6 +158,8 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan | |
| * @param 2 (table): A table of HTTP headers. :method, :path, and :authority must be defined. | ||
| * @param 3 (string): Body. Can be nil. | ||
| * @param 4 (int): Timeout in milliseconds for the call. | ||
| * @param 5 (bool): Optional flag. If true, filter continues without waiting for HTTP response | ||
| * from upstream service. False/synchronous by default. | ||
| * @return headers (table), body (string/nil) | ||
| */ | ||
| DECLARE_LUA_FUNCTION(StreamHandleWrapper, luaHttpCall); | ||
|
|
@@ -247,7 +251,8 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan | |
| */ | ||
| DECLARE_LUA_CLOSURE(StreamHandleWrapper, luaBodyIterator); | ||
|
|
||
| static void buildHeadersFromTable(Http::HeaderMap& headers, lua_State* state, int table_index); | ||
| int luaHttpCallSynchronous(lua_State* state); | ||
| int luaHttpCallAsynchronous(lua_State* state); | ||
|
|
||
| // Filters::Common::Lua::BaseLuaObject | ||
| void onMarkDead() override { | ||
|
|
@@ -287,6 +292,17 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan | |
| Http::AsyncClient::Request* http_request_{}; | ||
| }; | ||
|
|
||
| /** | ||
| * An empty Callbacks client. It will ignore everything, including successes and failures. | ||
| */ | ||
| class NoopCallbacks : public Http::AsyncClient::Callbacks { | ||
| public: | ||
| // Http::AsyncClient::Callbacks | ||
| void onSuccess(Http::ResponseMessagePtr&&) override {} | ||
|
|
||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: del newline |
||
| void onFailure(Http::AsyncClient::FailureReason) override {} | ||
| }; | ||
|
|
||
| /** | ||
| * Global configuration for the filter. | ||
| */ | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i'd do something like
* lua: added a parameter to `httpCall` that makes it possible to have the call be asynchronous.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good idea.