fix: httpc connection close

apisix/discovery/nacos/init.lua

@@ -19,6 +19,7 @@ local require            = require
 local local_conf         = require('apisix.core.config_local').local_conf()
 local http               = require('resty.http')
 local core               = require('apisix.core')
+local connection_util = require("apisix.utils.connection-util")
 local ipairs             = ipairs
 local type               = type
 local math               = math
@@ -89,6 +90,8 @@ local function request(request_uri, path, body, method, basic_auth)
         body = body,
         ssl_verify = true,
     })
+    connection_util.close_http_connection(httpc)
+
     if not res then
         return nil, err
     end

apisix/plugins/authz-casdoor.lua

@@ -15,6 +15,7 @@
 -- limitations under the License.
 --
 local core = require("apisix.core")
+local connection_util = require("apisix.utils.connection-util")
 local http = require("resty.http")
 local session = require("resty.session")
 local ngx = ngx
@@ -60,7 +61,7 @@ local function fetch_access_token(code, conf)
             ["Content-Type"] = "application/x-www-form-urlencoded"
         }
     })
-
+    connection_util.close_http_connection(client)
     if not res then
         return nil, nil, err
     end

apisix/plugins/batch-requests.lua

@@ -17,6 +17,7 @@
 local core      = require("apisix.core")
 local http      = require("resty.http")
 local plugin    = require("apisix.plugin")
+local connection_util = require("apisix.utils.connection-util")
 local ngx       = ngx
 local ipairs    = ipairs
 local pairs     = pairs
@@ -258,6 +259,7 @@ local function batch_requests(ctx)
 
     local responses, err = httpc:request_pipeline(data.pipeline)
     if not responses then
+        connection_util.close_http_connection(httpc)
         return 400, {error_msg = "request failed: " .. err}
     end
 
@@ -286,6 +288,7 @@ local function batch_requests(ctx)
         end
         core.table.insert(aggregated_resp, sub_resp)
     end
+    connection_util.close_http_connection(httpc)
     return 200, aggregated_resp
 end
 

apisix/plugins/cas-auth.lua

@@ -16,6 +16,7 @@
 ----
 local core = require("apisix.core")
 local http = require("resty.http")
+local connection_util = require("apisix.utils.connection-util")
 local ngx = ngx
 local ngx_re_match = ngx.re.match
 
@@ -130,6 +131,7 @@ local function validate(conf, ctx, ticket)
         core.log.error("validate ticket failed: status=", (res and res.status),
             ", has_body=", (res and res.body ~= nil or false), ", err=", err)
     end
+    connection_util.close_http_connection(httpc)
     return nil
 end
 

apisix/plugins/clickhouse-logger.lua

@@ -17,6 +17,7 @@
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local log_util        = require("apisix.utils.log-util")
+local connection_util = require("apisix.utils.connection-util")
 local core            = require("apisix.core")
 local http            = require("resty.http")
 local url             = require("net.url")
@@ -151,6 +152,7 @@ local function send_http_data(conf, log_message)
         }
     })
 
+    connection_util.close_http_connection(httpc)
     if not httpc_res then
         return false, "error while sending data to [" .. host .. "] port["
             .. tostring(port) .. "] " .. httpc_err

apisix/plugins/elasticsearch-logger.lua

@@ -18,6 +18,7 @@
 local core            = require("apisix.core")
 local http            = require("resty.http")
 local log_util        = require("apisix.utils.log-util")
+local connection_util = require("apisix.utils.connection-util")
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 
 local ngx             = ngx
@@ -178,6 +179,8 @@ local function send_to_elasticsearch(conf, entries)
         headers = headers,
         body = body
     })
+
+    connection_util.close_http_connection(httpc)
     if not resp then
         return false, err
     end

apisix/plugins/error-log-logger.lua

@@ -18,6 +18,7 @@
 local core = require("apisix.core")
 local errlog = require("ngx.errlog")
 local batch_processor = require("apisix.utils.batch-processor")
+local connection_util = require("apisix.utils.connection-util")
 local plugin = require("apisix.plugin")
 local timers = require("apisix.timers")
 local http = require("resty.http")
@@ -273,6 +274,7 @@ local function send_to_skywalking(log_message)
         }
     )
 
+    connection_util.close_http_connection(httpc)
     if not httpc_res then
         return false, "error while sending data to skywalking["
             .. config.skywalking.endpoint_addr .. "] " .. httpc_err
@@ -324,6 +326,8 @@ local function send_to_clickhouse(log_message)
         }
     )
 
+    connection_util.close_http_connection(httpc)
+
     if not httpc_res then
         return false, "error while sending data to clickhouse["
             .. config.clickhouse.endpoint_addr .. "] " .. httpc_err

apisix/plugins/google-cloud-logging.lua

@@ -19,6 +19,7 @@ local core            = require("apisix.core")
 local tostring        = tostring
 local http            = require("resty.http")
 local log_util        = require("apisix.utils.log-util")
+local connection_util = require("apisix.utils.connection-util")
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local google_oauth    = require("apisix.plugins.google-cloud-logging.oauth")
 
@@ -125,7 +126,7 @@ local function send_to_google(oauth, entries)
             ["Authorization"] = (oauth.access_token_type or "Bearer") .. " " .. access_token,
         },
     })
-
+    connection_util.close_http_connection(http_new)
     if not res then
         return nil, "failed to write log to google, " .. err
     end

apisix/plugins/google-cloud-logging/oauth.lua

@@ -16,6 +16,7 @@
 --
 
 local core = require("apisix.core")
+local connection_util = require("apisix.utils.connection-util")
 local type = type
 local setmetatable = setmetatable
 
@@ -57,6 +58,7 @@ function _M:refresh_access_token()
             ["Content-Type"] = "application/x-www-form-urlencoded",
         },
     })
+    connection_util.close_http_connection(httpc)
 
     if not res then
         core.log.error("failed to refresh google oauth access token, ", err)

apisix/plugins/http-logger.lua

@@ -17,6 +17,7 @@
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local log_util        = require("apisix.utils.log-util")
+local connection_util = require("apisix.utils.connection-util")
 local core            = require("apisix.core")
 local http            = require("resty.http")
 local url             = require("net.url")
@@ -146,6 +147,7 @@ local function send_http_data(conf, log_message)
     })
 
     if not httpc_res then
+        connection_util.close_http_connection(httpc)
         return false, "error while sending data to [" .. host .. "] port["
             .. tostring(port) .. "] " .. httpc_err
     end
@@ -158,6 +160,7 @@ local function send_http_data(conf, log_message)
             .. "body[" .. httpc_res:read_body() .. "]"
     end
 
+    connection_util.close_http_connection(httpc)
     return res, err_msg
 end
 

apisix/plugins/skywalking-logger.lua

@@ -17,6 +17,7 @@
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local log_util        = require("apisix.utils.log-util")
+local connection_util = require("apisix.utils.connection-util")
 local core            = require("apisix.core")
 local http            = require("resty.http")
 local url             = require("net.url")
@@ -115,7 +116,7 @@ local function send_http_data(conf, log_message)
             ["Content-Type"] = "application/json",
         }
     })
-
+    connection_util.close_http_connection(httpc)
     if not httpc_res then
         return false, "error while sending data to [" .. host .. "] port["
             .. tostring(port) .. "] " .. httpc_err

apisix/utils/connection-util.lua

@@ -0,0 +1,28 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements.  See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License.  You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+local core = require("apisix.core")
+
+local _M = {}
+
+function _M.close_http_connection(httpc)
+    local ok, con_err = httpc:close()
+    if not ok then
+        core.log.warn("Failed to close connection: ", con_err)
+    end
+end
+
+return _M

t/lib/grafana_loki.lua

@@ -17,6 +17,7 @@
 
 local cjson = require("cjson")
 local http = require("resty.http")
+local connection_util = require("apisix.utils.connection-util")
 
 local _M = {}
 
@@ -45,17 +46,22 @@ function _M.fetch_logs_from_loki(from, to, options)
     })
 
     if not res or err then
+        connection_util.close_http_connection(httpc)
         return nil, err
     end
 
     if res.status > 300 then
+        connection_util.close_http_connection(httpc)
         return nil, "HTTP status code: " .. res.status .. ", body: " .. res.body
     end
 
     local data = cjson.decode(res.body)
     if not data then
+        connection_util.close_http_connection(httpc)
         return nil, "failed to decode response body: " .. res.body
     end
+
+    connection_util.close_http_connection(httpc)
     return data, nil
 end
 

t/lib/keycloak.lua

@@ -15,6 +15,7 @@
 -- limitations under the License.
 --
 local http = require "resty.http"
+local connection_util = require("apisix.utils.log-util")
 
 local _M = {}
 
@@ -30,6 +31,7 @@ function _M.login_keycloak(uri, username, password)
     elseif res.status ~= 302 then
         -- Not a redirect which we expect.
         -- Use 500 to indicate error.
+        connection_util.close_http_connection(httpc)
         return nil, "Initial request was not redirected to ID provider authorization endpoint."
     else
         -- Extract cookies. Important since OIDC module tracks state with a session cookie.
@@ -44,15 +46,18 @@ function _M.login_keycloak(uri, username, password)
         -- possibly even the version used.
         res, err = httpc:request_uri(res.headers['Location'], {method = "GET"})
         if not res then
+            connection_util.close_http_connection(httpc)
             -- No response, must be an error.
             return nil, err
         elseif res.status ~= 200 then
+            connection_util.close_http_connection(httpc)
             -- Unexpected response.
             return nil, res.body
         end
 
         -- Check if response code was ok.
         if res.status ~= 200 then
+            connection_util.close_http_connection(httpc)
             return nil, "unexpected status " .. res.status
         end
 
@@ -81,9 +86,11 @@ function _M.login_keycloak(uri, username, password)
             })
         if not res then
             -- No response, must be an error.
+            connection_util.close_http_connection(httpc)
             return nil, err
         elseif res.status ~= 302 then
             -- Not a redirect which we expect.
+            connection_util.close_http_connection(httpc)
             return nil, "Login form submission did not return redirect to redirect URI."
         end
 
@@ -98,7 +105,7 @@ function _M.login_keycloak(uri, username, password)
                     ["Cookie"] = cookie_str
                 }
             })
-
+        connection_util.close_http_connection(httpc)
         if not res then
             -- No response, must be an error.
             return nil, err