plugin/statsd (feature) Statsd logger

kong-0.8.0rc2-0.rockspec

@@ -263,7 +263,11 @@ build = {
 
     ["kong.plugins.datadog.handler"] = "kong/plugins/datadog/handler.lua",
     ["kong.plugins.datadog.schema"] = "kong/plugins/datadog/schema.lua",
-    ["kong.plugins.datadog.statsd_logger"] = "kong/plugins/datadog/statsd_logger.lua"
+    ["kong.plugins.datadog.statsd_logger"] = "kong/plugins/datadog/statsd_logger.lua",
+
+    ["kong.plugins.statsd.handler"] = "kong/plugins/statsd/handler.lua",
+    ["kong.plugins.statsd.schema"] = "kong/plugins/statsd/schema.lua",
+    ["kong.plugins.statsd.statsd_logger"] = "kong/plugins/statsd/statsd_logger.lua"
 
   },
   install = {

kong/constants.lua

@@ -18,7 +18,7 @@ return {
     "http-log", "key-auth", "hmac-auth", "basic-auth", "ip-restriction",
     "mashape-analytics", "request-transformer", "response-transformer",
     "request-size-limiting", "rate-limiting", "response-ratelimiting", "syslog",
-    "loggly", "datadog", "runscope", "ldap-auth"
+    "loggly", "datadog", "runscope", "ldap-auth", "statsd"
   },
   -- Non standard headers, specific to Kong
   HEADERS = {

kong/plugins/statsd/handler.lua

@@ -0,0 +1,84 @@
+local BasePlugin = require "kong.plugins.base_plugin"
+local basic_serializer = require "kong.plugins.log-serializers.basic"
+local statsd_logger = require "kong.plugins.statsd.statsd_logger"
+
+local StatsdHandler = BasePlugin:extend()
+
+StatsdHandler.PRIORITY = 1
+
+local ngx_log = ngx.log
+local ngx_timer_at = ngx.timer.at
+local string_gsub = string.gsub
+local pairs = pairs
+local NGX_ERR = ngx.ERR
+
+local gauges = {
+  request_size = function (api_name, message, logger)
+    local stat = api_name..".request.size"
+    logger:gauge(stat, message.request.size, 1)
+  end,
+  response_size = function (api_name, message, logger)
+    local stat = api_name..".response.size"
+    logger:gauge(stat, message.response.size, 1)
+  end,
+  status_count = function (api_name, message, logger)
+    local stat = api_name..".request.status."..message.response.status
+    logger:counter(stat, 1, 1)
+  end,
+  latency = function (api_name, message, logger)
+    local stat = api_name..".latency"
+    logger:gauge(stat, message.latencies.request, 1)
+  end,
+  request_count = function (api_name, message, logger)
+    local stat = api_name..".request.count"
+    logger:counter(stat, 1, 1)
+  end,
+  unique_users = function (api_name, message, logger)
+    if message.authenticated_entity ~= nil and message.authenticated_entity.consumer_id ~= nil then
+      local stat = api_name..".user.uniques"
+      logger:set(stat, message.authenticated_entity.consumer_id)
+    end
+  end,
+  request_per_user = function (api_name, message, logger)
+    if message.authenticated_entity ~= nil and message.authenticated_entity.consumer_id ~= nil then
+      local stat = api_name.."."..string_gsub(message.authenticated_entity.consumer_id, "-", "_")..".request.count"
+      logger:counter(stat, 1, 1)    
+    end
+  end
+}
+
+local function log(premature, conf, message)
+  if premature then return end
+
+  local logger, err = statsd_logger:new(conf)
+  if err then
+    ngx_log(NGX_ERR, "failed to create Statsd logger: ", err)
+    return
+  end
+
+  local api_name = string_gsub(message.api.name, "%.", "_")
+  for _, metric in pairs(conf.metrics) do
+    local gauge = gauges[metric]
+    if gauge ~= nil then
+      gauge(api_name, message, logger)
+    end
+  end
+
+  logger:close_socket()
+end
+
+function StatsdHandler:new()
+  StatsdHandler.super.new(self, "statsd")
+end
+
+function StatsdHandler:log(conf)
+  StatsdHandler.super.log(self)
+  local message = basic_serializer.serialize(ngx)
+
+  local ok, err = ngx_timer_at(0, log, conf, message)
+  if not ok then
+    ngx_log(NGX_ERR, "failed to create timer: ", err)
+  end
+end
+
+return StatsdHandler

kong/plugins/statsd/schema.lua

@@ -0,0 +1,8 @@
+return {
+  fields = {
+    host = {required = true, type = "string", default = "localhost"},
+    port = {required = true, type = "number", default = 8125},
+    metrics = {required = true, type = "array", enum = {"request_count", "latency", "request_size", "status_count", "response_size", "unique_users", "request_per_user"}, default = {"request_count", "latency", "request_size", "status_count", "response_size", "unique_users", "request_per_user"}}, 
+    timeout = {type = "number", default = 10000}
+  }
+}

kong/plugins/statsd/statsd_logger.lua

@@ -0,0 +1,86 @@
+local ngx_socket_udp = ngx.socket.udp
+local ngx_log = ngx.log
+local table_concat = table.concat
+local setmetatable = setmetatable
+local NGX_ERR = ngx.ERR
+local NGX_DEBUG = ngx.DEBUG
+
+local statsd_mt = {}
+statsd_mt.__index = statsd_mt
+
+function statsd_mt:new(conf)
+  local sock = ngx_socket_udp()
+  sock:settimeout(conf.timeout)
+  local _, err = sock:setpeername(conf.host, conf.port)
+  if err then
+    return nil, "failed to connect to "..conf.host..":"..tostring(conf.port)..": "..err
+  end
+
+  local statsd = {
+    host = conf.host,
+    port = conf.port,
+    socket = sock
+  }
+  return setmetatable(statsd, statsd_mt)
+end
+
+function statsd_mt:create_statsd_message(stat, delta, kind, sample_rate)
+  local rate = ""
+  if sample_rate and sample_rate ~= 1 then 
+    rate = "|@"..sample_rate 
+  end
+
+  local message = {
+    "kong.",
+    stat,
+    ":",
+    delta,
+    "|",
+    kind,
+    rate
+  }
+  return table_concat(message, "")
+end
+
+function statsd_mt:close_socket()
+  local ok, err = self.socket:close()
+  if not ok then
+    ngx_log(NGX_ERR, "failed to close connection from "..self.host..":"..tostring(self.port)..": ", err)
+    return
+  end
+end
+
+function statsd_mt:send_statsd(stat, delta, kind, sample_rate)
+  local udp_message = self:create_statsd_message(stat, delta, kind, sample_rate)
+  ngx_log(NGX_DEBUG, "Sending data to statsd server: "..udp_message)
+  local ok, err = self.socket:send(udp_message)
+  if not ok then
+    ngx_log(NGX_ERR, "failed to send data to "..self.host..":"..tostring(self.port)..": ", err)
+  end
+end
+
+function statsd_mt:gauge(stat, value, sample_rate)
+  return self:send_statsd(stat, value, "g", sample_rate)
+end
+
+function statsd_mt:counter(stat, value, sample_rate)
+  return self:send_statsd(stat, value, "c", sample_rate)
+end
+
+function statsd_mt:timer(stat, ms)
+  return self:send_statsd(stat, ms, "ms")
+end
+
+function statsd_mt:histogram(stat, value)
+  return self:send_statsd(stat, value, "h")
+end
+
+function statsd_mt:meter(stat, value)
+  return self:send_statsd(stat, value, "m")
+end
+
+function statsd_mt:set(stat, value)
+  return self:send_statsd(stat, value, "s")
+end
+
+return statsd_mt

spec/plugins/statsd/log_spec.lua

@@ -0,0 +1,122 @@
+local spec_helper = require "spec.spec_helpers"
+local http_client = require "kong.tools.http_client"
+
+local STUB_GET_URL = spec_helper.STUB_GET_URL
+
+local UDP_PORT = spec_helper.find_port()
+
+describe("Statsd Plugin", function()
+
+  setup(function()
+    spec_helper.prepare_db()
+    spec_helper.insert_fixtures {
+      api = {
+        {request_host = "logging1.com", upstream_url = "http://mockbin.com"},
+        {request_host = "logging2.com", upstream_url = "http://mockbin.com"},
+        {request_host = "logging3.com", upstream_url = "http://mockbin.com"},
+        {request_host = "logging4.com", upstream_url = "http://mockbin.com"},
+        {request_host = "logging5.com", upstream_url = "http://mockbin.com"},
+        {request_host = "logging6.com", upstream_url = "http://mockbin.com"}
+      },
+      plugin = {
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"request_count"}}, __api = 1},
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"latency"}}, __api = 2},
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"status_count"}}, __api = 3},
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"request_size"}}, __api = 4},
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT}, __api = 5},
+        {name = "statsd", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"response_size"}}, __api = 6}
+      }
+    }
+    spec_helper.start_kong()
+  end)
+
+  teardown(function()
+    spec_helper.stop_kong()
+  end)
+
+  it("should log to UDP when metrics is request_count", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging1.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+    assert.equal("kong.logging1_com.request.count:1|c", res)
+  end)
+
+  it("should log to UDP when metrics is status_count", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging3.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+    assert.equal("kong.logging3_com.request.status.200:1|c", res)
+  end)
+
+  it("should log to UDP when metrics is request_size", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging4.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+    local message = {}
+    for w in string.gmatch(res,"kong.logging4_com.request.size:%d*|g") do
+      table.insert(message, w)
+    end
+    assert.equal(1, #message)
+  end)
+
+  it("should log to UDP when metrics is latency", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging2.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+
+    local message = {}
+    for w in string.gmatch(res,"kong.logging2_com.latency:.*|g") do
+      table.insert(message, w)
+    end
+
+    assert.equal(1, #message)
+  end)
+
+  it("should log to UDP when metrics is request_count", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging5.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+    assert.equal("kong.logging5_com.request.count:1|c", res)
+  end)
+
+  it("should log to UDP when metrics is response_size", function()
+    local thread = spec_helper.start_udp_server(UDP_PORT) -- Starting the mock UDP server
+
+    local _, status = http_client.get(STUB_GET_URL, nil, {host = "logging6.com"})
+    assert.equal(200, status)
+
+    local ok, res = thread:join()
+    assert.True(ok)
+    assert.truthy(res)
+    local message = {}
+    for w in string.gmatch(res,"kong.logging6_com.response.size:%d*|g") do
+      table.insert(message, w)
+    end
+    assert.equal(1, #message)
+  end)
+end)