#124

Committed 15 Mar 2025 05:41PM UTC coverage: 65.503% (-0.04%) from 65.54%

Build # #124

Build Type

push

Committed by Vladislav Grubov

Commit Message

ci: fix rpmbuild

Run Details

5271 of 8047 relevant lines covered (65.5%)

1507.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.38

/switchover/_tarantool.lua

---
---@class Tarantool
---reflects single Tarantool connection
---@field public _name? string Tarantool name (can be nil)
---@field public conn NetBoxConnection? Tarantool connection
---@field public endpoint string URI to Tarantool
---@field public masked_endpoint string URI to Tarantool (with stripped password)
---@field public need_discovery boolean flag to enable background fiber of discovery (default: true)
---@field public fqdn string retreived fqdn of Tarantool server
---@field public ipv4 string resolved ipv4 from endpoint of Tarantool server
---@field public port string listen port of Tarantool
---@field public discovery_timeout number discovery timeout
---@field public has_vshard boolean flag which is set when vshard is discovered
---@field public can_package_reload boolean flag which is set when package.reload is discovered
---@field public has_etcd boolean flag which is set when config.etcd is discovered
---@field public has_cartridge boolean flag which is set when cartridge is discovered
---@field public has_metrics boolean flag which is set when metrics is discovered
---@field public etcd_config table info about etcd config of the instance (endpoints, timeout, dc, etc.)
---@field public fetch table fetch more info
---@field private log_format string log format
---@field public rtt number? ping to instance (conn:ping())
---@field public spaces {name: string, len: number, bsize: number}[] space information
---@field public slab_stats { mem_free: number, mem_used: number, item_count: number, item_size: number, slab_count: number, slab_size: number }[] slab stats
---@field public runtime_info { lua: number, maxalloc: number, used: number } runtime info
---@field public vshard_alerts? SwitchoverTarantoolVshardAlerts kv-map of vshard alerts
local Tarantool = {}
Tarantool.__index = Tarantool

---@class SwitchoverTarantoolVshardAlerts
---@field storage {[1]:string, [2]: string}[]
---@field router {[1]:string, [2]: string}[]
---@field storage_status number
---@field router_status number
---@field is_storage boolean
---@field is_router boolean
---@field this_replicaset_master_uuid string?

local socket = require 'socket'
local G = require 'switchover._global'
local log = require 'log'
local uri = require 'uri'
local fun = require 'fun'
local json = require 'json'
local fiber = require 'fiber'
local clock = require 'clock'
local errno = require 'errno'
local netbox = require 'net.box'
local v = require 'switchover._semver'
local background = require 'switchover._background'
local reload_will_freeze_before = v'2.2.1'


local discovery_format = {
        {
                v = v"0.0.0",
                f = '{{endpoint}} - {{connected and "not supported version" or "not connected"}}'
        },
        {
                v = v"1.10.0",
                f = '{{fqdn}}:{{port}} '
                        ..'{{(etcd.dc or "").." "}}'
                        ..'{{info.version:gsub("^(%d+)%.(%d+)%.(%d+)%-(%d+).*$", "%1.%2.%3.%4")}} '
                        ..'id:{{info.id}} s:{{info.status}} v:{{info.vclock}} r:{{info.ro and "replica" or "master"}}'
                        ..'{{info.gc.checkpoint_is_in_progress and " snap:true" or ""}}',
        },
        {
                v = v"2.6.0",
                f = '{{fqdn}}:{{port}} {{info.version:gsub("^(%d+)%.(%d+)%.(%d+)%-(%d+).*$", "%1.%2.%3.%4")}}'
                        ..' id:{{info.id}} s:{{info.status}} v:{{info.vclock}} r:{{info.ro}} t:{{info.election.term}}'
                        ..' l:{{info.election.leader}} s:{{info.election.state}} m:{{cfg.election_mode}}'
                        ..' q:{{info.synchro.quorum}} qo:{{info.synchro.queue.owner}}',
        },
}

local function human_uptime(ts)
        local s = {}
        if ts > 86400 then
                local days = math.floor(ts/86400)
                ts = ts % 86400
                table.insert(s, ("%dd"):format(days))
        end
        if ts > 3600 then
                local hours = math.floor(ts/3600)
                ts = ts % 3600
                table.insert(s, ("%dh"):format(hours))
        end
        if ts > 60 then
                local mins = math.floor(ts/60)
                ts = ts % 60
                table.insert(s, ("%dm"):format(mins))
        end
        if ts > 0 then
                table.insert(s, ("%ds"):format(ts))
        end

        return table.concat(s, "")
end

local ffi = require 'ffi'
ffi.cdef[[
        struct hostent {
                char  *h_name;            /* official name of host */
                char **h_aliases;         /* alias list */
                int    h_addrtype;        /* host address type */
                int    h_length;          /* length of address */
                char **h_addr_list;       /* list of addresses */
        };

        struct hostent *gethostbyaddr(const void *addr, socklen_t len, int type);

        /* Internet address */
        struct in_addr {
                uint32_t       s_addr;     /* address in network byte order */
        };
        int inet_pton(int af, const char *restrict src, void *restrict dst);

        static const int AF_INET = 2;
        static const int AF_INET6 = 10;
]]

local function fqdn(endpoint)
        local u = uri.parse(endpoint)
        if not u.ipv4 then
                return u.host
        end

        local in_addr = ffi.new('struct in_addr', {})
        if 1 ~= ffi.C.inet_pton(ffi.C.AF_INET, u.ipv4, in_addr) then
                log.warn('Cant build inet_pton(%s, %s) => %s', ffi.C.AF_INET, u.ipv4, errno.strerror(errno()))
                return u.host
        end

        local hostent = ffi.C.gethostbyaddr(in_addr, ffi.sizeof(in_addr), ffi.C.AF_INET)
        if hostent == nil then
                log.warn('Cant gethostbyaddr(%s, %s, %s) => %s',
                        in_addr, ffi.sizeof(in_addr), ffi.C.AF_INET, errno.strerror(errno()))
                return u.host
        end

        local host = ffi.string(hostent.h_name):gsub("(%.[^.]+)", "")
        return host
end

local function ipv4(endpoint)
        local _uri = uri.parse(endpoint)
        if _uri.ipv4 then
                return _uri.ipv4
        end

        local resolv, err = socket.getaddrinfo(_uri.host, 0, G.dns_timeout, {
                type = 'SOCK_STREAM',
                family = 'AF_INET',
                protocol = 'tcp',
        })

        if not resolv then
                log.warn("Failed to resolve %s => %s", _uri.host, err)
                return
        end

        return resolv[1].host
end

---
---@return table
function Tarantool:__serialize()
        return {
                endpoint = self.endpoint:gsub("^([^:]+):([^@]+)@", function(u) return u..':@' end),
                connected = self.connected,
                info = self.connected and self:info(),
                cfg  = self.connected and self:cfg(),
                slab = self.connected and self:info() and self.slab_info,
                stat = self.connected and self:info() and self.stat_info,
                etcd = self.connected and self:etcd(),
                fqdn = self.fqdn,
                port = self.port,
                n_upstreams = self.connected and #self:followed_upstreams(),
                n_downstreams = self.connected and #self:followed_downstreams(),
                uptime = self.connected and human_uptime(self:info().uptime),
                vclock = self.connected and self:jvclock(),
                wall_clock = self.connected and self.wall_clock,
                n_fibers = self.connected and self.n_fibers,
        }
end

---Serializes entire tarantool info
---@return table
function Tarantool:serialize()
        return Tarantool.__serialize(self)
end

---
---__tostring metamethod for pretty print. Now calls `short_status`
---@see Tarantool.short_status
function Tarantool:__tostring()
        local vars = setmetatable(Tarantool.__serialize(self), { __index = _G })
        local format
        if self.log_format then
                format = self.log_format
        elseif self.connected then
                format = fun.range(1, #discovery_format)
                        :map(function(i) return discovery_format[i] end)
                        :take_while(function(vf) return vf.v <= self:version() end)
                        :reduce(function(_, vf) --[[@return {v:string, f:string}]] return vf end).f
        else
                format = discovery_format[1].f
        end
        return (format:gsub('{{([^{}]+)}}', function(var)
        local fnc = ([[return %s]]):format(var)
                local exec = assert(loadstring(fnc))
                setfenv(exec, vars)

                local ok, val = pcall(exec)
                if not ok then
                        log.verbose("Failed to execute %q: %s", var, val)
                        return '?'
                end
                if type(val) == 'table' then
                        val = json.encode(val)
                end
                return tostring(val)
        end))
end

---
---Basically pretty print of usefull instance information
---@return string
function Tarantool:short_status()
        return self:__tostring()
end

---
---on_connect callback is called when connection (re)established to tarantool
---calls `start_discovery` method
function Tarantool:on_connect()
        self.connected = true
        log.verbose("Connected %s", self)
        if self.need_discovery then
                self:start_discovery()
        end
end

---
---@class TarantoolOpts
---@field public async boolean `default:nil` issues asynchronous connection (net.box.connect {wait_connected=false})
---@field public discovery_timeout number|nil `default:0.1` timeout of refreshing tarantool info.
---@field public fetch table which fields should be fetched from Tarantool during discovery
---@field public log_format string
---@field public need_discovery boolean flag which enabled background discovery (default: true)
---@field public on_connect fun(tnt:Tarantool):nil on_connect callback

---
---Constructor of the class.
---Creates new object and connects to specified endpoint
---@param endpoint string endpoint `<host>:<port>` to the Tarantool instance
---@param opts TarantoolOpts
---@return Tarantool
function Tarantool:new(endpoint, opts)
        log.verbose("Connecting to %s, async=%s", endpoint, opts.async)
        local conn = netbox.connect(endpoint, {
                wait_connected = opts.async ~= true,
                reconnect_after = 0.1,
        })
        local tnt = setmetatable({
                conn = conn,
                endpoint = endpoint,
                fqdn = fqdn(endpoint),
                port = uri.parse(endpoint).service,
                ipv4 = ipv4(endpoint),
                connected = false,
                masked_endpoint = ("{host}:{service}"):gsub("{(%w+)}", uri.parse(endpoint)),
                need_discovery = opts.need_discovery ~= false,
                discovery_timeout = opts.discovery_timeout or 0.1,
                log_format = opts.log_format,
                fetch = opts.fetch,
        }, self)
        conn:on_connect(function(_)
                fiber.create(function()
                        assert(tnt.conn, "conn must be present")
                        fiber.name("c/"..tnt.masked_endpoint)
                        if opts.on_connect then
                                opts.on_connect(tnt)
                        end
                        if tnt.conn then tnt:on_connect() end
                end)
        end)
        conn:on_disconnect(function()
                tnt.connected = false
        end)
        if not opts.async then
                log.verbose("calling on_connect for %s", tnt.masked_endpoint);
                (opts.on_connect or Tarantool.on_connect)(tnt)
        end
        return tnt
end

---
---Spawns separate fiber `discovery_f` where constantly calls self:_get method
---@see Tarantool#_get
function Tarantool:start_discovery()
        self.discovery_f = background {
                name = 'd/' .. self.masked_endpoint,
                wait = false,
                restart = false,
                run_interval = self.discovery_timeout,
                args = { self },
                run_while = function ()
                        return self.connected == true and not self.destroyed
                end,
                func = function(_, tnt)
                        tnt:_get{ update = true }
                end,
                teardown = function()
                        log.verbose("Leaving discovery fiber of %s", self)
                end,
        }
        self.pinger_f = background {
                name = 'p/' .. self.masked_endpoint,
                wait = false,
                restart = false,
                run_interval = 0.1,
                args = { self },
                run_while = function ()
                        return self.connected == true and not self.destroyed
                end,
                func = function(_, tnt)
                        local t = clock.time()
                        if tnt.conn:ping({ timeout = math.max(0.1, self.discovery_timeout) }) then
                                tnt.rtt = clock.time()-t
                        end
                end,
                teardown = function()
                        log.verbose("Leaving discovery fiber of %s", self)
                end,
        }
        assert(self.discovery_f, "discovery job not created")
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return number replica_id ID of the replica `box.info.id`
function Tarantool:id(opts)
        return tonumber(self:info(opts).id) or 0
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string instance_uuid UUID of the replica `box.info.uuid`
function Tarantool:uuid(opts)
        return self:info(opts).uuid
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return  string status of the replica `box.info.status`
function Tarantool:status(opts)
        return self:info(opts).status
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string replicaset_uuid replicaset_uuid of the replica `box.info.cluster.uuid`
function Tarantool:cluster_uuid(opts)
        return self:info(opts).cluster.uuid
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string version tarantool version ex: 1.10.10-0-gxxxxxx
function Tarantool:version(opts)
        return v(self:info(opts).version)
end

---sets name for connection
---@param new_name string
function Tarantool:setname(new_name)
        self._name = new_name
end

---
---@return string
function Tarantool:name()
        return ("%s (%s)"):format(self._name or 'unknown', self.masked_endpoint)
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string jsoned_vclock `json.encode(box.info.vclock)`
function Tarantool:jvclock(opts)
        local max_id = 0
        local vclock = self:info(opts).vclock
        for id, _ in pairs(vclock) do
                if max_id < id then max_id = id end
        end
        local r = table.new(max_id, 0)
        for i = 1, max_id do
                if vclock[i] then
                        r[i] = vclock[i]
                else
                        r[i] = ''
                end
        end
        return json.encode(setmetatable(r, {__serialize='seq'}))
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return integer[] vclock `box.info.vclock`
function Tarantool:vclock(opts)
        return self:info(opts).vclock
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return number sign returns sum of lsn excluding vclock[0]
function Tarantool:signature(opts)
        return fun.sum(self:vclock(opts))
end

---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string[] list of box.cfg.replication (may be empty)
function Tarantool:replication(opts)
        return self:cfg(opts).replication or {}
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return boolean ro returns `box.info.ro`
function Tarantool:ro(opts)
        return self:info(opts).ro
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return string role evaluates role master/replica based on `box.info.ro`
function Tarantool:role(opts)
        return self:info(opts).ro == true and 'replica' or 'master'
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return table election returns `box.info.election`
function Tarantool:election(opts)
        return self:info(opts).election
end

---
---@param instance_uuid string UUID of the upstream
---@return ReplicaUpstreamInfo? upstream returns information about upstream identified by `instance_uuid`
function Tarantool:upstream(instance_uuid)
        return fun.iter(self:upstreams()):grep(function(u)
                return u.upstream and u.uuid == instance_uuid
        end):nth(1)
end

---
---@param instance_uuid string UUID of the downstream
---@return ReplicaDownstreamInfo? downstream returns information about downstream identified by `instance_uuid`
function Tarantool:downstream(instance_uuid)
        return fun.iter(self:downstreams()):grep(function(u)
                return u.downstream and u.downstream.uuid == instance_uuid
        end):nth(1)
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return ReplicaUpstreamInfo[] upstreams returns list of all upstreams of this instance `box.info.replication.?.upstream`
function Tarantool:upstreams(opts)
        local ret = {}
        for _, r in pairs(self:info(opts).replication) do
                if r.upstream and r.id ~= self:id() then
                        table.insert(ret, r)
                end
        end
        table.sort(ret, function(a, b) return a.id < b.id end)
        return ret
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return ReplicaDownstreamInfo[] downstreams returns list of all downstreams of this instance `box.info.replication.?.downstream`
function Tarantool:downstreams(opts)
        local ret = {}
        for _, r in pairs(self:info(opts).replication) do
                if r.downstream and r.id ~= self:id() then
                        table.insert(ret, r)
                end
        end
        table.sort(ret, function(a, b) return a.id < b.id end)
        return ret
end

---
---@param opts? SwitchoverTarantoolRefreshOptions refresh options
---@return ReplicaDownstreamInfo[] downstreams returns list of all followed downstreams of this instance `box.info.replication.?.downstream`
function Tarantool:followed_downstreams(opts)
        local ret = {}
        for _, d in ipairs(self:downstreams(opts)) do
                if d.downstream and d.downstream.status == 'follow' then
                        table.insert(ret, d)
                end
        end
        return ret
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return ReplicaUpstreamInfo[] upstreams returns list of all followed upstreams of this instance `box.info.replication.?.upstream`
function Tarantool:followed_upstreams(opts)
        local ret = {}
        for _, d in ipairs(self:upstreams(opts)) do
                if d.upstream and d.upstream.status == 'follow' then
                        table.insert(ret, d)
                end
        end
        return ret
end

---@class ReplicaUpstreamInfo:ReplicaInfo
---@field upstream UpstreamInfo

---@class ReplicaDownstreamInfo:ReplicaInfo
---@field downstream DownstreamInfo

---
---@param instance_uuid string UUID of the upstream
---@return ReplicaUpstreamInfo? upstream returns upstream info identified with `instance_uuid` (if it is followed)
function Tarantool:replicates_from(instance_uuid)
        return fun.iter(self:upstreams()):grep(function(r)
                return r.uuid == instance_uuid and r.upstream.status == 'follow'
        end):nth(1)
end

---
---@param instance_uuid string UUID of the downstream
---@return ReplicaDownstreamInfo? downstream returns downstream info identified with `instance_uuid` (if it is followed)
function Tarantool:replicated_by(instance_uuid)
        return fun.iter(self:downstreams()):grep(function(r)
                return r.uuid == instance_uuid and r.downstream.status == 'follow'
        end):nth(1)
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return BoxInfo info returns `box.info`
function Tarantool:info(opts)
        self:_get(opts)
        return self.cached_info
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return BoxCfg cfg returns `box.cfg`
function Tarantool:cfg(opts)
        self:_get(opts)
        return self.cached_cfg
end

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---@return table etcd returns `config.get('etcd')`
function Tarantool:etcd(opts)
        self:_get(opts)
        return self.etcd_config
end

---Checks if package.reload() will freeze on certain Tarantool versions
---@param opts? SwitchoverTarantoolRefreshOptions
---@return boolean
function Tarantool:reload_will_freeze(opts)
        self:_get(opts)
        if self:version() >= reload_will_freeze_before then
                return false
        end
        -- node will freeze on package.reload during box.snapshot() on rw node
        return self:info().gc.checkpoint_is_in_progress == true
end

---Checks that vshard.storage() is rw (box.info.ro==false) but vshard didn't update it's topology
---@param opts? SwitchoverTarantoolRefreshOptions
---@return boolean
function Tarantool:vshard_storage_rw_but_ro(opts)
        self:_get(opts)
        -- no master => no cry
        if self:ro() then
                return false
        end
        -- no vshard => no cry
        if not self.has_vshard then
                return false
        end
        -- no vshard storage => no cry
        if not self.vshard_alerts.is_storage then
                return false
        end
        -- tt is vshard storage and it's uuid is the same as master.uuid in vshard topology
        -- (NON_MASTER issue)
        if tostring(self.vshard_alerts.this_replicaset_master_uuid):lower() == self:uuid():lower() then
                return false
        end
        -- Okay :(( vshard has malformed configuration
        return true
end

---@class SwitchoverTarantoolRefreshOptions
---@field public update boolean if set to `true` refreshes cache from connection

---
---@param opts SwitchoverTarantoolRefreshOptions|nil refresh options
---Refreshes info about remote Tarantool:
---box.info
---box.cfg
---has package.reload, has vshard, has config.etcd, has cartridge
function Tarantool:_get(opts)
        opts = opts or {}
        if not self.cached_info or not self.cached_cfg or (opts or {}).update then
                assert(self.conn, "connection is not established")
                log.verbose("Tarantool._get(%s, %s)", self.endpoint, json.encode(opts))
                local info_memory, stat_net_info, info_gc
                self.cached_info,
                self.cached_cfg,
                self.can_package_reload,
                self.has_vshard,
                self.vshard_alerts,
                self.has_etcd,
                self.etcd_config,
                self.has_cartridge,
                self.has_metrics,
                self.cluster_len,
                self.cluster_max_id,
                info_memory,
                self.runtime_info,
                info_gc,
                self.slab_info,
                self.stat_info,
                stat_net_info,
                self.n_fibers,
                self.metrics_export,
                self.fiber_info,
                self.bucket_count,
                self.spaces,
                self.slab_stats,
                self.wall_clock
                        =
                self.conn:eval ([[
                        local fetch = ... or {}
                        local has_etcd = false
                        local has_vshard = rawget(_G, 'vshard') ~= nil
                        local is_storage = has_vshard and vshard.storage and vshard.storage.internal and vshard.storage.internal.current_cfg
                        local is_router = has_vshard and vshard.router and vshard.router.internal and vshard.router.internal.current_cfg
                        local etcd_config = {}
                        if type(package.loaded.config) == 'table' and package.loaded.config.etcd and package.loaded.config._flat then
                                has_etcd = true
                                etcd_config = package.loaded.config.get('etcd')
                        end
                        return
                                box.info,
                                box.cfg,
                                type(package.reload) ~= 'nil',
                                has_vshard,
                                fetch.vshard_alerts and has_vshard
                                        and {
                                                storage = is_storage and vshard.storage.info().alerts,
                                                router = is_router and vshard.router.info().alerts,
                                                storage_status = is_storage and vshard.storage.info().status,
                                                router_status = is_router and vshard.router.info().status,
                                                is_storage = is_storage ~= nil,
                                                is_router = is_router ~= nil,
                                                this_replicaset_master_uuid = is_storage
                                                        and vshard.storage.internal.this_replicaset
                                                        and vshard.storage.internal.this_replicaset.master
                                                        and vshard.storage.internal.this_replicaset.master.uuid,
                                        }
                                        or {},
                                has_etcd,
                                etcd_config,
                                type(package.loaded.cartridge) ~= 'nil',
                                type(package.loaded.metrics) ~= 'nil',
                                box.space._cluster:len(),
                                box.space._cluster.index[0]:max()[1],
                                box.info.memory(),
                                box.runtime.info(),
                                box.info.gc(),
                                box.slab.info(),
                                box.stat(),
                                box.stat.net(),
                                require 'fun'.length(require 'fiber'.info()),
                                type(package.loaded.metrics) ~= 'nil'
                                        and fetch.metrics
                                        and select(2, xpcall(function() require 'json'.decode(require 'metrics.plugins.json'.export()) end, function() return {} end))
                                        or box.NULL,
                                fetch.fiber_info
                                        and require 'fiber'.info() or box.NULL,
                                box.space._bucket
                                        and box.space._bucket:len()
                                        or 'no',
                                fetch.spaces
                                        and require 'fun'.iter(box.space)
                                                        :grep(function(k,s) return type(k) == 'number' and k>511 and next(s.index) end)
                                                        :map(function(_, s)
                                                                return {
                                                                        name=s.name,
                                                                        id=s.id,
                                                                        len=s:len(),
                                                                        bsize=s:bsize(),
                                                                        isize=require'fun'.range(0, #s.index)
                                                                                :map(function(i) return s.index[i]:bsize() end)
                                                                                :sum(),
                                                                        engine=s.engine
                                                                }
                                                        end)
                                                        :totable()
                                        or {},
                                fetch.slabs and box.slab.stats() or {},
                                require 'clock'.time()
                ]], {self.fetch}, {})
                self.cached_info.memory = info_memory
                self.cached_info.gc = info_gc
                self.stat_info.net = stat_net_info
        end
        return self.cached_info, self.cached_cfg
end

---
---force_promote calls `box.cfg{read_only=false} box.ctl.promote()` on remote
---@param args {allow_vshard: boolean}
---@return boolean is_safe, string? error_msg
function Tarantool:force_promote(args)
        if self.has_vshard and not args.allow_vshard then
                error("Cant force_promote: instance is in vshard cluster", 2)
        end
        log.warn("Executing force_promote on %s", self)
        local ok, err = pcall(function()
                self.cached_info, self.cached_cfg = self.conn:eval[[
                        box.cfg{ read_only = false }
                        if box.ctl.promote
                                and box.info.synchro
                                and box.info.synchro.queue
                                and box.info.synchro.queue.owner ~= 0
                                and box.info.synchro.queue.owner ~= box.info.id
                        then
                                box.ctl.promote()
                        end
                        return box.info, box.cfg
                ]]
        end)

        if not ok then
                return false, err
        end

        if not self:ro() then
                return true
        else
                return true, "candidate still ro"
        end
end

---Executes cartridge.failover_promote({ [replicaset_uuid] = instance_uuid })
---@param timeout number eval timeout
---@return unknown
function Tarantool:cartridge_promote(timeout)
        assert(self.has_cartridge, "instance must have cartridge")
        return self.conn:eval([[ local map = ... return require 'cartridge'.failover_promote(map) ]], {
                {[self:cluster_uuid()] = self:uuid()},
        }, { timeout = timeout })
end

---
---package_reload calls `package.reload()` on remote
---
---Raises if Tarantool cannot package_reload
---@param timeout number?
---@return boolean
function Tarantool:package_reload(timeout)
        if not self.can_package_reload then
                error(("Instance %s does not support package.reload"):format(self.endpoint), 0)
        end

        log.warn("Calling package.reload on %s", self)
        self.cached_info, self.cached_cfg = self.conn:eval([[
                package.reload()
                return box.info, box.cfg
        ]], {}, { timeout = timeout })
        return true
end

---Restarts replication on instance
---@param timeout number?
---@return true|false
---@return unknown|nil
function Tarantool:restart_replication(timeout)
        log.warn("Calling restart-replication on %s", self)

        timeout = tonumber(timeout) or 30
        timeout = math.max(timeout, 3)
        timeout = math.min(timeout, 30)

        local ok, err = pcall(function()
                self.conn:eval([[
                        repl = box.cfg.replication
                        box.cfg{ replication = {} }
                        box.cfg{ replication = repl }
                        repl = nil
                ]], {}, { timeout = timeout })
        end)

        if not ok then
                log.error("restart-replication failed on instance %s : %s", self, err)
        end
        return ok, err
end

function Tarantool:package_reload_check_freeze(allow_freeze)
        if self:ro() == false and self:reload_will_freeze() and not allow_freeze then
                return false
        end
        return self:package_reload()
end

---
---@return table
function Tarantool:raft()
        return {
                mode            = self:cfg().election_mode,
                quorum          = self:cfg().replication_synchro_quorum,
                election_quorum = math.ceil((self.cluster_len + 1)/2),
                state           = self:election().state,
                leader          = self:election().leader,
                term            = self:election().term,
        }
end

function Tarantool:shadow()
        return setmetatable({
                connected = true,
                endpoint = self.endpoint,
                cached_info = self.cached_info,
                cached_cfg = self.cached_cfg,
                rtt        = self.rtt,

                _name = self._name,

                has_vshard = self.has_vshard,
                vshard_alerts = self.vshard_alerts,
                has_etcd = self.has_etcd,
                etcd_config = self.etcd_config,
                has_cartridge = self.has_cartridge,
                has_metrics = self.has_metrics,
                cluster_len = self.cluster_len,
                cluster_max_id = self.cluster_max_id,
                runtime_info = self.runtime_info,
                slab_info = self.slab_info,
                stat_info = self.stat_info,
                n_fibers = self.n_fibers,
                metrics_export = self.metrics_export,
                fiber_info = self.fiber_info,
                bucket_count = self.bucket_count,
                spaces = self.spaces,
                slab_stats = self.slab_stats,
                wall_clock = self.wall_clock,
        }, Tarantool)
end

---Destroys tarantool connection
function Tarantool:destroy()
        log.verbose("destroying connection to instance: %s/%s", self.masked_endpoint, tostring(self.discovery_f))
        self.destroyed = true
        self:destroy_background()
        if self.conn then
                if type(self.conn.opts) == 'table' then
                        self.conn.opts.reconnect_after = nil
                end
                self.conn:close()
                self.conn = nil
        end
        self.connected = nil
end

function Tarantool:destroy_background()
        log.verbose("destroying background: %s/%s", self.masked_endpoint, tostring(self.discovery_f))
        if self.discovery_f then
                self.discovery_f:shutdown('force')
                self.discovery_f = nil
        end
        log.verbose("destroying background: %s/%s", self.masked_endpoint, tostring(self.pinger_f))
        if self.pinger_f then
                self.pinger_f:shutdown('force')
                self.pinger_f = nil
        end
end

return setmetatable(Tarantool, { __call = Tarantool.new })

ochaton / switchover / #124

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous