save fibers in the local table

Author: patapenka-alexey

metrics/plugins/graphite.lua

@@ -3,7 +3,6 @@ local fiber = require('fiber')
 local metrics = require('metrics')
 local checks = require('checks')
 local log = require('log')
-local fun = require('fun')
 
 local graphite = {}
 
@@ -16,6 +15,30 @@ local DEFAULT_SEND_INTERVAL = 2
 -- Constants
 local LABELS_SEP = ';'
 
+local GRAPHITE_FIBERS = {}
+
+local function create_fiber_table(opts)
+    local graphite_fiber = {}
+
+    if opts ~= nil then
+        graphite_fiber.prefix = opts.prefix or DEFAULT_PREFIX
+        graphite_fiber.host = opts.host or DEFAULT_HOST
+        graphite_fiber.port = opts.port or DEFAULT_PORT
+        graphite_fiber.send_interval = opts.send_interval or DEFAULT_SEND_INTERVAL
+        graphite_fiber.name = 'metrics_graphite_worker' .. '_' ..
+            graphite_fiber.prefix .. '_' .. graphite_fiber.host .. '_' ..
+            graphite_fiber.port .. '_' .. graphite_fiber.send_interval
+    else
+        graphite_fiber.prefix = DEFAULT_PREFIX
+        graphite_fiber.host = DEFAULT_HOST
+        graphite_fiber.port = DEFAULT_PORT
+        graphite_fiber.send_interval = DEFAULT_SEND_INTERVAL
+        graphite_fiber.name =  'metrics_graphite_worker'
+    end
+
+    return graphite_fiber
+end
+
 function graphite.format_observation(prefix, obs)
     local metric_path = #prefix > 0 and ('%s.%s'):format(prefix, obs.metric_name) or obs.metric_name
 
@@ -36,41 +59,23 @@ function graphite.format_observation(prefix, obs)
     return graph
 end
 
-local function create_fiber_name(opts)
-    local fiber_name
-
-    if opts ~= nil then
-        local prefix = opts.prefix or DEFAULT_PREFIX
-        local host = opts.host or DEFAULT_HOST
-        local port = opts.port or DEFAULT_PORT
-        local send_interval = opts.send_interval or DEFAULT_SEND_INTERVAL
-
-        fiber_name = 'metrics_graphite_worker' .. '_' ..
-            prefix .. '_' .. host .. '_' .. port .. '_' .. send_interval
-    else
-        fiber_name = 'metrics_graphite_worker'
-    end
-
-    return fiber_name
-end
-
-local function graphite_worker(opts)
-    fiber.name(create_fiber_name(opts))
+local function graphite_worker(args)
+    fiber.name(args.name)
 
     while true do
         metrics.invoke_callbacks()
         for _, c in pairs(metrics.collectors()) do
             for _, obs in ipairs(c:collect()) do
-                local data = graphite.format_observation(opts.prefix, obs)
-                local numbytes = opts.sock:sendto(opts.host, opts.port, data)
+                local data = graphite.format_observation(args.prefix, obs)
+                local numbytes = args.sock:sendto(args.host, args.port, data)
                 if numbytes == nil then
                     log.error('Error while sending to host %s port %s data %s',
-                              opts.host, opts.port, data)
+                              args.host, args.port, data)
                 end
             end
         end
 
-        fiber.sleep(opts.send_interval)
+        fiber.sleep(args.send_interval)
     end
 end
 
@@ -82,25 +87,32 @@ function graphite.init(opts)
         send_interval = '?number'
     }
 
+    local graphite_fiber = create_fiber_table(opts)
+
     local sock = socket('AF_INET', 'SOCK_DGRAM', 'udp')
     assert(sock ~= nil, 'Socket creation failed')
 
-    local prefix = opts.prefix or DEFAULT_PREFIX
-    local host = opts.host or DEFAULT_HOST
-    local port = opts.port or DEFAULT_PORT
-    local send_interval = opts.send_interval or DEFAULT_SEND_INTERVAL
-
-    fun.iter(fiber.info()):
-        filter(function(_, x) return x.name == create_fiber_name(opts) end):
-        each(function(x) fiber.kill(x) end)
+    -- for _, v in ipairs(GRAPHITE_FIBERS) do
+    --     if v.name == graphite_fiber.name then
+    --         error('failed to start fiber: ' .. graphite_fiber.name .. ", already exist")
+    --     end
+    -- end
+    for name in pairs(GRAPHITE_FIBERS) do
+        if name == graphite_fiber.name then
+            error('failed to start fiber: ' .. graphite_fiber.name .. ", already exist")
+        end
+    end
 
-    fiber.create(graphite_worker, {
-        prefix = prefix,
+    graphite_fiber.fiber = fiber.create(graphite_worker, {
+        name = graphite_fiber.name,
+        prefix = graphite_fiber.prefix,
         sock = sock,
-        host = host,
-        port = port,
-        send_interval = send_interval,
+        host = graphite_fiber.host,
+        port = graphite_fiber.port,
+        send_interval = graphite_fiber.send_interval,
     })
+
+    GRAPHITE_FIBERS[graphite_fiber.name] = graphite_fiber
 end
 
 function graphite.stop(opts)
@@ -111,9 +123,17 @@ function graphite.stop(opts)
         send_interval = '?number'
     }
 
-    fun.iter(fiber.info()):
-        filter(function(_, x) return x.name == create_fiber_name(opts) end):
-        each(function(x) fiber.kill(x) end)
+    local graphite_fiber = create_fiber_table(opts)
+
+    for name in pairs(GRAPHITE_FIBERS) do
+        if name == graphite_fiber.name then
+            fiber.kill(GRAPHITE_FIBERS[name].fiber)
+            GRAPHITE_FIBERS[name] = nil
+            return
+        end
+    end
+
+    error("failed to stop fiber: " .. graphite_fiber.name .. ", doesn't exist")
 end
 
 return graphite

test/plugins/graphite_test.lua

@@ -36,16 +36,8 @@ end)
 
 g.after_each(function(cg)
     cg.server:exec(function()
-        local fiber = require('fiber')
-        local fun = require('fun')
-        local metrics = require('metrics')
         -- Delete all collectors and global labels
-        metrics.clear()
-
-        fun.iter(fiber.info()):
-            filter(function(_, x) return string.find(x.name, 'metrics_graphite_worker') end):
-            each(function(x) fiber.kill(x) end)
-        fiber.yield() -- let cancelled fibers disappear from fiber.info()
+        require('metrics').clear()
     end)
 end)
 
@@ -153,46 +145,21 @@ g.test_graphite_sends_data_to_socket = function(cg)
 
         local cnt = metrics.counter('test_cnt', 'test-cnt')
         cnt:inc(1)
+
         graphite.init({port = port})
     end, {port})
 
-    require('fiber').sleep(0.5)
+    sock:readable(2)
+
     local graphite_obs = sock:recvfrom(50)
     local obs_table = graphite_obs:split(' ')
     t.assert_equals(obs_table[1], 'tarantool.test_cnt')
     t.assert_equals(obs_table[2], '1')
     sock:close()
-end
 
-g.test_graphite_kills_previous_fibers_on_init = function(cg)
     cg.server:exec(function()
-        local fiber = require('fiber')
-        local fun = require('fun')
-        local graphite = require('metrics.plugins.graphite')
-
-        local function mock_graphite_worker()
-            fiber.create(function()
-                fiber.name('metrics_graphite_worker_tarantool_127.0.0.1_2003_2')
-                fiber.sleep(math.huge)
-            end)
-        end
-
-        local function count_workers()
-            return fun.iter(fiber.info()):
-                filter(function(_, x) return string.find(x.name, 'metrics_graphite_worker') end):
-                length()
-        end
-
-        graphite.stop()
-
-        t.assert_equals(count_workers(), 0)
-        mock_graphite_worker()
-        mock_graphite_worker()
-        t.assert_equals(count_workers(), 2)
-
-        graphite.init({})
-        fiber.yield() -- let cancelled fibers disappear from fiber.info()
-        t.assert_equals(count_workers(), 1)
+        require('metrics.plugins.graphite').stop({port = 22003})
+        require('fiber').yield()
     end)
 end
 
@@ -214,7 +181,7 @@ g.test_graphite_stop_default_fibers = function(cg)
         t.assert_equals(count_workers(), 1)
 
         graphite.stop({})
-        fiber.yield() -- let cancelled fibers disappear from fiber.info()
+        fiber.yield()
         t.assert_equals(count_workers(), 0)
     end)
 end
@@ -238,17 +205,26 @@ g.test_graphite_stop_custom_fiber = function(cg)
             send_interval = 1,
         }
 
+        local opts2 = {
+            prefix = "tarantool",
+            host = "127.0.0.1",
+            port = 4444,
+            send_interval = 1,
+        }
+
         t.assert_equals(count_workers(), 0)
 
         graphite.init(opts)
+        graphite.init(opts2)
 
-        t.assert_equals(count_workers(), 1)
-
-        graphite.stop()
-        t.assert_equals(count_workers(), 1) -- fiber still works
+        t.assert_equals(count_workers(), 2)
 
         graphite.stop(opts)
-        fiber.yield() -- let cancelled fibers disappear from fiber.info()
+        fiber.yield()
+        t.assert_equals(count_workers(), 1)
+
+        graphite.stop(opts2)
+        fiber.yield()
         t.assert_equals(count_workers(), 0)
     end)
 end