diff --git a/CHANGELOG.md b/CHANGELOG.md index a660ed1f5..6d971a819 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/). - Set request timeout for OIDC query. [PR #1601](https://github.com/3scale/APIcast/pull/1601) [THREESCALE-8006](https://redhat.atlassian.net/browse/THREESCALE-8006) - Set upstream metrics when sending request via proxy. [PR #1598](https://github.com/3scale/APIcast/pull/1598) [THREESCALE-10571](https://redhat.atlassian.net/browse/THREESCALE-15071) - Certificate is never verified in API request for https backend via proxy [PR #1573](https://github.com/3scale/APIcast/pull/1568) [THREESCALE-11944](https://redhat.atlassian.net/browse/THREESCALE-11944) +- 3scale Batcher Policy: read the batched report keys while holding the lock, so a concurrent flush cannot leave the lock orphaned and stall requests for the lock timeout [Issue #1610](https://github.com/3scale/APIcast/issues/1610) +- TimerTask: run the task inside a `pcall` so a failing run cannot stop the task from being scheduled again [Issue #1610](https://github.com/3scale/APIcast/issues/1610) ### Added - Update APIcast schema manifest [PR #1550](https://github.com/3scale/APIcast/pull/1550) diff --git a/gateway/src/apicast/policy/3scale_batcher/reports_batcher.lua b/gateway/src/apicast/policy/3scale_batcher/reports_batcher.lua index 2d9e0f1c6..28c198e3a 100644 --- a/gateway/src/apicast/policy/3scale_batcher/reports_batcher.lua +++ b/gateway/src/apicast/policy/3scale_batcher/reports_batcher.lua @@ -72,8 +72,6 @@ end function _M:get_all(service_id) local cached_reports = {} - local cached_report_keys = self.storage:get_keys() - local lock, new_lock_err = resty_lock:new(self.lock_shdict_name, lock_options) if not lock then ngx.log(ngx.ERR, 'failed to create lock: ', new_lock_err) @@ -86,17 +84,30 @@ function _M:get_all(service_id) return {} end + -- The keys need to be read while holding the lock. There is one reporting + -- timer per worker, so reading them before acquiring it allows another + -- worker to delete a key in between, and then storage:get() below returns + -- nil for it. + local cached_report_keys = self.storage:get_keys() + for _, key in ipairs(cached_report_keys) do local value = self.storage:get(key) - local report, err = keys_helper.report_from_key_batched_report(key, value) - if not err then - if value > 0 and report and report.service_id == service_id then - insert(cached_reports, report) - self.storage:delete(key) + -- Deletes only happen while holding the lock, but the entry can still be + -- gone, for example if it was evicted. Skip it rather than raising, an + -- error here would leave the lock to expire on its own and would prevent + -- the caller from scheduling the next run. + if value then + local report, err = keys_helper.report_from_key_batched_report(key, value) + + if not err then + if value > 0 and report and report.service_id == service_id then + insert(cached_reports, report) + self.storage:delete(key) + end + else + ngx.log(ngx.WARN, 'failed to get report for key ', key, ' err: ', err) end - else - ngx.log(ngx.WARN, 'failed to get report for key ', key, ' err: ', err) end end diff --git a/gateway/src/resty/concurrent/timer_task.lua b/gateway/src/resty/concurrent/timer_task.lua index fd91a943b..21596b991 100644 --- a/gateway/src/resty/concurrent/timer_task.lua +++ b/gateway/src/resty/concurrent/timer_task.lua @@ -2,6 +2,7 @@ local GC = require 'apicast.gc' local uuid = require 'resty.jit-uuid' local unpack = unpack +local pcall = pcall local _M = {} @@ -68,7 +69,15 @@ run_periodic = function(run_now, id, func, args, interval) end if run_now then - func(unpack(args)) + -- The task is run inside a pcall so that a failing run cannot prevent the + -- next one from being scheduled. Letting the error propagate aborts the + -- timer thread before schedule_next() below, which stops the task from + -- ever running again on this worker. + local ok, err = pcall(func, unpack(args)) + + if not ok then + ngx.log(ngx.ERR, 'timer task failed to run: ', err) + end end if not _M.last_one[id] then diff --git a/spec/policy/3scale_batcher/reports_batcher_spec.lua b/spec/policy/3scale_batcher/reports_batcher_spec.lua index 0b074c210..aa5dda352 100644 --- a/spec/policy/3scale_batcher/reports_batcher_spec.lua +++ b/spec/policy/3scale_batcher/reports_batcher_spec.lua @@ -5,11 +5,16 @@ local Transaction = require 'apicast.policy.3scale_batcher.transaction' local resty_lock = require 'resty.lock' describe('reports batcher', function() + -- Records lock/unlock calls so the tests that care about them can assert on + -- the order. Not relevant for most of these tests. + local events + before_each(function() - -- Stub locks. Not relevant for these tests. + events = {} + stub(resty_lock, 'new').returns({ - lock = function() return true end, - unlock = function() return true end + lock = function() table.insert(events, 'lock') return true end, + unlock = function() table.insert(events, 'unlock') return true end }) end) @@ -93,5 +98,55 @@ describe('reports batcher', function() assert.same({}, reports_batcher:get_all('1')) end) end) + + it('reads the keys while holding the lock', function() + -- There is one reporting timer per worker. Listing the keys before + -- taking the lock lets another worker delete them in between. + local test_dict = { + get_keys = function() table.insert(events, 'get_keys') return {} end, + get = function() end, + delete = function() end + } + + ReportsBatcher.new(test_dict, 'batched_reports_locks'):get_all('1') + + assert.same({ 'lock', 'get_keys', 'unlock' }, events) + end) + + describe('when a key is gone after it has been listed', function() + local service_id = '1' + local credentials = { user_key = 'uk' } + + local missing_key = keys_helper.key_for_batched_report( + service_id, credentials, 'missing') + local existing_key = keys_helper.key_for_batched_report( + service_id, credentials, 'existing') + + local function build_dict() + local storage = { [existing_key] = 1 } + + return { + get_keys = function() return { missing_key, existing_key } end, + get = function(_, key) return storage[key] end, + delete = function(_, key) storage[key] = nil end + } + end + + it('skips it and returns the remaining reports', function() + local reports_batcher = ReportsBatcher.new( + build_dict(), 'batched_reports_locks') + + local reports = reports_batcher:get_all(service_id) + + assert.equals(1, #reports) + assert.equals('existing', reports[1].metric) + end) + + it('releases the lock', function() + ReportsBatcher.new(build_dict(), 'batched_reports_locks'):get_all(service_id) + + assert.same({ 'lock', 'unlock' }, events) + end) + end) end) end) diff --git a/spec/resty/concurrent/timer_task_spec.lua b/spec/resty/concurrent/timer_task_spec.lua index 65f140ac3..b5bfaa6b0 100644 --- a/spec/resty/concurrent/timer_task_spec.lua +++ b/spec/resty/concurrent/timer_task_spec.lua @@ -132,6 +132,24 @@ describe('TimerTask', function() end) end) + describe('when the task raises an error', function() + local failing_task = function() error('something went wrong') end + + it('does not propagate the error', function() + local timer_task = TimerTask.new(failing_task, { args = args, interval = interval }) + + assert.has_no.errors(function() timer_task:execute(true) end) + end) + + it('schedules the next one', function() + local timer_task = TimerTask.new(failing_task, { args = args, interval = interval }) + + timer_task:execute(true) + + assert.stub(ngx_timer_stub).was_called() + end) + end) + describe('when the task is not active', function() it('does not run the task', function() local timer_task = TimerTask.new(func, { args = args, interval = interval })