Merge pull request #758 from Mashape/feat/dd-statsd-logger

[plugin/datadog] Logging to statsd server

kong-0.5.4-1.rockspec

@@ -230,7 +230,12 @@ build = {
 
     ["kong.plugins.loggly.handler"] = "kong/plugins/loggly/handler.lua",
     ["kong.plugins.loggly.log"] = "kong/plugins/loggly/log.lua",
-    ["kong.plugins.loggly.schema"] = "kong/plugins/loggly/schema.lua"
+    ["kong.plugins.loggly.schema"] = "kong/plugins/loggly/schema.lua",
+
+    ["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"
+
   },
   install = {
     conf = { "kong.yml" },

kong/plugins/datadog/handler.lua

@@ -0,0 +1,77 @@
+local BasePlugin = require "kong.plugins.base_plugin"
+local basic_serializer = require "kong.plugins.log-serializers.basic"
+local statsd_logger = require "kong.plugins.datadog.statsd_logger"
+
+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 function request_counter(api_name, logger)
+  local stat = api_name..".request.count"
+  logger:counter(stat, 1, 1)
+end
+
+local function status_counter(api_name, message, logger)
+  local stat = api_name..".request.status."..message.response.status
+  logger:counter(stat, 1, 1)
+end
+
+local function request_size_gauge(api_name, message, logger)
+  local stat = api_name..".request.size"
+  logger:gauge(stat, message.request.size, 1)
+end
+
+local function latency_gauge(api_name, message, logger)
+  local stat = api_name..".latency"
+  logger:gauge(stat, message.latencies.request, 1)
+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
+    if metric == "request_size" then
+      request_size_gauge(api_name, message, logger)
+    end
+    if metric == "status_count" then
+      status_counter(api_name, message, logger)
+    end
+    if metric == "latency" then
+      latency_gauge(api_name, message, logger)
+    end
+    if metric == "request_count" then
+      request_counter(api_name, logger)
+    end
+  end
+
+  logger:close_socket()
+end
+
+local DatadogHandler = BasePlugin:extend()
+
+function DatadogHandler:new()
+  DatadogHandler.super.new(self, "datadog")
+end
+
+function DatadogHandler:log(conf)
+  DatadogHandler.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
+
+DatadogHandler.PRIORITY = 1
+
+return DatadogHandler

kong/plugins/datadog/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"}, default = {"request_count", "latency", "request_size", "status_count"}}, 
+    timeout = {type = "number", default = 10000}
+  }
+}

kong/plugins/datadog/statsd_logger.lua

@@ -0,0 +1,85 @@
+local setmetatable = setmetatable
+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 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)
+  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

kong/tools/config_defaults.lua

@@ -3,7 +3,7 @@ return {
     default = {"ssl", "jwt", "acl", "cors", "oauth2", "tcp-log", "udp-log", "file-log",
                "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"}
+               "request-size-limiting", "rate-limiting", "response-ratelimiting", "syslog", "loggly", "datadog"}
   },
   ["nginx_working_dir"] = {type = "string", default = "/usr/local/kong"},
   ["proxy_port"] = {type = "number", default = 8000},

spec/plugins/datadog/log_spec.lua

@@ -0,0 +1,100 @@
+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("Datadog 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"}
+      },
+      plugin = {
+        {name = "datadog", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"request_count"}}, __api = 1},
+        {name = "datadog", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"latency"}}, __api = 2},
+        {name = "datadog", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"status_count"}}, __api = 3},
+        {name = "datadog", config = {host = "127.0.0.1", port = UDP_PORT, metrics = {"request_size"}}, __api = 4},
+        {name = "datadog", config = {host = "127.0.0.1", port = UDP_PORT}, __api = 5}
+      }
+    }
+    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)
+    assert.equal("kong.logging4_com.request.size:111|g", res)
+  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)
+end)