refactor(db): change postgres ttl cleanup timer to delete rows based …

…on database server-side timestamp
Kong · Mar 28, 2023 · 4eca3a3 · 4eca3a3
1 parent e8546bd
commit 4eca3a3
Showing 1 changed file with 19 additions and 7 deletions.
diff --git a/kong/db/strategies/postgres/connector.lua b/kong/db/strategies/postgres/connector.lua
@@ -332,24 +332,36 @@ function _mt:init_worker(strategies)
     WITH rows AS (
   SELECT ctid
     FROM %s
-   WHERE %s < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
+   WHERE %s < TO_TIMESTAMP(%s) AT TIME ZONE 'UTC'
 ORDER BY %s LIMIT 50000 FOR UPDATE SKIP LOCKED)
   DELETE
     FROM %s
-   WHERE ctid IN (TABLE rows);]], table_name_escaped, column_name, column_name, table_name_escaped):gsub("CURRENT_TIMESTAMP", "TO_TIMESTAMP(%%s)")
+   WHERE ctid IN (TABLE rows);]], table_name_escaped, column_name, "%s", column_name, table_name_escaped)
     end
 
     return timer_every(self.config.ttl_cleanup_interval, function(premature)
       if premature then
         return
       end
 
+      -- Fetch the end timestamp from database to avoid problems caused by the difference
+      -- between nodes and database time.
+      local cleanup_end_timestamp
+      local ok, err = self:query("SELECT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP AT TIME ZONE 'UTC') AS NOW;")
+      if not ok then
+        log(WARN, "unable to fetch current timestamp from PostgreSQL database (",
+                  err, ")")
+        return
+      end
+
+      cleanup_end_timestamp = ok[1]["now"]
+
       for i, statement in ipairs(cleanup_statements) do
-        local cleanup_start_time = self:escape_literal(tonumber(fmt("%.3f", now_updated())))
+        local _tracing_cleanup_start_time = now()
 
         while true do -- batch delete looping
-          -- avoid using CURRENT_TIMESTAMP in the real query to prevent infinite loop
-          local ok, err = self:query(fmt(statement, cleanup_start_time))
+          -- using the server-side timestamp in the whole loop to prevent infinite loop
+          local ok, err = self:query(fmt(statement, cleanup_end_timestamp))
           if not ok then
             if err then
               log(WARN, "unable to clean expired rows from table '",
@@ -368,8 +380,8 @@ ORDER BY %s LIMIT 50000 FOR UPDATE SKIP LOCKED)
           end
         end
 
-        local cleanup_end_time = now_updated()
-        local time_elapsed = tonumber(fmt("%.3f", cleanup_end_time - cleanup_start_time))
+        local _tracing_cleanup_end_time = now()
+        local time_elapsed = tonumber(fmt("%.3f", _tracing_cleanup_end_time - _tracing_cleanup_start_time))
         log(DEBUG, "cleaning up expired rows from table '", table_names[i],
                    "' took ", time_elapsed, " seconds")
       end