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

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

@@ -244,6 +244,23 @@ data to send. *timeout* is an integer that specifies the call timeout in millise
 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.
 
+httpCallAsync()
+^^^^^^^^^^^^^^^
+
+.. code-block:: lua
+
+  headers, body = handle:httpCallAsync(cluster, headers, body, timeout)
+
+Makes an HTTP call to an upstream host. Same behavior as httpCall, except that Envoy will fire and forget.
+
+*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.
+
+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.
+
 respond()
 ^^^^^^^^^^
 

docs/root/intro/version_history.rst

@@ -9,6 +9,7 @@ Version history
   minimum.
 * config: use type URL to select an extension whenever the config type URL (or its previous versions) uniquely identify a typed extension, see :ref:`extension configuration <config_overview_extension_configuration>`.
 * http: fixing a bug in HTTP/1.0 responses where Connection: keep-alive was not appended for connections which were kept alive.
+* lua: added `httpCallAsync` function. Same as `httpCall` but fire-and-forget.
 * 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.
 * rbac: added :ref:`url_path <envoy_api_field_config.rbac.v2.Permission.url_path>` for matching URL path without the query and fragment string.

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

@@ -23,7 +23,10 @@ StreamHandleWrapper::StreamHandleWrapper(Filters::Common::Lua::Coroutine& corout
         if (state_ == State::Running) {
           throw Filters::Common::Lua::LuaException("script performed an unexpected yield");
         }
-      }) {}
+      }),
+      fireAndForgetWriter_(new FireAndForgetWriter(filter)) {}
+
+StreamHandleWrapper::~StreamHandleWrapper() { delete fireAndForgetWriter_; }
 
 Http::FilterHeadersStatus StreamHandleWrapper::start(int function_ref) {
   // We are on the top of the stack.
@@ -116,7 +119,7 @@ int StreamHandleWrapper::luaRespond(lua_State* state) {
   size_t body_size;
   const char* raw_body = luaL_optlstring(state, 3, nullptr, &body_size);
   auto headers = std::make_unique<Http::ResponseHeaderMapImpl>();
-  buildHeadersFromTable(*headers, state, 2);
+  LuaFilterUtil::buildHeadersFromTable(*headers, state, 2);
 
   uint64_t status;
   if (headers->Status() == nullptr ||
@@ -138,71 +141,9 @@ 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);
-  }
-
-  http_request_ = filter_.clusterManager().httpAsyncClientForCluster(cluster).send(
-      std::move(message), *this, Http::AsyncClient::RequestOptions().setTimeout(timeout));
+  http_request_ = LuaFilterUtil::makeHttpCall(state, filter_, *this);
   if (http_request_) {
     state_ = State::HttpCall;
     return lua_yield(state, 0);
@@ -213,6 +154,10 @@ int StreamHandleWrapper::luaHttpCall(lua_State* state) {
   }
 }
 
+int StreamHandleWrapper::luaHttpCallAsync(lua_State* state) {
+  return fireAndForgetWriter_->luaHttpCallAsync(state);
+}
+
 void StreamHandleWrapper::onSuccess(Http::ResponseMessagePtr&& response) {
   ASSERT(state_ == State::HttpCall || state_ == State::Running);
   ENVOY_LOG(debug, "async HTTP response complete");
@@ -485,6 +430,84 @@ int StreamHandleWrapper::luaImportPublicKey(lua_State* state) {
   return 1;
 }
 
+FireAndForgetWriter::FireAndForgetWriter(Filter& filter) : filter_(filter) {}
+
+int FireAndForgetWriter::luaHttpCallAsync(lua_State* state) {
+  if (LuaFilterUtil::makeHttpCall(state, filter_, *this)) {
+    return 0;
+  } else {
+    return 2;
+  }
+}
+
+Http::AsyncClient::Request*
+LuaFilterUtil::makeHttpCall(lua_State* state, Filter& filter,
+                            Http::AsyncClient::Callbacks& callbacksListener) {
+  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>();
+  LuaFilterUtil::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), callbacksListener,
+      Http::AsyncClient::RequestOptions().setTimeout(timeout));
+}
+
+void LuaFilterUtil::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);
+  }
+}
+
 FilterConfig::FilterConfig(const std::string& lua_code, ThreadLocal::SlotAllocator& tls,
                            Upstream::ClusterManager& cluster_manager)
     : cluster_manager_(cluster_manager), lua_state_(lua_code, tls) {

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

@@ -86,6 +86,8 @@ class FilterCallbacks {
 
 class Filter;
 
+class FireAndForgetWriter;
+
 /**
  * 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.
@@ -117,6 +119,7 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
 
   StreamHandleWrapper(Filters::Common::Lua::Coroutine& coroutine, Http::HeaderMap& headers,
                       bool end_stream, Filter& filter, FilterCallbacks& callbacks);
+  ~StreamHandleWrapper();
 
   Http::FilterHeadersStatus start(int function_ref);
   Http::FilterDataStatus onData(Buffer::Instance& data, bool end_stream);
@@ -142,6 +145,7 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
             {"logErr", static_luaLogErr},
             {"logCritical", static_luaLogCritical},
             {"httpCall", static_luaHttpCall},
+            {"httpCallAsync", static_luaHttpCallAsync},
             {"respond", static_luaRespond},
             {"streamInfo", static_luaStreamInfo},
             {"connection", static_luaConnection},
@@ -160,6 +164,16 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
    */
   DECLARE_LUA_FUNCTION(StreamHandleWrapper, luaHttpCall);
 
+  /**
+   * Perform an asynchronous HTTP call to an upstream host. Fires and forgets.
+   * @param 1 (string): The name of the upstream cluster to call. This cluster must be configured.
+   * @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.
+   * @return headers (table), body (string/nil)
+   */
+  DECLARE_LUA_FUNCTION(StreamHandleWrapper, luaHttpCallAsync);
+
   /**
    * Perform an inline response. This call is currently only valid on the request path. Further
    * filter iteration will stop. No further script code will run after this call.
@@ -247,8 +261,6 @@ 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);
-
   // Filters::Common::Lua::BaseLuaObject
   void onMarkDead() override {
     // Headers/body/trailers wrappers do not survive any yields. The user can request them
@@ -285,6 +297,38 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
   State state_{State::Running};
   std::function<void()> yield_callback_;
   Http::AsyncClient::Request* http_request_{};
+  FireAndForgetWriter* fireAndForgetWriter_;
+};
+
+/**
+ * Implementation that takes incoming requests and implements
+ * "fire and forget" behavior using an async client.
+ */
+class FireAndForgetWriter : public Filters::Common::Lua::BaseLuaObject<FireAndForgetWriter>,
+                            public Http::AsyncClient::Callbacks {
+public:
+  FireAndForgetWriter(Filter& filter);
+
+  int luaHttpCallAsync(lua_State* state);
+
+  // Http::AsyncClient::Callbacks
+  void onSuccess(Http::ResponseMessagePtr&&) override {}
+
+  void onFailure(Http::AsyncClient::FailureReason) override {}
+
+private:
+  Filter& filter_;
+};
+
+/**
+ * A class with shared code for building and making http calls
+ */
+class LuaFilterUtil : public Filters::Common::Lua::BaseLuaObject<LuaFilterUtil> {
+public:
+  static Http::AsyncClient::Request* makeHttpCall(lua_State* state, Filter& filter,
+                                                  Http::AsyncClient::Callbacks& callbacksListener);
+
+  static void buildHeadersFromTable(Http::HeaderMap& headers, lua_State* state, int table_index);
 };
 
 /**

test/extensions/filters/http/lua/lua_filter_test.cc

@@ -799,6 +799,59 @@ TEST_F(LuaHttpFilterTest, HttpCall) {
   callbacks->onSuccess(std::move(response_message));
 }
 
+// Basic Asynchronous, Fire and Forget HTTP request flow.
+TEST_F(LuaHttpFilterTest, HttpCallAsync) {
+  const std::string SCRIPT{R"EOF(
+        function envoy_on_request(request_handle)
+          local headers, body = request_handle:httpCallAsync(
+            "cluster",
+            {
+              [":method"] = "POST",
+              [":path"] = "/",
+              [":authority"] = "foo",
+              ["set-cookie"] = { "flavor=chocolate; Path=/", "variant=chewy; Path=/" }
+            },
+            "hello world",
+            5000)
+        end
+      )EOF"};
+
+  InSequence s;
+  setup(SCRIPT);
+
+  Http::TestRequestHeaderMapImpl request_headers{{":path", "/"}};
+  Http::MockAsyncClientRequest request(&cluster_manager_.async_client_);
+  Http::AsyncClient::Callbacks* callbacks;
+  EXPECT_CALL(cluster_manager_, get(Eq("cluster")));
+  EXPECT_CALL(cluster_manager_, httpAsyncClientForCluster("cluster"));
+  EXPECT_CALL(cluster_manager_.async_client_, send_(_, _, _))
+      .WillOnce(
+          Invoke([&](Http::RequestMessagePtr& message, Http::AsyncClient::Callbacks& cb,
+                     const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
+            EXPECT_EQ((Http::TestHeaderMapImpl{{":path", "/"},
+                                               {":method", "POST"},
+                                               {":authority", "foo"},
+                                               {"set-cookie", "flavor=chocolate; Path=/"},
+                                               {"set-cookie", "variant=chewy; Path=/"},
+                                               {"content-length", "11"}}),
+                      message->headers());
+            callbacks = &cb;
+            return &request;
+          }));
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers, false));
+
+  Buffer::OwnedImpl data("hello");
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->decodeData(data, false));
+
+  Http::TestRequestTrailerMapImpl request_trailers{{"foo", "bar"}};
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->decodeTrailers(request_trailers));
+
+  Http::ResponseMessagePtr response_message(new Http::ResponseMessageImpl(
+      Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
+  response_message->body() = std::make_unique<Buffer::OwnedImpl>("response");
+}
+
 // Double HTTP call. Responses before request body.
 TEST_F(LuaHttpFilterTest, DoubleHttpCall) {
   const std::string SCRIPT{R"EOF(