feat(plugins/datadog) new metrics and more flexible configuration

- Now each metric supports configurable stat type,
  sample rate and customer identifier if applicable.
- Custom tags for Datadog.
- New metrics `upstream_latency`, `kong_latency` and `status_count_per_user`.

kong-0.10.1-0.rockspec

@@ -260,6 +260,8 @@ build = {
     ["kong.plugins.loggly.handler"] = "kong/plugins/loggly/handler.lua",
     ["kong.plugins.loggly.schema"] = "kong/plugins/loggly/schema.lua",
 
+    ["kong.plugins.datadog.migrations.cassandra"] = "kong/plugins/datadog/migrations/cassandra.lua",
+    ["kong.plugins.datadog.migrations.postgres"] = "kong/plugins/datadog/migrations/postgres.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",

kong/plugins/datadog/handler.lua

@@ -9,60 +9,122 @@ DatadogHandler.PRIORITY = 1
 local ngx_timer_at = ngx.timer.at
 local string_gsub = string.gsub
 local ipairs = ipairs
+local NGX_ERR = ngx.ERR
+local ngx_log = ngx.log
+
+local function allow_user_metric(message, identifier)
+  if message.consumer ~= nil and
+          message.consumer[identifier] ~= nil then
+    return true, message.consumer[identifier]
+  end
+  return false
+end
 
 local gauges = {
-  request_size = function (api_name, message, logger, tags)
-    local stat = api_name..".request.size"
-    logger:gauge(stat, message.request.size, 1, tags)
+  gauge = function (stat_name, stat_value, metric_config, logger)
+    logger:gauge(stat_name, stat_value, metric_config.sample_rate, metric_config.tags)
   end,
-  response_size = function (api_name, message, logger, tags)
-    local stat = api_name..".response.size"
-    logger:gauge(stat, message.response.size, 1, tags)
+  timer = function (stat_name, stat_value, metric_config, logger)
+    logger:timer(stat_name, stat_value, metric_config.tags)
   end,
-  status_count = function (api_name, message, logger, tags)
-    local stat = api_name..".request.status."..message.response.status
-    logger:counter(stat, 1, 1, tags)
+  counter = function (stat_name, _, metric_config, logger)
+    logger:counter(stat_name, 1, metric_config.sample_rate, metric_config.tags)
   end,
-  latency = function (api_name, message, logger, tags)
-    local stat = api_name..".latency"
-    logger:gauge(stat, message.latencies.request, 1, tags)
+  set = function (stat_name, stat_value, metric_config, logger)
+    logger:set(stat_name, stat_value, metric_config.tags)
   end,
-  request_count = function (api_name, message, logger, tags)
-    local stat = api_name..".request.count"
-    logger:counter(stat, 1, 1, tags)
+  histogram = function (stat_name, stat_value, metric_config, logger)
+    logger:histogram(stat_name, stat_value, metric_config.tags)
   end,
-  unique_users = function (api_name, message, logger, tags)
-    if message.authenticated_entity ~= nil and message.authenticated_entity.consumer_id ~= nil then
+  meter = function (stat_name, stat_value, metric_config, logger)
+    logger:meter(stat_name, stat_value, metric_config.tags)
+  end,
+  status_count = function (api_name, message, metric_config, logger)
+    local stat = api_name..".request.status."..message.response.status
+    local total_count = api_name..".request.status.total"
+    local sample_rate = metric_config.sample_rate
+    logger:counter(stat, 1, sample_rate, metric_config.tags)
+    logger:counter(total_count, 1, sample_rate, metric_config.tags)
+  end,
+  unique_users = function (api_name, message, metric_config, logger)
+    local identifier = metric_config.consumer_identifier
+    local allow_metric, cust_id = allow_user_metric(message, identifier)
+    if allow_metric then
       local stat = api_name..".user.uniques"
-      logger:set(stat, message.authenticated_entity.consumer_id, tags)
+      logger:set(stat, cust_id, metric_config.tags)
     end
   end,
-  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, tags)
+  request_per_user = function (api_name, message, metric_config, logger)
+    local identifier = metric_config.consumer_identifier
+    local allow_metric, cust_id = allow_user_metric(message, identifier)
+    if allow_metric then
+      local sample_rate = metric_config.sample_rate
+      local stat = api_name..".user."..string_gsub(cust_id, "-", "_")
+              ..".request.count"
+      logger:counter(stat, 1, sample_rate, metric_config.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)
+  status_count_per_user = function (api_name, message, metric_config, logger)
+    local identifier = metric_config.consumer_identifier
+    local allow_metric, cust_id = allow_user_metric(message, identifier)
+    if allow_metric then
+      local stat = api_name..".user."..string_gsub(cust_id, "-", "_")
+              ..".request.status."..message.response.status
+      local total_count = api_name..".user."..string_gsub(cust_id, "-", "_")
+              ..".request.status.total"
+      local sample_rate = metric_config.sample_rate
+      logger:counter(stat, 1, sample_rate, metric_config.tags)
+      logger:counter(total_count, 1, sample_rate, metric_config.tags)
+    end
   end
 }
 
 local function log(premature, conf, message)
   if premature then return end
+
+  local api_name = string_gsub(message.api.name, "%.", "_")
+
+  local stat_name = {
+    request_size = api_name..".request.size",
+    response_size = api_name..".response.size",
+    latency = api_name..".latency",
+    upstream_latency = api_name..".upstream_latency",
+    kong_latency = api_name..".kong_latency",
+    request_count = api_name..".request.count"
+  }
+
+  local stat_value = {
+    request_size = message.request.size,
+    response_size = message.response.size,
+    latency = message.latencies.request,
+    upstream_latency = message.latencies.proxy,
+    kong_latency = message.latencies.kong,
+    request_count = 1
+  }
+
   local logger, err = statsd_logger:new(conf)
+
   if err then
-    ngx.log(ngx.ERR, "failed to create Statsd logger: ", err)
+    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, conf.tags[metric])
+  for _, metric_config in ipairs(conf.metrics) do
+    if metric_config.name ~= "status_count"
+            and metric_config.name ~= "unique_users"
+            and metric_config.name ~= "request_per_user"
+            and metric_config.name ~= "status_count_per_user" then
+      local stat_name = stat_name[metric_config.name]
+      local stat_value = stat_value[metric_config.name]
+      local gauge = gauges[metric_config.stat_type]
+      if stat_name ~= nil and gauge ~= nil and stat_value ~= nil then
+        gauge(stat_name, stat_value, metric_config, logger)
+      end
+
+    else
+      local gauge = gauges[metric_config.name]
+      if gauge ~= nil then
+        gauge(api_name, message, metric_config, logger)
+      end
     end
   end
 
@@ -79,7 +141,7 @@ function DatadogHandler:log(conf)
 
   local ok, err = ngx_timer_at(0, log, conf, message)
   if not ok then
-    ngx.log(ngx.ERR, "failed to create timer: ", err)
+    ngx_log(NGX_ERR, "failed to create timer: ", err)
   end
 end
 

kong/plugins/datadog/migrations/cassandra.lua

@@ -0,0 +1,89 @@
+return {
+  {
+    name = "2017-02-09-160000_datadog_schema_changes",
+    up = function(_, _, factory)
+
+      local plugins, err = factory.plugins:find_all {name = "datadog"}
+      if err then
+        return err
+      end
+
+      local default_metrics = {
+        request_count = {
+          name = "request_count",
+          stat_type = "counter",
+          sample_rate = 1,
+          tags = {"app:kong"}
+        },
+        latency = {
+          name = "latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        request_size = {
+          name = "request_size",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        status_count = {
+          name = "status_count",
+          stat_type = "counter",
+          sample_rate = 1,
+          tags = {"app:kong"}
+        },
+        response_size = {
+          name = "response_size",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        unique_users = {
+          name = "unique_users",
+          stat_type = "set",
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        },
+        request_per_user = {
+          name = "request_per_user",
+          stat_type = "counter",
+          sample_rate = 1,
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        },
+        upstream_latency = {
+          name = "upstream_latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        kong_latency = {
+          name = "kong_latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        status_count_per_user = {
+          name = "status_count_per_user",
+          stat_type = "counter",
+          sample_rate = 1,
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        }
+      }
+
+      for _, plugin in ipairs(plugins) do
+        local new_metrics = {}
+        plugin.config.tags = nil
+        plugin.config.timeout = nil
+        if plugin.config.metrics ~=nil then
+          for _, metric in ipairs(plugin.config.metrics) do
+            table.insert(new_metrics, default_metrics[metric])
+          end
+          plugin.config.metrics = new_metrics
+          plugin.config.timeout = nil
+          local _, err = factory.plugins:update(plugin, plugin, {full = true})
+          if err then
+            return err
+          end
+        end
+      end
+    end
+  }
+}

kong/plugins/datadog/migrations/postgres.lua

@@ -0,0 +1,95 @@
+return {
+  {
+    name = "2017-02-09-160000_datadog_schema_changes",
+    up = function(_, _, dao)
+      local rows, err = dao.plugins:find_all {name = "datadog"}
+      if err then return err end
+
+      local default_metrics = {
+        request_count = {
+          name = "request_count",
+          stat_type = "counter",
+          sample_rate = 1,
+          tags = {"app:kong"}
+        },
+        latency = {
+          name = "latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        request_size = {
+          name = "request_size",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        status_count = {
+          name = "status_count",
+          stat_type = "counter",
+          sample_rate = 1,
+          tags = {"app:kong"}
+        },
+        response_size = {
+          name = "response_size",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        unique_users = {
+          name = "unique_users",
+          stat_type = "set",
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        },
+        request_per_user = {
+          name = "request_per_user",
+          stat_type = "counter",
+          sample_rate = 1,
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        },
+        upstream_latency = {
+          name = "upstream_latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        kong_latency = {
+          name = "kong_latency",
+          stat_type = "timer",
+          tags = {"app:kong"}
+        },
+        status_count_per_user = {
+          name = "status_count_per_user",
+          stat_type = "counter",
+          sample_rate = 1,
+          consumer_identifier = "custom_id",
+          tags = {"app:kong"}
+        }
+      }
+
+
+      for i = 1, #rows do
+        local row = rows[i]
+
+        local _, err = dao.plugins:delete(row)
+        if err then return err end
+        local new_metrics = {}
+        if row.config.metrics ~=nil then
+          for _, metric in ipairs(row.config.metrics) do
+            table.insert(new_metrics, default_metrics[metric])
+          end
+        end
+
+        local _, err = dao.plugins:insert {
+          name = "datadog",
+          api_id = row.api_id,
+          enabled = row.enabled,
+          config = {
+            host = row.config.host,
+            port = row.config.port,
+            metrics = new_metrics
+          }
+        }
+        if err then return err end
+      end
+    end
+  }
+}