[patches] [daemon 1/1] daemon: Introduce feedback daemon
Georgy Kirichenko
georgy at tarantool.org
Wed Mar 7 20:12:49 MSK 2018
Please rebase the commit against 1.9
On Wednesday, March 7, 2018 11:24:08 AM MSK Ilya Markov wrote:
> From: Roman Proskin <opomuc at gmail.com>
>
> * feedback daemon sends information about instance to the
> specified host.
> * Add new options to box.cfg:
> - feedback_enabled - switch on/off daemon, default=true.
> - feedback_host - host to which feedbacks are sent,
> default="https://feedback.tarantool.io".
> - feedback_interval - time interval in seconds of feedbacks
> sending, default=3600.
> * Add possibility to generate feedback file in json format with
> function box.feedback.save
>
> Relates #2762
>
> branch gh-2762-feedback-daemon
> ---
> src/box/CMakeLists.txt | 1 +
> src/box/lua/feedback_daemon.lua | 140
> ++++++++++++++++++++++++++++++++++ src/box/lua/init.c |
> 2 +
> src/box/lua/load_cfg.lua | 9 +++
> src/box/lua/schema.lua | 14 ++++
> src/errinj.h | 1 +
> src/httpc.c | 3 +-
> test/app-tap/init_script.result | 77 ++++++++++---------
> test/box-tap/feedback_daemon.test.lua | 104 +++++++++++++++++++++++++
> test/box/admin.result | 6 ++
> test/box/cfg.result | 12 +++
> test/box/errinj.result | 2 +
> test/box/misc.result | 1 +
> test/engine/iterator.result | 2 +-
> test/engine/savepoint.result | 12 +--
> 15 files changed, 341 insertions(+), 45 deletions(-)
> create mode 100644 src/box/lua/feedback_daemon.lua
> create mode 100755 test/box-tap/feedback_daemon.test.lua
>
> diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
> index ad7f910..fb6602b 100644
> --- a/src/box/CMakeLists.txt
> +++ b/src/box/CMakeLists.txt
> @@ -6,6 +6,7 @@ lua_source(lua_sources lua/schema.lua)
> lua_source(lua_sources lua/tuple.lua)
> lua_source(lua_sources lua/session.lua)
> lua_source(lua_sources lua/checkpoint_daemon.lua)
> +lua_source(lua_sources lua/feedback_daemon.lua)
> lua_source(lua_sources lua/net_box.lua)
> lua_source(lua_sources lua/upgrade.lua)
> lua_source(lua_sources lua/console.lua)
> diff --git a/src/box/lua/feedback_daemon.lua
> b/src/box/lua/feedback_daemon.lua new file mode 100644
> index 0000000..0663951
> --- /dev/null
> +++ b/src/box/lua/feedback_daemon.lua
> @@ -0,0 +1,140 @@
> +-- feedback_daemon.lua (internal file)
> +--
> +local log = require('log')
> +local json = require('json')
> +local fiber = require('fiber')
> +local http = require('http.client')
> +
> +local PREFIX = "feedback_daemon"
> +
> +local daemon = {
> + enabled = false,
> + interval = 0,
> + host = nil,
> + fiber = nil,
> + control = nil,
> + guard = nil,
> + shutdown = nil
> +}
> +
> +local function get_fiber_id(f)
> + local fid = 0
> + if f ~= nil and f:status() ~= "dead" then
> + fid = f:id()
> + end
> + return fid
> +end
> +
> +local function fill_in_feedback(feedback)
> + if box.info.status ~= "running" then
> + return nil, "not running"
> + end
> + feedback.tarantool_version = box.info.version
> + feedback.server_id = box.info.uuid
> + feedback.cluster_id = box.info.cluster.uuid
> + return feedback
> +end
> +
> +local function send(self)
> + fiber.name(PREFIX, { truncate = true })
> + local header = { feedback_type = "version", feedback_version = 1 }
> +
> + while true do
> + local feedback = fill_in_feedback(header)
> + local msg = self.control:get(self.interval)
> + if msg == "shutdown" then
> + break
> + elseif feedback ~= nil then
> + pcall(http.post, self.host, json.encode(feedback), {timeout=1})
> + end
> + end
> + self.shutdown:put("shutdown")
> +end
> +
> +local function guard_loop(daemon)
> + fiber.name(string.format("guard of %s", PREFIX), {truncate=true})
> +
> + while true do
> +
> + if get_fiber_id(daemon.fiber) == 0 then
> + daemon.fiber = fiber.create(send, daemon)
> + log.verbose("%s restarted", PREFIX)
> + end
> + local st, err = pcall(fiber.sleep, daemon.interval)
> + if not st then
> + daemon.shutdown:put("shutdown")
> + end
> + end
> +end
> +
> +-- these functions are used for test purposes only
> +local function start(self)
> + self:stop()
> + if self.enabled then
> + self.control = fiber.channel()
> + self.shutdown = fiber.channel()
> + self.guard = fiber.create(guard_loop, self)
> + end
> + log.verbose("%s started", PREFIX)
> +end
> +
> +local function stop(self)
> + if (get_fiber_id(self.guard) ~= 0) then
> + self.guard:cancel()
> + --
> + -- this workaround with channels may be replaced with fiber_join
> + -- so guard fiber may be set joinable
> + --
> + self.shutdown:get()
> + self.guard = nil
> + end
> + if (get_fiber_id(self.fiber) ~= 0) then
> + self.control:put("shutdown")
> + self.shutdown:get()
> + self.fiber = nil
> + self.control = nil
> + self.shutdown = nil
> + end
> + log.verbose("%s stopped", PREFIX)
> +end
> +
> +local function reload(self)
> + self:stop()
> + self:start()
> +end
> +
> +setmetatable(daemon, {
> + __index = {
> + set_feedback_params = function()
> + daemon.enabled = box.cfg.feedback_enabled
> + daemon.host = box.cfg.feedback_host
> + daemon.interval = box.cfg.feedback_interval
> + reload(daemon)
> + return
> + end,
> + -- this function is used in saving feedback in file
> + generate_feedback = function()
> + return fill_in_feedback({ feedback_type = "version",
> feedback_version = 1 }) + end,
> + start = function()
> + start(daemon)
> + end,
> + stop = function()
> + stop(daemon)
> + end,
> + reload = function()
> + reload(daemon)
> + end,
> + send_test = function()
> + if daemon.control ~= nil then
> + daemon.control:put("send")
> + end
> + end
> + }
> +})
> +
> +if box.internal == nil then
> + box.internal = { [PREFIX] = daemon }
> +else
> + box.internal[PREFIX] = daemon
> +end
> diff --git a/src/box/lua/init.c b/src/box/lua/init.c
> index 7547758..d4b5788 100644
> --- a/src/box/lua/init.c
> +++ b/src/box/lua/init.c
> @@ -65,6 +65,7 @@ extern char session_lua[],
> load_cfg_lua[],
> xlog_lua[],
> checkpoint_daemon_lua[],
> + feedback_daemon_lua[],
> net_box_lua[],
> upgrade_lua[],
> console_lua[];
> @@ -74,6 +75,7 @@ static const char *lua_sources[] = {
> "box/tuple", tuple_lua,
> "box/schema", schema_lua,
> "box/checkpoint_daemon", checkpoint_daemon_lua,
> + "box/feedback_daemon", feedback_daemon_lua,
> "box/upgrade", upgrade_lua,
> "box/net_box", net_box_lua,
> "box/console", console_lua,
> diff --git a/src/box/lua/load_cfg.lua b/src/box/lua/load_cfg.lua
> index d4f2128..89fd774 100644
> --- a/src/box/lua/load_cfg.lua
> +++ b/src/box/lua/load_cfg.lua
> @@ -59,6 +59,9 @@ local default_cfg = {
> replication_sync_lag = 10,
> replication_connect_timeout = 4,
> replication_connect_quorum = nil, -- connect all
> + feedback_enabled = true,
> + feedback_host = "https://feedback.tarantool.io",
> + feedback_interval = 3600,
> }
>
> -- types of available options
> @@ -115,6 +118,9 @@ local template_cfg = {
> replication_sync_lag = 'number',
> replication_connect_timeout = 'number',
> replication_connect_quorum = 'number',
> + feedback_enabled = 'boolean',
> + feedback_host = 'string',
> + feedback_interval = 'number',
> }
>
> local function normalize_uri(port)
> @@ -175,6 +181,9 @@ local dynamic_cfg = {
> checkpoint_count = private.cfg_set_checkpoint_count,
> checkpoint_interval =
> private.checkpoint_daemon.set_checkpoint_interval, worker_pool_threads
> = private.cfg_set_worker_pool_threads, + feedback_enabled =
> private.feedback_daemon.set_feedback_params, + feedback_host =
> private.feedback_daemon.set_feedback_params, + feedback_interval =
> private.feedback_daemon.set_feedback_params, -- do nothing, affects new
> replicas, which query this value on start wal_dir_rescan_delay =
> function() end,
> custom_proc_title = function()
> diff --git a/src/box/lua/schema.lua b/src/box/lua/schema.lua
> index 30c6bc6..8d2f11d 100644
> --- a/src/box/lua/schema.lua
> +++ b/src/box/lua/schema.lua
> @@ -5,6 +5,8 @@ local msgpack = require('msgpack')
> local msgpackffi = require('msgpackffi')
> local fun = require('fun')
> local log = require('log')
> +local fio = require('fio')
> +local json = require('json')
> local session = box.session
> local internal = require('box.internal')
> local function setmap(table)
> @@ -2234,4 +2236,16 @@ box.internal.schema.init = function()
> box_sequence_init()
> end
>
> +box.feedback = {}
> +box.feedback.save = function(file_name)
> + local feedback =
> json.encode(box.internal.feedback_daemon.generate_feedback()) + local
> fh, err = fio.open(file_name, {'O_CREAT', 'O_RDWR', 'O_TRUNC'}, +
> tonumber('0777', 8))
> + if not fh then
> + error(err)
> + end
> + fh:write(feedback)
> + fh:close()
> +end
> +
> box.NULL = msgpack.NULL
> diff --git a/src/errinj.h b/src/errinj.h
> index 352a3c3..20383e0 100644
> --- a/src/errinj.h
> +++ b/src/errinj.h
> @@ -107,6 +107,7 @@ struct errinj {
> _(ERRINJ_RELAY_EXIT_DELAY, ERRINJ_DOUBLE, {.dparam = 0}) \
> _(ERRINJ_VY_DELAY_PK_LOOKUP, ERRINJ_BOOL, {.bparam = false}) \
> _(ERRINJ_VY_RUN_WRITE_STMT_TIMEOUT, ERRINJ_DOUBLE, {.dparam = 0}) \
> + _(ERRINJ_HTTPC_EXECUTE, ERRINJ_BOOL, {.bparam = false}) \
>
> ENUM0(errinj_id, ERRINJ_LIST);
> extern struct errinj errinjs[];
> diff --git a/src/httpc.c b/src/httpc.c
> index cce7e51..633e688 100644
> --- a/src/httpc.c
> +++ b/src/httpc.c
> @@ -35,6 +35,7 @@
> #include <curl/curl.h>
>
> #include "fiber.h"
> +#include "errinj.h"
>
> /**
> * libcurl callback for CURLOPT_WRITEFUNCTION
> @@ -318,7 +319,7 @@ httpc_execute(struct httpc_request *req, double timeout)
>
> if (curl_execute(&req->curl_request, &env->curl_env, timeout) !=
CURLM_OK)
> return -1;
> -
> + ERROR_INJECT_RETURN(ERRINJ_HTTPC_EXECUTE);
> long longval = 0;
> switch (req->curl_request.code) {
> case CURLE_OK:
> diff --git a/test/app-tap/init_script.result
> b/test/app-tap/init_script.result index 80153e3..ae01b48 100644
> --- a/test/app-tap/init_script.result
> +++ b/test/app-tap/init_script.result
> @@ -7,43 +7,46 @@ box.cfg
> 2 checkpoint_count:2
> 3 checkpoint_interval:3600
> 4 coredump:false
> -5 force_recovery:false
> -6 hot_standby:false
> -7 listen:port
> -8 log:tarantool.log
> -9 log_format:plain
> -10 log_level:5
> -11 log_nonblock:true
> -12 memtx_dir:.
> -13 memtx_max_tuple_size:1048576
> -14 memtx_memory:107374182
> -15 memtx_min_tuple_size:16
> -16 pid_file:box.pid
> -17 read_only:false
> -18 readahead:16320
> -19 replication_connect_timeout:4
> -20 replication_sync_lag:10
> -21 replication_timeout:1
> -22 rows_per_wal:500000
> -23 slab_alloc_factor:1.05
> -24 too_long_threshold:0.5
> -25 vinyl_bloom_fpr:0.05
> -26 vinyl_cache:134217728
> -27 vinyl_dir:.
> -28 vinyl_max_tuple_size:1048576
> -29 vinyl_memory:134217728
> -30 vinyl_page_size:8192
> -31 vinyl_range_size:1073741824
> -32 vinyl_read_threads:1
> -33 vinyl_run_count_per_level:2
> -34 vinyl_run_size_ratio:3.5
> -35 vinyl_timeout:60
> -36 vinyl_write_threads:2
> -37 wal_dir:.
> -38 wal_dir_rescan_delay:2
> -39 wal_max_size:268435456
> -40 wal_mode:write
> -41 worker_pool_threads:4
> +5 feedback_enabled:true
> +6 feedback_host:https://feedback.tarantool.io
> +7 feedback_interval:3600
> +8 force_recovery:false
> +9 hot_standby:false
> +10 listen:port
> +11 log:tarantool.log
> +12 log_format:plain
> +13 log_level:5
> +14 log_nonblock:true
> +15 memtx_dir:.
> +16 memtx_max_tuple_size:1048576
> +17 memtx_memory:107374182
> +18 memtx_min_tuple_size:16
> +19 pid_file:box.pid
> +20 read_only:false
> +21 readahead:16320
> +22 replication_connect_timeout:4
> +23 replication_sync_lag:10
> +24 replication_timeout:1
> +25 rows_per_wal:500000
> +26 slab_alloc_factor:1.05
> +27 too_long_threshold:0.5
> +28 vinyl_bloom_fpr:0.05
> +29 vinyl_cache:134217728
> +30 vinyl_dir:.
> +31 vinyl_max_tuple_size:1048576
> +32 vinyl_memory:134217728
> +33 vinyl_page_size:8192
> +34 vinyl_range_size:1073741824
> +35 vinyl_read_threads:1
> +36 vinyl_run_count_per_level:2
> +37 vinyl_run_size_ratio:3.5
> +38 vinyl_timeout:60
> +39 vinyl_write_threads:2
> +40 wal_dir:.
> +41 wal_dir_rescan_delay:2
> +42 wal_max_size:268435456
> +43 wal_mode:write
> +44 worker_pool_threads:4
> --
> -- Test insert from detached fiber
> --
> diff --git a/test/box-tap/feedback_daemon.test.lua
> b/test/box-tap/feedback_daemon.test.lua new file mode 100755
> index 0000000..84ab55a
> --- /dev/null
> +++ b/test/box-tap/feedback_daemon.test.lua
> @@ -0,0 +1,104 @@
> +#!/usr/bin/env tarantool
> +
> +-- Testing feedback module
> +
> +local tap = require('tap')
> +local json = require('json')
> +local fiber = require('fiber')
> +local test = tap.test('feedback_daemon')
> +local socket = require('socket')
> +
> +test:plan(10)
> +
> +box.cfg{log = 'report.log', log_level = 6}
> +
> +local function http_handle(s)
> + s:write("HTTP/1.1 200 OK\r\n")
> + s:write("Accept: */*\r\n")
> + s:write("Connection: keep-alive\r\n")
> + s:write("Content-Type: application/json\r\n")
> + s:write("Access-Control-Allow-Origin: *\r\n")
> + s:write("Access-Control-Allow-Credentials: true\r\n")
> + s:write("Content-Length: 11\r\n\r\n")
> + s:write('{"key": 12}')
> +
> + local buf = s:read('\n')
> + while not buf:match("version") do
> + buf = s:read('\n')
> + end
> + local _, index = buf:find(".*}")
> + buf = buf:sub(1, index)
> + local ok, data = pcall(json.decode, buf)
> + if ok then
> + box.space._schema:put({'feedback', buf })
> + end
> +end
> +
> +local server = socket.tcp_server("127.0.0.1", 0, http_handle)
> +local port = server:name().port
> +
> +local interval = 0.01
> +box.cfg{
> + feedback_host = string.format("127.0.0.1:%i", port),
> + feedback_interval = interval,
> +}
> +
> +local function check(message)
> + while box.space._schema:get('feedback') == nil do fiber.sleep(0.001)
> end + local data = box.space._schema:get('feedback')
> + test:ok(data ~= nil, message)
> + box.space._schema:delete('feedback')
> +end
> +
> +check("initial check")
> +local daemon = box.internal.feedback_daemon
> +-- check if feedback has been sent and received
> +daemon.reload()
> +check("feedback received after reload")
> +
> +local errinj = box.error.injection
> +errinj.set("ERRINJ_HTTPC", true)
> +check('feedback received after errinj')
> +errinj.set("ERRINJ_HTTPC", false)
> +
> +daemon.send_test()
> +check("feedback received after explicit sending")
> +
> +box.cfg{feedback_enabled = false}
> +daemon.send_test()
> +while box.space._schema:get('feedback') ~= nil do fiber.sleep(0.001) end
> +test:ok(box.space._schema:get('feedback') == nil, "no feedback after
> disabling") +
> +box.cfg{feedback_enabled = true}
> +daemon.send_test()
> +check("feedback after start")
> +
> +daemon.stop()
> +daemon.send_test()
> +while box.space._schema:get('feedback') ~= nil do fiber.sleep(0.001) end
> +test:ok(box.space._schema:get('feedback') == nil, "no feedback after stop")
> +
> +daemon.start()
> +daemon.send_test()
> +check("feedback after start")
> +
> +box.feedback.save("feedback.json")
> +daemon.send_test()
> +while box.space._schema:get('feedback') == nil do fiber.sleep(0.001) end
> +local data = box.space._schema:get('feedback')
> +local fio = require("fio")
> +local fh = fio.open("feedback.json")
> +test:ok(fh, "file is created")
> +local file_data = fh:read()
> +test:is(file_data, data[2], "data is equal")
> +fh:close()
> +fio.unlink("feedback.json")
> +
> +server:close()
> +-- check it does not fail without server
> +local daemon = box.internal.feedback_daemon
> +daemon.start()
> +daemon.send_test()
> +
> +test:check()
> +os.exit(0)
> diff --git a/test/box/admin.result b/test/box/admin.result
> index 7a3e937..2969766 100644
> --- a/test/box/admin.result
> +++ b/test/box/admin.result
> @@ -26,6 +26,12 @@ cfg_filter(box.cfg)
> - 3600
> - - coredump
> - false
> + - - feedback_enabled
> + - true
> + - - feedback_host
> + - https://feedback.tarantool.io
> + - - feedback_interval
> + - 3600
> - - force_recovery
> - false
> - - hot_standby
> diff --git a/test/box/cfg.result b/test/box/cfg.result
> index 67539cd..717fa31 100644
> --- a/test/box/cfg.result
> +++ b/test/box/cfg.result
> @@ -22,6 +22,12 @@ cfg_filter(box.cfg)
> - 3600
> - - coredump
> - false
> + - - feedback_enabled
> + - true
> + - - feedback_host
> + - https://feedback.tarantool.io
> + - - feedback_interval
> + - 3600
> - - force_recovery
> - false
> - - hot_standby
> @@ -111,6 +117,12 @@ cfg_filter(box.cfg)
> - 3600
> - - coredump
> - false
> + - - feedback_enabled
> + - true
> + - - feedback_host
> + - https://feedback.tarantool.io
> + - - feedback_interval
> + - 3600
> - - force_recovery
> - false
> - - hot_standby
> diff --git a/test/box/errinj.result b/test/box/errinj.result
> index 054045c..667c309 100644
> --- a/test/box/errinj.result
> +++ b/test/box/errinj.result
> @@ -16,6 +16,8 @@ errinj.info()
> state: 0
> ERRINJ_WAL_WRITE:
> state: false
> + ERRINJ_HTTPC_EXECUTE:
> + state: false
> ERRINJ_VYRUN_DATA_READ:
> state: false
> ERRINJ_VY_SCHED_TIMEOUT:
> diff --git a/test/box/misc.result b/test/box/misc.result
> index cd3af7f..57717c4 100644
> --- a/test/box/misc.result
> +++ b/test/box/misc.result
> @@ -63,6 +63,7 @@ t
> - commit
> - ctl
> - error
> + - feedback
> - index
> - info
> - internal
> diff --git a/test/engine/iterator.result b/test/engine/iterator.result
> index 9cedc6a..0c4fcc9 100644
> --- a/test/engine/iterator.result
> +++ b/test/engine/iterator.result
> @@ -4215,7 +4215,7 @@ s:replace{35}
> ...
> state, value = gen(param,state)
> ---
> -- error: 'builtin/box/schema.lua:983: usage: next(param, state)'
> +- error: 'builtin/box/schema.lua:985: usage: next(param, state)'
> ...
> value
> ---
> diff --git a/test/engine/savepoint.result b/test/engine/savepoint.result
> index 9d238ec..d440efa 100644
> --- a/test/engine/savepoint.result
> +++ b/test/engine/savepoint.result
> @@ -14,7 +14,7 @@ s1 = box.savepoint()
> ...
> box.rollback_to_savepoint(s1)
> ---
> -- error: 'builtin/box/schema.lua:298: Usage:
> box.rollback_to_savepoint(savepoint)' +- error:
> 'builtin/box/schema.lua:300: Usage: box.rollback_to_savepoint(savepoint)'
> ...
> box.begin() s1 = box.savepoint()
> ---
> @@ -323,27 +323,27 @@ test_run:cmd("setopt delimiter ''");
> ok1, errmsg1
> ---
> - false
> -- 'builtin/box/schema.lua:298: Usage: box.rollback_to_savepoint(savepoint)'
> +- 'builtin/box/schema.lua:300: Usage:
> box.rollback_to_savepoint(savepoint)' ...
> ok2, errmsg2
> ---
> - false
> -- 'builtin/box/schema.lua:298: Usage: box.rollback_to_savepoint(savepoint)'
> +- 'builtin/box/schema.lua:300: Usage:
> box.rollback_to_savepoint(savepoint)' ...
> ok3, errmsg3
> ---
> - false
> -- 'builtin/box/schema.lua:298: Usage: box.rollback_to_savepoint(savepoint)'
> +- 'builtin/box/schema.lua:300: Usage:
> box.rollback_to_savepoint(savepoint)' ...
> ok4, errmsg4
> ---
> - false
> -- 'builtin/box/schema.lua:298: Usage: box.rollback_to_savepoint(savepoint)'
> +- 'builtin/box/schema.lua:300: Usage:
> box.rollback_to_savepoint(savepoint)' ...
> ok5, errmsg5
> ---
> - false
> -- 'builtin/box/schema.lua:298: Usage: box.rollback_to_savepoint(savepoint)'
> +- 'builtin/box/schema.lua:300: Usage:
> box.rollback_to_savepoint(savepoint)' ...
> s:select{}
> ---
-------------- next part --------------
A non-text attachment was scrubbed...
Name: signature.asc
Type: application/pgp-signature
Size: 488 bytes
Desc: This is a digitally signed message part.
URL: <https://lists.tarantool.org/pipermail/tarantool-patches/attachments/20180307/d7336e3d/attachment.sig>
More information about the Tarantool-patches
mailing list