feat(plugin/datadog) added new metric upstream_latency and tagging su…

…pport (#1473)

kong/plugins/datadog/handler.lua

@@ -11,54 +11,58 @@ local string_gsub = string.gsub
 local ipairs = ipairs
 
 local gauges = {
-  request_size = function (api_name, message, logger)
+  request_size = function (api_name, message, logger, tags)
     local stat = api_name..".request.size"
-    logger:gauge(stat, message.request.size, 1)
+    logger:gauge(stat, message.request.size, 1, tags)
   end,
-  response_size = function (api_name, message, logger)
+  response_size = function (api_name, message, logger, tags)
     local stat = api_name..".response.size"
-    logger:gauge(stat, message.response.size, 1)
+    logger:gauge(stat, message.response.size, 1, tags)
   end,
-  status_count = function (api_name, message, logger)
+  status_count = function (api_name, message, logger, tags)
     local stat = api_name..".request.status."..message.response.status
-    logger:counter(stat, 1, 1)
+    logger:counter(stat, 1, 1, tags)
   end,
-  latency = function (api_name, message, logger)
+  latency = function (api_name, message, logger, tags)
     local stat = api_name..".latency"
-    logger:gauge(stat, message.latencies.request, 1)
+    logger:gauge(stat, message.latencies.request, 1, tags)
   end,
-  request_count = function (api_name, message, logger)
+  request_count = function (api_name, message, logger, tags)
     local stat = api_name..".request.count"
-    logger:counter(stat, 1, 1)
+    logger:counter(stat, 1, 1, tags)
   end,
-  unique_users = function (api_name, message, logger)
+  unique_users = function (api_name, message, logger, tags)
     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)
+      logger:set(stat, message.authenticated_entity.consumer_id, tags)
     end
   end,
-  request_per_user = function (api_name, message, logger)
+  request_per_user = function (api_name, message, logger, tags)
     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)
+      logger:counter(stat, 1, 1, tags)
     end
+  end,
+  upstream_latency = function (api_name, message, logger, tags)
+    local stat = api_name..".upstream_latency"
+    logger:gauge(stat, message.latencies.proxy, 1, tags)
   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 ipairs(conf.metrics) do
     local gauge = gauges[metric]
     if gauge then
-      gauge(api_name, message, logger)
+
+      gauge(api_name, message, logger, conf.tags[metric])
     end
   end
 

kong/plugins/datadog/schema.lua

@@ -1,22 +1,47 @@
+local find = string.find
+local pl_utils = require "pl.utils"
 local metrics = {
   "request_count",
   "latency",
   "request_size",
   "status_count",
   "response_size",
   "unique_users",
-  "request_per_user"
+  "request_per_user",
+  "upstream_latency"
 }
-
+-- entries must have colons to set the key and value apart
+local function check_for_value(value)
+  for i, entry in ipairs(value) do
+    local ok = find(entry, ":")
+    if ok then 
+      local _,next = pl_utils.splitv(entry, ':')
+      if not next or #next == 0 then
+        return false, "key '"..entry.."' has no value, "
+      end
+    end
+  end
+  return true
+end
 return {
   fields = {
     host = {required = true, type = "string", default = "localhost"},
     port = {required = true, type = "number", default = 8125},
-    metrics = {
-      required = true,
-      type = "array",
-      enum = metrics,
-      default = metrics
+    metrics = {required = true, type = "array", enum = metrics, default = metrics},
+    tags = {
+      type = "table",
+      schema = {
+        fields = {
+          request_count = {type = "array", default = {}, func = check_for_value},
+          latency = {type = "array", default = {}, func = check_for_value},
+          request_size = {type = "array", default = {}, func = check_for_value},
+          status_count = {type = "array", default = {}, func = check_for_value},
+          response_size = {type = "array", default = {}, func = check_for_value},
+          unique_users = {type = "array", default = {}, func = check_for_value},
+          request_per_user = {type = "array", default = {}, func = check_for_value},
+          upstream_latency = {type = "array", default = {}, func = check_for_value}
+        }
+      }
     },
     timeout = {type = "number", default = 10000}
   }

kong/plugins/datadog/statsd_logger.lua

@@ -25,11 +25,16 @@ function statsd_mt:new(conf)
   return setmetatable(statsd, statsd_mt)
 end
 
-function statsd_mt:create_statsd_message(stat, delta, kind, sample_rate)
+function statsd_mt:create_statsd_message(stat, delta, kind, sample_rate, tags)
   local rate = ""
+  local str_tags = ""
   if sample_rate and sample_rate ~= 1 then
     rate = "|@"..sample_rate
   end
+
+  if tags and #tags > 0 then
+    str_tags = "|#"..table_concat(tags, ",")
+  end
 
   local message = {
     "kong.",
@@ -38,7 +43,8 @@ function statsd_mt:create_statsd_message(stat, delta, kind, sample_rate)
     delta,
     "|",
     kind,
-    rate
+    rate,
+    str_tags
   }
   return table_concat(message, "")
 end
@@ -50,8 +56,8 @@ function statsd_mt:close_socket()
   end
 end
 
-function statsd_mt:send_statsd(stat, delta, kind, sample_rate)
-  local udp_message = self:create_statsd_message(stat, delta, kind, sample_rate)
+function statsd_mt:send_statsd(stat, delta, kind, sample_rate, tags)
+  local udp_message = self:create_statsd_message(stat, delta, kind, sample_rate, tags)
 
   ngx_log(NGX_DEBUG, "[udp-log] sending data to statsd server: ", udp_message)
 
@@ -61,28 +67,28 @@ function statsd_mt:send_statsd(stat, delta, kind, sample_rate)
   end
 end
 
-function statsd_mt:gauge(stat, value, sample_rate)
-  return self:send_statsd(stat, value, "g", sample_rate)
+function statsd_mt:gauge(stat, value, sample_rate, tags)
+  return self:send_statsd(stat, value, "g", sample_rate, tags)
 end
 
-function statsd_mt:counter(stat, value, sample_rate)
-  return self:send_statsd(stat, value, "c", sample_rate)
+function statsd_mt:counter(stat, value, sample_rate, tags)
+  return self:send_statsd(stat, value, "c", sample_rate, tags)
 end
 
-function statsd_mt:timer(stat, ms)
-  return self:send_statsd(stat, ms, "ms")
+function statsd_mt:timer(stat, ms, tags)
+  return self:send_statsd(stat, ms, "ms", nil, tags)
 end
 
-function statsd_mt:histogram(stat, value)
-  return self:send_statsd(stat, value, "h")
+function statsd_mt:histogram(stat, value, tags)
+  return self:send_statsd(stat, value, "h", nil, tags)
 end
 
-function statsd_mt:meter(stat, value)
-  return self:send_statsd(stat, value, "m")
+function statsd_mt:meter(stat, value, tags)
+  return self:send_statsd(stat, value, "m", nil, tags)
 end
 
-function statsd_mt:set(stat, value)
-  return self:send_statsd(stat, value, "s")
+function statsd_mt:set(stat, value, tags)
+  return self:send_statsd(stat, value, "s", nil, tags)
 end
 
 return statsd_mt

spec/03-plugins/008-datadog/01-log_spec.lua

@@ -9,13 +9,15 @@ describe("Plugin: datadog (log)", function()
 
     local api1 = assert(helpers.dao.apis:insert {request_host = "datadog1.com", upstream_url = "http://mockbin.com"})
     local api2 = assert(helpers.dao.apis:insert {request_host = "datadog2.com", upstream_url = "http://mockbin.com"})
+    local api3 = assert(helpers.dao.apis:insert {request_host = "datadog3.com", upstream_url = "http://mockbin.com"})
 
     assert(helpers.dao.plugins:insert {
       name = "datadog",
       api_id = api1.id,
       config = {
         host = "127.0.0.1",
-        port = 9999
+        port = 9999,
+        tags = {}
       }
     })
     assert(helpers.dao.plugins:insert {
@@ -24,9 +26,24 @@ describe("Plugin: datadog (log)", function()
       config = {
         host = "127.0.0.1",
         port = 9999,
-        metrics = "request_count,status_count"
+        metrics = "request_count,status_count",
+        tags = {}
       }
     })
+    assert(helpers.dao.plugins:insert {
+      name = "datadog",
+      api_id = api3.id,
+      config = {
+        host = "127.0.0.1",
+        port = 9999,
+        metrics = "request_count,status_count,latency",
+        tags = {
+          request_count = {"T1:V1"},
+          status_count = {"T2:V2,T3:V3,T4"},
+          latency = {"T2:V2:V3,T4"},
+        }
+      }  
+    })
   end)
   teardown(function()
     if client then client:close() end
@@ -42,7 +59,7 @@ describe("Plugin: datadog (log)", function()
         server:setoption("reuseaddr", true)
         server:setsockname("127.0.0.1", 9999)
         local gauges = {}
-        for i = 1, 5 do
+        for i = 1, 6 do
           gauges[#gauges+1] = server:receive()
         end
         server:close()
@@ -62,12 +79,13 @@ describe("Plugin: datadog (log)", function()
 
     local ok, gauges = thread:join()
     assert.True(ok)
-    assert.equal(5, #gauges)
+    assert.equal(6, #gauges)
     assert.contains("kong.datadog1_com.request.count:1|c", gauges)
     assert.contains("kong.datadog1_com.latency:%d+|g", gauges, true)
     assert.contains("kong.datadog1_com.request.size:%d+|g", gauges, true)
     assert.contains("kong.datadog1_com.request.status.200:1|c", gauges)
     assert.contains("kong.datadog1_com.response.size:%d+|g", gauges, true)
+    assert.contains("kong.datadog1_com.upstream_latency:%d+|g", gauges, true)
   end)
 
   it("logs only given metrics", function()
@@ -103,4 +121,39 @@ describe("Plugin: datadog (log)", function()
     assert.contains("kong.datadog2_com.request.count:1|c", gauges)
     assert.contains("kong.datadog2_com.request.status.200:1|c", gauges)
   end)
+
+  it("logs metrics with tags", function()
+    local thread = threads.new({
+      function()
+        local socket = require "socket"
+        local server = assert(socket.udp())
+        server:settimeout(1)
+        server:setoption("reuseaddr", true)
+        server:setsockname("127.0.0.1", 9999)
+        local gauges = {}
+        for i = 1, 3 do
+          gauges[#gauges+1] = server:receive()
+        end
+        server:close()
+        return gauges
+      end
+    })
+    thread:start()
+
+    local res = assert(client:send {
+      method = "GET",
+      path = "/status/200",
+      headers = {
+        ["Host"] = "datadog3.com"
+      }
+    })
+    assert.res_status(200, res)
+
+    local ok, gauges = thread:join()
+    assert.True(ok)
+    assert.equal(3, #gauges)
+    assert.contains("kong.datadog3_com.request.count:1|c|#T1:V1", gauges)
+    assert.contains("kong.datadog3_com.request.status.200:1|c|#T2:V2,T3:V3,T4", gauges)
+    assert.contains("kong.datadog3_com.latency:%d+|g|#T2:V2:V3,T4", gauges, true)
+  end)
 end)