fix: httpc connection close

apisix/plugins/batch-requests.lua

@@ -249,6 +249,7 @@ local function batch_requests(ctx)
     httpc:set_timeout(data.timeout)
     local ok, err = httpc:connect("127.0.0.1", ngx.var.server_port)
     if not ok then
+        httpc:close()
         return 500, {error_msg = "connect to apisix failed: " .. err}
     end
 
@@ -258,6 +259,7 @@ local function batch_requests(ctx)
 
     local responses, err = httpc:request_pipeline(data.pipeline)
     if not responses then
+        httpc:close()
         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
+    httpc:close()
     return 200, aggregated_resp
 end
 

apisix/plugins/clickhouse-logger.lua

@@ -38,6 +38,9 @@ local schema = {
         database = {type = "string", default = ""},
         logtable = {type = "string", default = ""},
         timeout = {type = "integer", minimum = 1, default = 3},
+        keepalive = {type = "boolean", default = true},
+        keepalive_timeout = {type = "integer", minimum = 1, default = 60},
+        keepalive_pool = {type = "integer", minimum = 1, default = 5},
         name = {type = "string", default = "clickhouse logger"},
         ssl_verify = {type = "boolean", default = true},
         log_format = {type = "object"},
@@ -120,6 +123,25 @@ local function send_http_data(conf, log_message)
         end
     end
 
+    local request_uri = url_decoded.scheme .. "://" .. host .. ":" .. tostring(port)
+    .. (#url_decoded.path ~= 0 and url_decoded.path or "/")
+
+    local auth_headers = {
+         ["Host"] = host,
+         ["Content-Type"] = "application/json",
+         ["X-ClickHouse-User"] = conf.user,
+         ["X-ClickHouse-Key"] = conf.password,
+         ["X-ClickHouse-Database"] = conf.database
+    }
+
+    local params = {
+        headers = auth_headers,
+        keepalive = conf.keepalive,
+        method = "POST",
+        query = url_decoded.query,
+        body = "INSERT INTO " .. conf.logtable .." FORMAT JSONEachRow " .. log_message,
+    }
+
     local httpc = http.new()
     httpc:set_timeout(conf.timeout * 1000)
     local ok, err = httpc:connect(host, port)
@@ -136,20 +158,12 @@ local function send_http_data(conf, log_message)
                 .. "port[" .. tostring(port) .. "] " .. err
         end
     end
+    if conf.keepalive then
+        params.keepalive_timeout = conf.keepalive_timeout * 1000
+        params.keepalive_pool = conf.keepalive_pool
+    end
 
-    local httpc_res, httpc_err = httpc:request({
-        method = "POST",
-        path = url_decoded.path,
-        query = url_decoded.query,
-        body = "INSERT INTO " .. conf.logtable .." FORMAT JSONEachRow " .. log_message,
-        headers = {
-            ["Host"] = url_decoded.host,
-            ["Content-Type"] = "application/json",
-            ["X-ClickHouse-User"] = conf.user,
-            ["X-ClickHouse-Key"] = conf.password,
-            ["X-ClickHouse-Database"] = conf.database
-        }
-    })
+    local httpc_res, httpc_err = httpc:request_uri(request_uri, params)
 
     if not httpc_res then
         return false, "error while sending data to [" .. host .. "] port["

apisix/plugins/http-logger.lua

@@ -33,6 +33,9 @@ local schema = {
         uri = core.schema.uri_def,
         auth_header = {type = "string"},
         timeout = {type = "integer", minimum = 1, default = 3},
+        keepalive = {type = "boolean", default = true},
+        keepalive_timeout = {type = "integer", minimum = 1, default = 60},
+        keepalive_pool = {type = "integer", minimum = 1, default = 5},
         log_format = {type = "object"},
         include_req_body = {type = "boolean", default = false},
         include_req_body_expr = {
@@ -133,17 +136,30 @@ local function send_http_data(conf, log_message)
         content_type = "text/plain"
     end
 
-    local httpc_res, httpc_err = httpc:request({
+    local auth_headers = {
+         ["Host"] = host,
+         ["Content-Type"] = content_type,
+         ["Authorization"] = conf.auth_header
+    }
+
+    local params = {
+        headers = auth_headers,
+        keepalive = conf.keepalive,
+        ssl_verify = conf.ssl_verify,
         method = "POST",
-        path = #url_decoded.path ~= 0 and url_decoded.path or "/",
         query = url_decoded.query,
         body = log_message,
-        headers = {
-            ["Host"] = url_decoded.host,
-            ["Content-Type"] = content_type,
-            ["Authorization"] = conf.auth_header
-        }
-    })
+    }
+
+    local request_uri = url_decoded.scheme .. "://" .. host .. ":" .. tostring(port)
+    .. (#url_decoded.path ~= 0 and url_decoded.path or "/")
+
+    if conf.keepalive then
+        params.keepalive_timeout = conf.keepalive_timeout * 1000
+        params.keepalive_pool = conf.keepalive_pool
+    end
+
+    local httpc_res, httpc_err = httpc:request_uri(request_uri, params)
 
     if not httpc_res then
         return false, "error while sending data to [" .. host .. "] port["

apisix/plugins/skywalking-logger.lua

@@ -38,6 +38,10 @@ local schema = {
         service_instance_name = {type = "string", default = "APISIX Instance Name"},
         log_format = {type = "object"},
         timeout = {type = "integer", minimum = 1, default = 3},
+        keepalive = {type = "boolean", default = true},
+        keepalive_timeout = {type = "integer", minimum = 1, default = 60},
+        keepalive_pool = {type = "integer", minimum = 1, default = 5},
+        ssl_verify = {type = "boolean", default = false},
         include_req_body = {type = "boolean", default = false},
         include_req_body_expr = {
             type = "array",
@@ -97,24 +101,50 @@ local function send_http_data(conf, log_message)
 
     core.log.info("sending a batch logs to ", conf.endpoint_addr)
 
+    if ((not port) and url_decoded.scheme == "https") then
+        port = 443
+    elseif not port then
+        port = 80
+    end
+
+    local request_uri = url_decoded.scheme .. "://" .. host .. ":" .. tostring(port) .. "/v3/logs"
+
+    local auth_headers = {
+         ["Host"] = host,
+         ["Content-Type"] = "application/json",
+    }
+
+    local params = {
+        headers = auth_headers,
+        keepalive = conf.keepalive,
+        method = "POST",
+        body = log_message,
+    }
+
     local httpc = http.new()
     httpc:set_timeout(conf.timeout * 1000)
+
+    if conf.keepalive then
+        params.keepalive_timeout = conf.keepalive_timeout * 1000
+        params.keepalive_pool = conf.keepalive_pool
+    end
+
     local ok, err = httpc:connect(host, port)
 
     if not ok then
         return false, "failed to connect to host[" .. host .. "] port["
             .. tostring(port) .. "] " .. err
     end
 
-    local httpc_res, httpc_err = httpc:request({
-        method = "POST",
-        path = "/v3/logs",
-        body = log_message,
-        headers = {
-            ["Host"] = url_decoded.host,
-            ["Content-Type"] = "application/json",
-        }
-    })
+    if url_decoded.scheme == "https" then
+        ok, err = httpc:ssl_handshake(true, host, conf.ssl_verify)
+        if not ok then
+            return false, "failed to perform SSL with host[" .. host .. "] "
+                .. "port[" .. tostring(port) .. "] " .. err
+        end
+    end
+
+    local httpc_res, httpc_err = httpc:request_uri(request_uri, params)
 
     if not httpc_res then
         return false, "error while sending data to [" .. host .. "] port["