Log empty or nil select calls

Empty or nil select calls on user spaces tend to be dangerous.

A critical log entry containing the current stack traceback is created
upon empty or nil select calls — a user can explicitly request a full
scan though by passing fullscan = true to select's options table
argument in which a case a log entry will not be created.

Tarantool has a similar implementation[1].

1. tarantool/tarantool#7064

Closes #276

Author: oleg-jukovec

CHANGELOG.md

@@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 ### Changed
+* Behaviour of empty or nil select calls: a critical log entry containing the
+  current stack traceback is created upon such function calls — a user can
+  explicitly request a full scan though by passing fullscan=true to select's
+  options table argument in which a case a log entry will not be created (#276).
 
 ### Fixed
 

crud/ratelimit.lua

@@ -0,0 +1,102 @@
+-- Mostly it's a copy-paste from tarantool/src/lua/log.lua. We have three
+-- reasons:
+-- 1. Tarantool has not log.crit() (a function for logging with CRIT level).
+-- 2. Only new versions of Tarantool have Ratelimit type.
+-- 3. We want own copy of Ratelimit in case the implementation in Tarantool
+-- changes. Less pain between Tarantool versions.
+local ffi = require('ffi')
+
+local S_CRIT = ffi.C.S_CRIT
+local S_WARN = ffi.C.S_WARN
+
+local function say(level, fmt, ...)
+    if ffi.C.log_level < level then
+        -- don't waste cycles on debug.getinfo()
+        return
+    end
+    local type_fmt = type(fmt)
+    local format = "%s"
+    if select('#', ...) ~= 0 then
+        local stat
+        stat, fmt = pcall(string.format, fmt, ...)
+        if not stat then
+            error(fmt, 3)
+        end
+    elseif type_fmt == 'table' then
+        -- An implementation in Tarantool supports encoding a table in JSON,
+        -- but it requires more dependencies from FFI. So we just deleted it
+		-- because we don't need such encoding in the module.
+        error("table not supported", 3)
+    elseif type_fmt ~= 'string' then
+        fmt = tostring(fmt)
+    end
+
+    local debug = require('debug')
+    local frame = debug.getinfo(3, "Sl")
+    local line, file = 0, 'eval'
+    if type(frame) == 'table' then
+        line = frame.currentline or 0
+        file = frame.short_src or frame.src or 'eval'
+    end
+
+    ffi.C._say(level, file, line, nil, format, fmt)
+end
+
+local Ratelimit = {
+    interval = 60,
+    burst = 10,
+    emitted = 0,
+    suppressed = 0,
+    start = 0,
+}
+
+function Ratelimit:new(object)
+    object = object or {}
+    setmetatable(object, self)
+    self.__index = self
+    return object
+end
+
+function Ratelimit:check()
+    local clock = require('clock')
+
+    local now = clock.monotonic()
+    local saved_suppressed = 0
+    if now > self.start + self.interval then
+        saved_suppressed = self.suppressed
+        self.suppressed = 0
+        self.emitted = 0
+        self.start = now
+    end
+
+    if self.emitted < self.burst then
+        self.emitted = self.emitted + 1
+        return saved_suppressed, true
+    end
+    self.suppressed = self.suppressed + 1
+    return saved_suppressed, false
+end
+
+function Ratelimit:log_check(lvl)
+    local suppressed, ok = self:check()
+    if lvl >= S_WARN and suppressed > 0 then
+        say(S_WARN, '%d messages suppressed due to rate limiting', suppressed)
+    end
+    return ok
+end
+
+function Ratelimit:log(lvl, fmt, ...)
+    if self:log_check(lvl) then
+        say(lvl, fmt, ...)
+    end
+end
+
+local function log_ratelimited_closure(lvl)
+    return function(self, fmt, ...)
+        self:log(lvl, fmt, ...)
+    end
+end
+
+Ratelimit.log_crit = log_ratelimited_closure(S_CRIT)
+
+return Ratelimit

crud/select/compat/common.lua

@@ -1,6 +1,20 @@
+local ratelimit = require('crud.ratelimit')
+local check_select_args_rl = ratelimit:new()
+
 local common = {}
 
 common.SELECT_FUNC_NAME = '_crud.select_on_storage'
 common.DEFAULT_BATCH_SIZE = 100
 
+common.check_select_args = function(space_name, user_conditions, opts)
+    local rl = check_select_args_rl
+    local uc = user_conditions
+
+    if (type(uc) == 'nil' or (type(uc) == 'table' and next(uc) == nil)) and
+       (opts == nil or not opts.fullscan) then
+        rl:log_crit("empty or nil `select` call on user space=%s\n %s",
+                    space_name, debug.traceback())
+    end
+end
+
 return common

crud/select/compat/select.lua

@@ -193,6 +193,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         fields = '?table',
+        fullscan = '?boolean',
 
         mode = '?vshard_call_mode',
         prefer_replica = '?boolean',
@@ -252,15 +253,18 @@ local function select_module_call_xc(space_name, user_conditions, opts)
     checks('string', '?table', {
         after = '?table|cdata',
         first = '?number',
-        timeout = '?number',
         batch_size = '?number',
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         fields = '?table',
+        fullscan = '?boolean',
+
+        mode = '?vshard_call_mode',
         prefer_replica = '?boolean',
         balance = '?boolean',
-        mode = '?vshard_call_mode',
+        timeout = '?number',
     })
+    common.check_select_args(space_name, user_conditions, opts)
 
     opts = opts or {}
 

crud/select/compat/select_old.lua

@@ -225,6 +225,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         fields = '?table',
+        fullscan = '?boolean',
 
         mode = '?vshard_call_mode',
         prefer_replica = '?boolean',
@@ -302,6 +303,7 @@ end
 
 local function select_module_call_xc(space_name, user_conditions, opts)
     dev_checks('string', '?table', '?table')
+    common.check_select_args(space_name, user_conditions, opts)
 
     opts = opts or {}
 
@@ -366,14 +368,16 @@ function select_module.call(space_name, user_conditions, opts)
     checks('string', '?table', {
         after = '?table',
         first = '?number',
-        timeout = '?number',
         batch_size = '?number',
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         fields = '?table',
+        fullscan = '?boolean',
+
+        mode = '?vshard_call_mode',
         prefer_replica = '?boolean',
         balance = '?boolean',
-        mode = '?vshard_call_mode',
+        timeout = '?number',
     })
 
     return sharding.wrap_method(select_module_call_xc, space_name, user_conditions, opts)

test/unit/select_nil_test.lua

@@ -0,0 +1,54 @@
+local t = require('luatest')
+local g = t.group('select_nil')
+local luatest_capture = require('luatest.capture')
+local crud = require('crud')
+
+-- We throw a warning message when checking arguments for syntax correctness.
+-- It doesn't matter if a space exists or not (but a string argument must be)
+-- and if a select will be executed or not. We will not care about errors too
+-- because for that we have other tests.
+local any_space = 'non_existent_space'
+local expected_log = "empty or nil `select` call on user space=" .. any_space
+
+g.test_warnings = function()
+    local opts = {
+        {uc = nil, opts = nil},
+        {uc = nil, opts = {}},
+        {uc = nil, opts = {fullscan = false}},
+        {uc = {}, opts = nil},
+        {uc = {}, opts = {}},
+        {uc = {}, opts = {fullscan = false}},
+    }
+
+    for _, o in ipairs(opts) do
+        local capture = luatest_capture:new()
+        capture:wrap(true, function()
+            pcall(crud.select, any_space, o.uc, o.opts)
+        end)
+
+        local captured = capture:flush()
+	    t.assert_str_contains(captured.stderr, expected_log)
+    end
+end
+
+g.test_no_warnings = function()
+    local opts = {
+        {uc = nil, opts = {fullscan = true}},
+        {uc = {}, opts = {fullscan = true}},
+        {uc = {0}, opts = nil},
+        {uc = {0}, opts = {}},
+        {uc = {0}, opts = {fullscan = false}},
+    }
+
+    for _, o in ipairs(opts) do
+        local capture = luatest_capture:new()
+
+        capture:wrap(true, function()
+            pcall(crud.select, any_space, o.uc, o.opts)
+        end)
+
+        local captured = capture:flush()
+        t.assert_equals(captured.stderr, "")
+    end
+end
+