feat(db): exposed postgres connection pooling config

kong.conf.default

@@ -1124,6 +1124,27 @@
                                  # Detailed discussion of this behavior is
                                  # available in the online documentation.
 
+#pg_connection_keepalive_timeout = 60000 # Specify the maximal idle timeout ( in ms )
+                                         # for the postgres connections in the pool.
+                                         # If values is set to 0 then the timeout interval
+                                         # is unlimited.
+
+#pg_connection_pool_size = 30    # Specifies the size limit (in terms of connection count) for
+                                 # postgres server.
+                                 # Note that this connection pool is per nginx worker rather than per kong instance,
+                                 # so size limit specified applies to every single Nginx worker process.
+
+#pg_connection_backlog =         # If specified, this will limit the total of number of opened connections
+                                 # to postgres server to `pg_connection_pool_size`. If the connection pool is full,
+                                 # subsequent connect operations will be queued into a queue equal to this option's value.
+                                 #
+                                 # If the number of queued connect operations is equal to `pg_connection_backlog`,
+                                 # subsequent connect operations will fail and return nil plus the
+                                 # error string "too many waiting connect operations"
+                                 #
+                                 # If not specified, then number of open connections to postgres server is not limited
+                                 # and this the default behaviour.
+
 #pg_ro_host =                    # Same as `pg_host`, but for the
                                  # read-only connection.
                                  # **Note:** Refer to the documentation

kong/conf_loader/init.lua

@@ -355,6 +355,9 @@ local CONF_INFERENCES = {
   pg_ssl_verify = { typ = "boolean" },
   pg_max_concurrent_queries = { typ = "number" },
   pg_semaphore_timeout = { typ = "number" },
+  pg_connection_keepalive_timeout = { typ = "number" },
+  pg_connection_pool_size = { typ = "number" },
+  pg_connection_backlog = { typ = "number" },
 
   pg_ro_port = { typ = "number" },
   pg_ro_timeout = { typ = "number" },
@@ -363,6 +366,9 @@ local CONF_INFERENCES = {
   pg_ro_ssl_verify = { typ = "boolean" },
   pg_ro_max_concurrent_queries = { typ = "number" },
   pg_ro_semaphore_timeout = { typ = "number" },
+  pg_ro_connection_keepalive_timeout = { typ = "number" },
+  pg_ro_connection_pool_size = { typ = "number" },
+  pg_ro_connection_backlog = { typ = "number" },
 
   cassandra_contact_points = { typ = "array" },
   cassandra_port = { typ = "number" },
@@ -959,6 +965,32 @@ local function check_and_infer(conf, opts)
     errors[#errors + 1] = "pg_semaphore_timeout must be an integer greater than 0"
   end
 
+  if conf.pg_connection_keepalive_timeout < 0 then
+    errors[#errors + 1] = "pg_connection_keepalive_timeout must be greater than 0"
+  end
+
+  if conf.pg_connection_keepalive_timeout ~= floor(conf.pg_connection_keepalive_timeout) then
+    errors[#errors + 1] = "pg_connection_keepalive_timeout must be an integer greater than 0"
+  end
+
+  if conf.pg_connection_pool_size < 0 then
+    errors[#errors + 1] = "pg_connection_pool_size must be greater than 0"
+  end
+
+  if conf.pg_connection_pool_size ~= floor(conf.pg_connection_pool_size) then
+    errors[#errors + 1] = "pg_connection_pool_size must be an integer greater than 0"
+  end
+
+  if conf.pg_connection_backlog then
+    if conf.pg_connection_backlog < 0 then
+      errors[#errors + 1] = "pg_connection_backlog must be greater than 0"
+    end
+
+    if conf.pg_connection_backlog ~= floor(conf.pg_connection_backlog) then
+      errors[#errors + 1] = "pg_connection_backlog must be an integer greater than 0"
+    end
+  end
+
   if conf.pg_ro_max_concurrent_queries then
     if conf.pg_ro_max_concurrent_queries < 0 then
       errors[#errors + 1] = "pg_ro_max_concurrent_queries must be greater than 0"
@@ -979,6 +1011,36 @@ local function check_and_infer(conf, opts)
     end
   end
 
+  if conf.pg_ro_connection_keepalive_timeout then
+    if conf.pg_ro_connection_keepalive_timeout < 0 then
+      errors[#errors + 1] = "pg_ro_connection_keepalive_timeout must be greater than 0"
+    end
+
+    if conf.pg_ro_connection_keepalive_timeout ~= floor(conf.pg_ro_connection_keepalive_timeout) then
+      errors[#errors + 1] = "pg_ro_connection_keepalive_timeout must be an integer greater than 0"
+    end
+  end
+
+  if conf.pg_ro_connection_pool_size then
+    if conf.pg_ro_connection_pool_size < 0 then
+      errors[#errors + 1] = "pg_ro_connection_pool_size must be greater than 0"
+    end
+
+    if conf.pg_ro_connection_pool_size ~= floor(conf.pg_ro_connection_pool_size) then
+      errors[#errors + 1] = "pg_ro_connection_pool_size must be an integer greater than 0"
+    end
+  end
+
+  if conf.pg_ro_connection_backlog then
+    if conf.pg_ro_connection_backlog < 0 then
+      errors[#errors + 1] = "pg_ro_connection_backlog must be greater than 0"
+    end
+
+    if conf.pg_ro_connection_backlog ~= floor(conf.pg_ro_connection_backlog) then
+      errors[#errors + 1] = "pg_ro_connection_backlog must be an integer greater than 0"
+    end
+  end
+
   if conf.worker_state_update_frequency <= 0 then
     errors[#errors + 1] = "worker_state_update_frequency must be greater than 0"
   end
@@ -1785,7 +1847,7 @@ local function load(path, custom_conf, opts)
                       conf.stream_proxy_ssl_enabled or
                       conf.admin_ssl_enabled or
                       conf.status_ssl_enabled
-  
+
   for _, name in ipairs({ "nginx_http_directives", "nginx_stream_directives" }) do
     for i, directive in ipairs(conf[name]) do
       if directive.name == "ssl_dhparam" then

kong/db/strategies/postgres/connector.lua

@@ -233,7 +233,7 @@ local function reconnect(config)
       "SET TIME ZONE ", connection:escape_literal("UTC"), ";",
     })
     if not ok then
-      setkeepalive(connection)
+      setkeepalive(connection, config.keepalive_timeout)
       return nil, err
     end
   end
@@ -247,7 +247,7 @@ local function connect(config)
 end
 
 
-setkeepalive = function(connection)
+setkeepalive = function(connection, keepalive_timeout)
   if not connection or not connection.sock then
     return true
   end
@@ -259,7 +259,7 @@ setkeepalive = function(connection)
     end
 
   else
-    local _, err = connection:keepalive()
+    local _, err = connection:keepalive(keepalive_timeout)
     if err then
       return nil, err
     end
@@ -284,6 +284,14 @@ function _mt:get_stored_connection(operation)
   end
 end
 
+function _mt:get_keepalive_timeout(operation)
+  if self.config_ro and operation == 'read' then
+    return self.config_ro.keepalive_timeout
+  end
+
+  return self.config.keepalive_timeout
+end
+
 
 function _mt:init()
   local res, err = self:query("SHOW server_version_num;")
@@ -433,7 +441,8 @@ function _mt:setkeepalive()
   for operation in pairs(OPERATIONS) do
     local conn = self:get_stored_connection(operation)
     if conn then
-      local _, err = setkeepalive(conn)
+      local keepalive_timeout = self:get_keepalive_timeout(operation)
+      local _, err = setkeepalive(conn, keepalive_timeout)
 
       self:store_connection(nil, operation)
 
@@ -526,7 +535,8 @@ function _mt:query(sql, operation)
 
     res, err, partial, num_queries = connection:query(sql)
 
-    setkeepalive(connection)
+    local keepalive_timeout = self:get_keepalive_timeout(operation)
+    setkeepalive(connection, keepalive_timeout)
   end
 
   self:release_query_semaphore_resource(operation)
@@ -922,6 +932,11 @@ function _M.new(kong_config)
     cafile      = kong_config.lua_ssl_trusted_certificate_combined,
     sem_max     = kong_config.pg_max_concurrent_queries or 0,
     sem_timeout = (kong_config.pg_semaphore_timeout or 60000) / 1000,
+    pool_size   = kong_config.pg_connection_pool_size,
+    backlog     = kong_config.pg_connection_backlog or nil,
+
+    --- not used directly by pgmoon, but used internally in connector to set the keepalive timeout
+    keepalive_timeout = kong_config.pg_connection_keepalive_timeout,
   }
 
   local refs = kong_config["$refs"]
@@ -972,6 +987,11 @@ function _M.new(kong_config)
       sem_max     = kong_config.pg_ro_max_concurrent_queries,
       sem_timeout = kong_config.pg_ro_semaphore_timeout and
                     (kong_config.pg_ro_semaphore_timeout / 1000) or nil,
+      pool_size   = kong_config.pg__ro_connection_pool_size,
+      backlog     = kong_config.pg_ro_connection_backlog or nil,
+
+      --- not used directly by pgmoon, but used internally in connector to set the keepalive timeout
+      keepalive_timeout = kong_config.pg_ro_connection_keepalive_timeout,
     }
 
     if refs then

kong/templates/kong_defaults.lua

@@ -100,6 +100,9 @@ pg_ssl = off
 pg_ssl_verify = off
 pg_max_concurrent_queries = 0
 pg_semaphore_timeout = 60000
+pg_connection_keepalive_timeout = 60000
+pg_connection_pool_size = 30
+pg_connection_backlog = NONE
 
 pg_ro_host = NONE
 pg_ro_port = NONE
@@ -112,6 +115,9 @@ pg_ro_ssl = NONE
 pg_ro_ssl_verify = NONE
 pg_ro_max_concurrent_queries = NONE
 pg_ro_semaphore_timeout = NONE
+pg_ro_connection_keepalive_timeout = NONE
+pg_ro_connection_pool_size = NONE
+pg_ro_connection_backlog = NONE
 
 cassandra_contact_points = 127.0.0.1
 cassandra_port = 9042

spec/01-unit/03-conf_loader_spec.lua

@@ -1388,6 +1388,106 @@ describe("Configuration loader", function()
     end)
   end)
 
+  describe("pg connection pool options", function()
+    it("rejects a pg_connection_keepalive_timeout with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_keepalive_timeout = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_keepalive_timeout must be greater than 0", err)
+    end)
+
+    it("rejects a pg_connection_keepalive_timeout with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_keepalive_timeout = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_keepalive_timeout must be an integer greater than 0", err)
+    end)
+
+    it("rejects a pg_connection_pool_size with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_pool_size = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_pool_size must be greater than 0", err)
+    end)
+
+    it("rejects a pg_connection_pool_size with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_pool_size = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_pool_size must be an integer greater than 0", err)
+    end)
+
+    it("rejects a pg_connection_backlog with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_backlog = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_backlog must be greater than 0", err)
+    end)
+
+    it("rejects a pg_connection_backlog with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_connection_backlog = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_connection_backlog must be an integer greater than 0", err)
+    end)
+  end)
+
+  describe("pg read-only connection pool options", function()
+    it("rejects a pg_ro_connection_keepalive_timeout with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_keepalive_timeout = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_keepalive_timeout must be greater than 0", err)
+    end)
+
+    it("rejects a pg_ro_connection_keepalive_timeout with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_keepalive_timeout = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_keepalive_timeout must be an integer greater than 0", err)
+    end)
+
+    it("rejects a pg_ro_connection_pool_size with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_pool_size = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_pool_size must be greater than 0", err)
+    end)
+
+    it("rejects a pg_ro_connection_pool_size with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_pool_size = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_pool_size must be an integer greater than 0", err)
+    end)
+
+    it("rejects a pg_ro_connection_backlog with a negative number", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_backlog = -1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_backlog must be greater than 0", err)
+    end)
+
+    it("rejects a pg_ro_connection_backlog with a decimal", function()
+      local conf, err = conf_loader(nil, {
+        pg_ro_connection_backlog = 0.1,
+      })
+      assert.is_nil(conf)
+      assert.equal("pg_ro_connection_backlog must be an integer greater than 0", err)
+    end)
+  end)
+
   describe("worker_state_update_frequency option", function()
     it("is rejected with a zero", function()
       local conf, err = conf_loader(nil, {