[plugin/datadog] Logging to statsd server

Compiles metrics like Request count, size, Response status and latency and send it to
Datadog statsd server

kong-0.5.4-1.rockspec

@@ -230,7 +230,13 @@ 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.log"] = "kong/plugins/datadog/log.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,32 @@
+local log = require "kong.plugins.datadog.log"
+local BasePlugin = require "kong.plugins.base_plugin"
+local basic_serializer = require "kong.plugins.log-serializers.basic"
+
+local ngx_now = ngx.now
+
+local DatadogHandler = BasePlugin:extend()
+
+function DatadogHandler:new()
+  DatadogHandler.super.new(self, "datadog")
+end
+
+function DatadogHandler:body_filter(conf)
+  DatadogHandler.super.body_filter(self)
+
+  ngx.ctx.datadog = {}
+  local eof = ngx.arg[2]
+  if eof then -- latest chunk
+    ngx.ctx.datadog.response_received = ngx_now() * 1000
+  end
+end
+
+function DatadogHandler:log(conf)
+  DatadogHandler.super.log(self)
+  local message = basic_serializer.serialize(ngx)
+  message.response.response_received = ngx.ctx.datadog.response_received
+  log.execute(conf, message)
+end
+
+DatadogHandler.PRIORITY = 1
+
+return DatadogHandler

kong/plugins/datadog/log.lua

@@ -0,0 +1,71 @@
+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 _M = {}
+
+local METRICS = { "request_count", "latency", "request_size", "status_count" }
+
+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 latency = message.response.response_received - message.started_at
+  local stat = api_name..".latency"
+  logger:gauge(stat, latency, 1)
+end
+
+local function log(premature, conf, message, logger)
+
+  local logger, err = statsd_logger:new(conf)
+  if err then
+    ngx_log(ngx.ERR, "failed to create Statsd logger: ", err)
+    return
+  end
+
+  local metrics = conf.metrics
+  if not conf.metrics then
+    metrics = METRICS
+  end
+
+  local api_name = string_gsub(message.api.name, "%.", "_")
+  for _, metric in pairs(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
+
+function _M.execute(conf, message)
+  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 _M

kong/plugins/datadog/schema.lua

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

kong/plugins/datadog/statsd_logger.lua

@@ -0,0 +1,86 @@
+local setmetatable = setmetatable
+local ngx_socket_udp = ngx.socket.udp
+local ngx_log = ngx.log
+local table_concat = table.concat
+local setmetatable = setmetatable
+
+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
+     ngx_log(ngx.ERR, "failed to connect to "..conf.host..":"..tostring(conf.port)..": ", err)
+    return
+  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,86 @@
+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" }
+      },
+      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 },
+      }
+    }
+    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.are.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(#message, 1)
+  end)
+end)