lua: add fire-and-forget functionality to http call

docs/root/configuration/http/http_filters/lua_filter.rst

@@ -233,14 +233,17 @@ httpCall()
 
 .. code-block:: lua
 
-  headers, body = handle:httpCall(cluster, headers, body, timeout)
+  headers, body = handle:httpCall(cluster, headers, body, timeout, asynchronous)
 
-Makes an HTTP call to an upstream host. Envoy will yield the script until the call completes or
-has an error. *cluster* is a string which maps to a configured cluster manager cluster. *headers*
+Makes an HTTP call to an upstream host. *cluster* is a string which maps to a configured cluster manager cluster. *headers*
 is a table of key/value pairs to send (the value can be a string or table of strings). Note that
 the *:method*, *:path*, and *:authority* headers must be set. *body* is an optional string of body
 data to send. *timeout* is an integer that specifies the call timeout in milliseconds.
 
+*asynchronous* is a boolean flag. If asynchronous is set to true, Envoy will make the HTTP request and continue,
+regardless of response success or failure. If this is set to false, or not set, Envoy will yield the script
+until the call completes or has an error.
+
 Returns *headers* which is a table of response headers. Returns *body* which is the string response
 body. May be nil if there is no body.
 

docs/root/intro/version_history.rst

@@ -20,6 +20,7 @@ Version history
 * listener filters: listener filter extensions use the "envoy.filters.listener" name space. A
   mapping of extension names is available in the :ref:`deprecated <deprecated>` documentation.
 * listeners: fixed issue where :ref:`TLS inspector listener filter <config_listener_filters_tls_inspector>` could have been bypassed by a client using only TLS 1.3.
+* lua: added a parameter to `httpCall` that makes it possible to have the call be asynchronous.
 * mongo: the stat emitted for queries without a max time set in the :ref:`MongoDB filter<config_network_filters_mongo_proxy>` was modified to emit correctly for Mongo v3.2+.
 * network filters: network filter extensions use the "envoy.filters.network" name space. A mapping
   of extension names is available in the :ref:`deprecated <deprecated>` documentation.

source/extensions/filters/http/lua/lua_filter.cc

@@ -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,23 @@ 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);
+  const int async_flag_index = 6;
+  if (!lua_isnone(state, async_flag_index) && !lua_isboolean(state, async_flag_index)) {
+    luaL_error(state, "http call asynchronous flag must be 'true', 'false', or empty");
   }
 
-  absl::optional<std::chrono::milliseconds> timeout;
-  if (timeout_ms > 0) {
-    timeout = std::chrono::milliseconds(timeout_ms);
+  if (lua_toboolean(state, async_flag_index)) {
+    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 +238,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");

source/extensions/filters/http/lua/lua_filter.h

@@ -156,6 +156,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 +249,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 +290,16 @@ 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 {}
+  void onFailure(Http::AsyncClient::FailureReason) override {}
+};
+
 /**
  * Global configuration for the filter.
  */