fix(db/declarative): fix TTL not working in DB-less and Hybrid mode (#…

…11464)

* fix(declarative): fix TTL not working in DB-less and Hybrid mode

* Previously, in DB-less and Hybrid mode, the ttl/updated_at fields were
  not copied from the original entities to the flattened entities. As a
  result, the entities were loaded without the TTL field.
* Additionally, for loading the TTL field, the "off" DB strategy
  (lmdb) did not properly filter expired items, nor returned right
  TTL value for DAO.

FTI-4512

* fix coding style
* fix coding style: improved function name
* added test case: hybrid mode for key-auth
* fix test case warnings
* fixed test case consumer domain
* export ttl as absolute value
* delete unused defination
* move ttl-fixing logic into row_to_entity()
* still use pg to caculate relative value
* clean code
* add changelog entry
* fixed test cases
* fixed test cases warning
* fixed test failure
* fix test case issue: ttl expiration
* fix test case: unsed local variable
* add an entry in CHANGELOG.md
* fix changelog scope
* remove release-related information in CHANGELOG.md
* fix test case: sleep before attempting unnecessary requests
* sleep before attempting unnecessary requests
* decrease the ttl to expedite the case's execution
* fix CHANGELOG typo
* fix the tense problem of changelog entry
* add export options for "page_*_for_export" sql statement
* fix warning: setting non-standard global variable
* fix error reporting: options is nil
* fix an issue where the off strategy returned the expired entity
* run ttl processing before schema:process_auto_fields()

changelog/unreleased/kong/11464.yml

@@ -0,0 +1,7 @@
+message: Fix an issue that the TTL of the key-auth plugin didnt work in DB-less and Hybrid mode.
+type: bugfix
+scope: Core
+prs:
+  - 11464
+jiras:
+  - "FTI-4512"

kong/db/dao/init.lua

@@ -283,6 +283,12 @@ local function validate_options_value(self, options)
     end
   end
 
+  if options.export ~= nil then
+    if type(options.export) ~= "boolean" then
+      errors.export = "must be a boolean"
+    end
+  end
+
   if next(errors) then
     return nil, errors
   end
@@ -1103,6 +1109,21 @@ function DAO:each(size, options)
 end
 
 
+function DAO:each_for_export(size, options)
+  if self.strategy.schema.ttl then
+    if not options then
+      options = get_pagination_options(self, options)
+    else
+      options = utils.cycle_aware_deep_copy(options, true)
+    end
+
+    options.export = true
+  end
+
+  return self:each(size, options)
+end
+
+
 function DAO:insert(entity, options)
   validate_entity_type(entity)
 

kong/db/declarative/export.lua

@@ -142,7 +142,7 @@ local function export_from_db_impl(emitter, skip_ws, skip_disabled_entities, exp
     if db[name].pagination then
       page_size = db[name].pagination.max_page_size
     end
-    for row, err in db[name]:each(page_size, GLOBAL_QUERY_OPTS) do
+    for row, err in db[name]:each_for_export(page_size, GLOBAL_QUERY_OPTS) do
       if not row then
         end_transaction(db)
         kong.log.err(err)

kong/db/schema/others/declarative_config.lua

@@ -830,6 +830,10 @@ local function flatten(self, input)
         end
       end
 
+      if schema.ttl and entry.ttl and entry.ttl ~= null then
+        flat_entry.ttl = entry.ttl
+      end
+
       entities[entity][id] = flat_entry
     end
   end

kong/db/strategies/off/init.lua

@@ -54,6 +54,19 @@ local function ws(schema, options)
 end
 
 
+local function process_ttl_field(entity)
+  if entity and entity.ttl and entity.ttl ~= null then
+    local ttl_value = entity.ttl - ngx.time()
+    if ttl_value > 0 then
+      entity.ttl = ttl_value
+    else
+      entity = nil  -- do not return the expired entity
+    end
+  end
+  return entity
+end
+
+
 -- Returns a dict of entity_ids tagged according to the given criteria.
 -- Currently only the following kinds of keys are supported:
 -- * A key like `services|<ws_id>|@list` will only return service keys
@@ -157,6 +170,7 @@ local function page_for_key(self, key, size, offset, options)
   yield()
 
   local ret = {}
+  local ret_idx = 1
   local schema = self.schema
   local schema_name = schema.name
 
@@ -194,7 +208,14 @@ local function page_for_key(self, key, size, offset, options)
       return nil, "stale data detected while paginating"
     end
 
-    ret[i - offset + 1] = schema:process_auto_fields(item, "select", true, PROCESS_AUTO_FIELDS_OPTS)
+    if schema.ttl then
+      item = process_ttl_field(item)
+    end
+
+    if item then
+      ret[ret_idx] = schema:process_auto_fields(item, "select", true, PROCESS_AUTO_FIELDS_OPTS)
+      ret_idx = ret_idx + 1
+    end
   end
 
   if offset then
@@ -211,6 +232,13 @@ local function select_by_key(schema, key)
     return nil, err
   end
 
+  if schema.ttl then
+    entity = process_ttl_field(entity)
+    if not entity then
+      return nil
+    end
+  end
+
   entity = schema:process_auto_fields(entity, "select", true, PROCESS_AUTO_FIELDS_OPTS)
 
   return entity

kong/db/strategies/postgres/init.lua

@@ -481,6 +481,10 @@ local function page(self, size, token, foreign_key, foreign_entity_name, options
     statement_name = "page" .. suffix
   end
 
+  if options and options.export then
+    statement_name = statement_name .. "_for_export"
+  end
+
   if token then
     local token_decoded = decode_base64(token)
     if not token_decoded then
@@ -1022,6 +1026,7 @@ function _M.new(connector, schema, errors)
     ws_id_select_where = "(" .. ws_id_escaped .. " = $0)"
   end
 
+  local select_for_export_expressions
   local ttl_select_where
   if has_ttl then
     fields_hash.ttl = { timestamp = true }
@@ -1030,6 +1035,13 @@ function _M.new(connector, schema, errors)
     insert(insert_expressions, "$" .. #insert_names)
     insert(insert_columns, ttl_escaped)
 
+    select_for_export_expressions = concat {
+      select_expressions, ",",
+      "FLOOR(EXTRACT(EPOCH FROM (",
+        ttl_escaped, " AT TIME ZONE 'UTC'",
+      "))) AS ", ttl_escaped
+    }
+
     select_expressions = concat {
       select_expressions, ",",
       "FLOOR(EXTRACT(EPOCH FROM (",
@@ -1078,6 +1090,7 @@ function _M.new(connector, schema, errors)
   self.statements["truncate_global"] = self.statements["truncate"]
 
   local add_statement
+  local add_statement_for_export
   do
     local function add(name, opts, add_ws)
       local orig_argn = opts.argn
@@ -1106,6 +1119,14 @@ function _M.new(connector, schema, errors)
       add(name .. "_global", opts, false)
       add(name, opts, true)
     end
+
+    add_statement_for_export = function(name, opts)
+      add_statement(name, opts)
+      if has_ttl then
+        opts.code[2] = select_for_export_expressions
+        add_statement(name .. "_for_export", opts)
+      end
+    end
   end
 
   add_statement("insert", {
@@ -1181,7 +1202,7 @@ function _M.new(connector, schema, errors)
     }
   })
 
-  add_statement("page_first", {
+  add_statement_for_export("page_first", {
     operation = "read",
     argn = { LIMIT },
     argv = single_args,
@@ -1196,7 +1217,7 @@ function _M.new(connector, schema, errors)
     }
   })
 
-  add_statement("page_next", {
+  add_statement_for_export("page_next", {
     operation = "read",
     argn = page_next_names,
     argv = page_next_args,
@@ -1246,7 +1267,7 @@ function _M.new(connector, schema, errors)
 
       local statement_name = "page_for_" .. foreign_entity_name
 
-      add_statement(statement_name .. "_first", {
+      add_statement_for_export(statement_name .. "_first", {
         operation = "read",
         argn = argn_first,
         argv = argv_first,
@@ -1262,7 +1283,7 @@ function _M.new(connector, schema, errors)
         }
       })
 
-      add_statement(statement_name .. "_next", {
+      add_statement_for_export(statement_name .. "_next", {
         operation = "read",
         argn = argn_next,
         argv = argv_next,
@@ -1297,7 +1318,7 @@ function _M.new(connector, schema, errors)
 
     for cond, op in pairs({["_and"] = "@>", ["_or"] = "&&"}) do
 
-      add_statement("page_by_tags" .. cond .. "_first", {
+      add_statement_for_export("page_by_tags" .. cond .. "_first", {
         operation = "read",
         argn = argn_first,
         argv = {},
@@ -1313,7 +1334,7 @@ function _M.new(connector, schema, errors)
         },
       })
 
-      add_statement("page_by_tags" .. cond .. "_next", {
+      add_statement_for_export("page_by_tags" .. cond .. "_next", {
         operation = "read",
         argn = argn_next,
         argv = {},

spec/03-plugins/09-key-auth/04-hybrid_mode_spec.lua

@@ -0,0 +1,123 @@
+local helpers = require "spec.helpers"
+
+for _, strategy in helpers.each_strategy({"postgres"}) do
+  describe("Plugin: key-auth (access) [#" .. strategy .. "] auto-expiring keys", function()
+    -- Give a bit of time to reduce test flakyness on slow setups
+    local ttl = 10
+    local inserted_at
+    local proxy_client
+
+    lazy_setup(function()
+      local bp = helpers.get_db_utils(strategy, {
+        "routes",
+        "services",
+        "plugins",
+        "consumers",
+        "keyauth_credentials",
+      })
+
+      local r = bp.routes:insert {
+        hosts = { "key-ttl-hybrid.com" },
+      }
+
+      bp.plugins:insert {
+        name = "key-auth",
+        route = { id = r.id },
+      }
+
+      bp.consumers:insert {
+        username = "Jafar",
+      }
+
+      assert(helpers.start_kong({
+        role = "control_plane",
+        database = strategy,
+        cluster_cert = "spec/fixtures/kong_clustering.crt",
+        cluster_cert_key = "spec/fixtures/kong_clustering.key",
+        lua_ssl_trusted_certificate = "spec/fixtures/kong_clustering.crt",
+        cluster_listen = "127.0.0.1:9005",
+        cluster_telemetry_listen = "127.0.0.1:9006",
+        nginx_conf = "spec/fixtures/custom_nginx.template",
+      }))
+
+      assert(helpers.start_kong({
+        role = "data_plane",
+        database = "off",
+        prefix = "servroot2",
+        cluster_cert = "spec/fixtures/kong_clustering.crt",
+        cluster_cert_key = "spec/fixtures/kong_clustering.key",
+        lua_ssl_trusted_certificate = "spec/fixtures/kong_clustering.crt",
+        cluster_control_plane = "127.0.0.1:9005",
+        cluster_telemetry_endpoint = "127.0.0.1:9006",
+        proxy_listen = "0.0.0.0:9002",
+      }))
+    end)
+
+    lazy_teardown(function()
+      if proxy_client then
+        proxy_client:close()
+      end
+
+      helpers.stop_kong("servroot2")
+      helpers.stop_kong()
+    end)
+
+    it("authenticate for up to 'ttl'", function()
+
+      -- add credentials after nginx has started to avoid TTL expiration
+      local admin_client = helpers.admin_client()
+      local res = assert(admin_client:send {
+          method  = "POST",
+          path  = "/consumers/Jafar/key-auth",
+          headers = {
+            ["Content-Type"] = "application/json",
+          },
+          body = {
+            key = "kong",
+            ttl = 10,
+          },
+      })
+      assert.res_status(201, res)
+      admin_client:close()
+
+      ngx.update_time()
+      inserted_at = ngx.now()
+
+      helpers.wait_until(function()
+        proxy_client = helpers.http_client("127.0.0.1", 9002)
+        res = assert(proxy_client:send {
+          method  = "GET",
+          path  = "/status/200",
+          headers = {
+            ["Host"] = "key-ttl-hybrid.com",
+            ["apikey"] = "kong",
+          }
+        })
+
+        proxy_client:close()
+        return res and res.status == 200
+      end, 5)
+
+      ngx.update_time()
+      local elapsed = ngx.now() - inserted_at
+
+      ngx.sleep(ttl - elapsed)
+
+      helpers.wait_until(function()
+        proxy_client = helpers.http_client("127.0.0.1", 9002)
+        res = assert(proxy_client:send {
+          method  = "GET",
+          path  = "/status/200",
+          headers = {
+            ["Host"] = "key-ttl-hybrid.com",
+            ["apikey"] = "kong",
+          }
+        })
+
+        proxy_client:close()
+        return res and res.status == 401
+      end, 5)
+
+    end)
+  end)
+end