graphite: multiple servers support

This patch adds the ability to send metrics to the multiple servers.

Part of #TNTP-6584

Author: patapenka-alexey

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `graphite`: ability to send metrics to the multiple servers.
+
 ### Changed
 
 ### Fixed

metrics/plugins/graphite.lua

@@ -36,8 +36,26 @@ 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('metrics_graphite_worker')
+    fiber.name(create_fiber_name(opts))
 
     while true do
         metrics.invoke_callbacks()
@@ -73,7 +91,7 @@ function graphite.init(opts)
     local send_interval = opts.send_interval or DEFAULT_SEND_INTERVAL
 
     fun.iter(fiber.info()):
-        filter(function(_, x) return x.name == 'metrics_graphite_worker' end):
+        filter(function(_, x) return x.name == create_fiber_name(opts) end):
         each(function(x) fiber.kill(x) end)
 
     fiber.create(graphite_worker, {
@@ -85,4 +103,17 @@ function graphite.init(opts)
     })
 end
 
+function graphite.stop(opts)
+    checks {
+        prefix = '?string',
+        host = '?string',
+        port = '?number',
+        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)
+end
+
 return graphite

test/plugins/graphite_test.lua

@@ -41,8 +41,9 @@ g.after_each(function(cg)
         local metrics = require('metrics')
         -- Delete all collectors and global labels
         metrics.clear()
+
         fun.iter(fiber.info()):
-            filter(function(_, x) return x.name == 'metrics_graphite_worker' end):
+            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()
     end)
@@ -171,17 +172,19 @@ g.test_graphite_kills_previous_fibers_on_init = function(cg)
 
         local function mock_graphite_worker()
             fiber.create(function()
-                fiber.name('metrics_graphite_worker')
+                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 x.name == 'metrics_graphite_worker' end):
+                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()
@@ -192,3 +195,60 @@ g.test_graphite_kills_previous_fibers_on_init = function(cg)
         t.assert_equals(count_workers(), 1)
     end)
 end
+
+g.test_graphite_stop_default_fibers = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local fun = require('fun')
+        local graphite = require('metrics.plugins.graphite')
+
+        local function count_workers()
+            return fun.iter(fiber.info()):
+                filter(function(_, x) return string.find(x.name, 'metrics_graphite_worker') end):
+                length()
+        end
+
+        t.assert_equals(count_workers(), 0)
+
+        graphite.init({})
+        t.assert_equals(count_workers(), 1)
+
+        graphite.stop({})
+        fiber.yield() -- let cancelled fibers disappear from fiber.info()
+        t.assert_equals(count_workers(), 0)
+    end)
+end
+
+g.test_graphite_stop_custom_fiber = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local fun = require('fun')
+        local graphite = require('metrics.plugins.graphite')
+
+        local function count_workers()
+            return fun.iter(fiber.info()):
+                filter(function(_, x) return string.find(x.name, 'metrics_graphite_worker') end):
+                length()
+        end
+
+        local opts = {
+            prefix = "master",
+            host = "127.0.0.1",
+            port = 3333,
+            send_interval = 1,
+        }
+
+        t.assert_equals(count_workers(), 0)
+
+        graphite.init(opts)
+
+        t.assert_equals(count_workers(), 1)
+
+        graphite.stop()
+        t.assert_equals(count_workers(), 1) -- fiber still works
+
+        graphite.stop(opts)
+        fiber.yield() -- let cancelled fibers disappear from fiber.info()
+        t.assert_equals(count_workers(), 0)
+    end)
+end