kong--kong/kong/plugins/statsd/log.lua

local Queue = require "kong.tools.queue"
local constants = require "kong.plugins.statsd.constants"
local statsd_logger = require "kong.plugins.statsd.statsd_logger"
local ws = require "kong.workspaces"

local ngx = ngx
local kong = kong
local ngx_time = ngx.time
local re_gsub = ngx.re.gsub
local pairs = pairs
local ipairs = ipairs
local string_format = string.format
local match = ngx.re.match
local ipairs = ipairs
local tonumber = tonumber
local knode = kong and kong.node or require "kong.pdk.node".new()
local null = ngx.null

local START_RANGE_IDX = 1
local END_RANGE_IDX   = 2

local result_cache = setmetatable({}, { __mode = "k" })
local range_cache  = setmetatable({}, { __mode = "k" })

local _M = {}


local function get_cache_value(cache, cache_key)
  local cache_value = cache[cache_key]
  if not cache_value then
    cache_value = {}
    cache[cache_key] = cache_value
  end
  return cache_value
end

local function extract_range(status_code_list, range)
  local start_code, end_code
  local ranges = get_cache_value(range_cache, status_code_list)

  -- If range isn't in the cache, extract and put it in
  if not ranges[range] then
    local range_result, err = match(range, constants.REGEX_SPLIT_STATUS_CODES_BY_DASH, "oj")

    if err then
      kong.log.err(err)
      return
    end
    ranges[range] = { range_result[START_RANGE_IDX], range_result[END_RANGE_IDX] }
  end

  start_code = ranges[range][START_RANGE_IDX]
  end_code = ranges[range][END_RANGE_IDX]

  return start_code, end_code
end

-- Returns true if a given status code is within status code ranges
local function is_in_range(status_code_list, status_code)
  -- If there is no configuration then pass all response codes
  if not status_code_list then
    return true
  end

  local result_list = get_cache_value(result_cache, status_code_list)
  local result = result_list[status_code]

  -- If result is found in a cache then return results instantly
  if result ~= nil then
    return result
  end

  for _, range in ipairs(status_code_list) do
    -- Get status code range splitting by "-" character
    local start_code, end_code = extract_range(status_code_list, range)

    -- Checks if there is both interval numbers
    if start_code and end_code then
      -- If HTTP response code is in the range return true
      if status_code >= tonumber(start_code) and status_code <= tonumber(end_code) then
        -- Storing results in a cache
        result_list[status_code] = true
        return true
      end
    end
  end

  -- Return false if there are no match for a given status code ranges and store it in cache
  result_list[status_code] = false
  return false
end


local worker_id
local hostname = re_gsub(knode.get_hostname(), [[\.]], "_", "oj")

-- downsample timestamp
local shdict_metrics_last_sent = 0
local SHDICT_METRICS_SEND_THRESHOLD = 60


local get_consumer_id = {
  consumer_id = function(consumer)
    return consumer and consumer.id
  end,
  custom_id   = function(consumer)
    return consumer and consumer.custom_id
  end,
  username    = function(consumer)
    return consumer and consumer.username
  end
}

local get_service_id = {
  service_id           = function(service)
    return service and service.id
  end,
  service_name         = function(service)
    return service and service.name
  end,
  service_host         = function(service)
    return service and service.host
  end,
  service_name_or_host = function(service)
    return service and (service.name ~= null and
      service.name or service.host)
  end
}

local get_workspace_id = {
  workspace_id   = function()
    return ws.get_workspace_id()
  end,
  workspace_name = function()
    return ws.get_workspace_name()
  end
}

local metrics = {
  unique_users = function (scope_name, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip unique_users in tag mode
      return
    end
    local get_consumer_id = get_consumer_id[metric_config.consumer_identifier or conf.consumer_identifier_default]
    local consumer_id     = get_consumer_id(message.consumer)

    if consumer_id then
      local stat = string_format("%s.user.uniques", scope_name)
      logger:send_statsd(stat, consumer_id, logger.stat_types.set)
    end
  end,
  request_per_user = function (scope_name, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip request_per_user in tag mode
      return
    end
    local get_consumer_id = get_consumer_id[metric_config.consumer_identifier or conf.consumer_identifier_default]
    local consumer_id     = get_consumer_id(message.consumer)

    if consumer_id then
      local stat = string_format("%s.user.%s.request.count", scope_name, consumer_id)
      logger:send_statsd(stat, 1, logger.stat_types.counter,
        metric_config.sample_rate)
    end
  end,
  status_count = function (scope_name, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip status_count in tag mode
      return
    end

    logger:send_statsd(string_format("%s.status.%s", scope_name, message.response.status),
      1, logger.stat_types.counter, metric_config.sample_rate)
  end,
  status_count_per_user = function (scope_name, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip status_count_per_user in tag mode
      return
    end
    local get_consumer_id = get_consumer_id[metric_config.consumer_identifier or conf.consumer_identifier_default]
    local consumer_id     = get_consumer_id(message.consumer)

    if consumer_id then
      logger:send_statsd(string_format("%s.user.%s.status.%s", scope_name,
        consumer_id, message.response.status),
        1, logger.stat_types.counter,
        metric_config.sample_rate)
    end
  end,
  status_count_per_workspace = function (scope_name, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip status_count_per_workspace in tag mode
      return
    end
    local get_workspace_id = get_workspace_id[metric_config.workspace_identifier or conf.workspace_identifier_default]
    local workspace_id     = get_workspace_id()

    if workspace_id then
      logger:send_statsd(string_format("%s.workspace.%s.status.%s", scope_name,
        workspace_id, message.response.status),
        1, logger.stat_types.counter,
        metric_config.sample_rate)
    end
  end,
  status_count_per_user_per_route = function (_, message, metric_config, logger, conf, tags)
    if conf.tag_style then
      -- skip status_count_per_user_per_route in tag mode
      return
    end

    local get_consumer_id = get_consumer_id[metric_config.consumer_identifier or conf.consumer_identifier_default]
    local consumer_id     = get_consumer_id(message.consumer)
    if not consumer_id then
      return
    end

    local route = message.route

    if route.id then
      logger:send_statsd(string_format("route.%s.user.%s.status.%s", route.id,
        consumer_id, message.response.status),
        1, logger.stat_types.counter,
        metric_config.sample_rate)
    end
  end,
}

-- add shdict metrics
if ngx.config.ngx_lua_version >= 10011 then
  metrics.shdict_usage = function (_, message, metric_config, logger, conf, tags)
    -- we don't need this for every request, send every 1 minute
    -- also only one worker needs to send this because it's shared
    if worker_id ~= 0 then
      return
    end

    local now = ngx_time()
    if shdict_metrics_last_sent + SHDICT_METRICS_SEND_THRESHOLD < now then
      shdict_metrics_last_sent = now
      for shdict_name, shdict in pairs(ngx.shared) do
        if conf.tag_style then
          local tags = {
            ["node"] = hostname,
            ["shdict"] = shdict_name,
          }
          logger:send_statsd(string_format("shdict.free_space"),
            shdict:free_space(), logger.stat_types.gauge,
            metric_config.sample_rate, tags, conf.tag_style)
          logger:send_statsd(string_format("shdict.capacity"),
            shdict:capacity(), logger.stat_types.gauge,
            metric_config.sample_rate, tags, conf.tag_style)
        else
          if conf.hostname_in_prefix then
            logger:send_statsd(string_format("shdict.%s.free_space", shdict_name),
              shdict:free_space(), logger.stat_types.gauge,
              metric_config.sample_rate)
            logger:send_statsd(string_format("shdict.%s.capacity", shdict_name),
              shdict:capacity(), logger.stat_types.gauge,
              metric_config.sample_rate)
          else
            logger:send_statsd(string_format("node.%s.shdict.%s.free_space",
              hostname, shdict_name),
              shdict:free_space(), logger.stat_types.gauge,
              metric_config.sample_rate)
            logger:send_statsd(string_format("node.%s.shdict.%s.capacity",
              hostname, shdict_name),
              shdict:capacity(), logger.stat_types.gauge,
              metric_config.sample_rate)
          end
        end

      end
    end
  end
end

local function get_scope_name(conf, message, service_identifier)
  local api = message.api
  local service = message.service
  local scope_name

  if service then
    scope_name = "service."
    -- don't fail on ce schema where service_identifier is not defined
    if not service_identifier then
      service_identifier = "service_name_or_host"
    end

    local service_name = get_service_id[service_identifier](service)
    if not service_name or service_name == null  then
      scope_name = scope_name .. "unnamed"
    else
      scope_name = scope_name .. re_gsub(service_name, [[\.]], "_", "oj")
    end
  elseif api then
    scope_name = "api."

    if not api or api == null then
      scope_name = scope_name .. "unnamed"
    else
      scope_name = scope_name .. re_gsub(api.name, [[\.]], "_", "oj")
    end
  else
    -- TODO: this follows the pattern used by
    -- https://github.com/Kong/kong/pull/2702 (which prevents an error from
    -- being thrown and avoids confusing reports as per our metrics keys), but
    -- as it stands, hides traffic from monitoring tools when the plugin is
    -- configured globally. In fact, this basically disables this plugin when
    -- it is configured to run globally, or per-consumer without an
    -- API/Route/Service.

    -- Changes in statsd-advanced: we still log these requests, but into a namespace of
    -- "global.unmatched".
    -- And we don't send upstream_latency and metrics with consumer or route
    scope_name = "global.unmatched"
  end

  return scope_name
end


local function get_tags(conf, message, metric_config)
  local tags = {}

  local get_workspace_id = get_workspace_id[metric_config.workspace_identifier or conf.workspace_identifier_default]
  local workspace_id     = get_workspace_id()


  local get_service_id = get_service_id[metric_config.service_identifier or conf.service_identifier_default]
  local service_id = get_service_id(message.service)

  local get_consumer_id = get_consumer_id[metric_config.consumer_identifier or conf.consumer_identifier_default]
  local consumer_id = get_consumer_id(message.consumer)

  if service_id then
    tags["service"] = service_id
  else
    -- do not record any stats if the service is not present
    return
  end

  local route_name
  if message and message.route then
    route_name = message.route.name or message.route.id
    tags["route"] = route_name
  end

  if workspace_id then
    tags["workspace"] = workspace_id
  end

  if workspace_id then
    tags["consumer"] = consumer_id
  end

  if hostname then
    tags["node"] = hostname
  end

  if message and message.response and message.response.status then
    tags["status"] = message.response.status
  end

  return tags
end


local function send_entries_to_upstream_server(conf, messages)
  local logger, err = statsd_logger:new(conf)
  if err then
    kong.log.err("failed to create Statsd logger: ", err)
    return false, err
  end

  for _, message in ipairs(messages) do
    local stat_name  = {
      request_size     = "request.size",
      response_size    = "response.size",
      latency          = "latency",
      upstream_latency = "upstream_latency",
      kong_latency     = "kong_latency",
      request_count    = "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,
    }

    for _, metric_config in pairs(conf.metrics) do
      local metric_config_name = metric_config.name
      local metric = metrics[metric_config_name]

      local name
      local tags
      if conf.tag_style then
        tags = get_tags(conf, message, metric_config)

      else
        name = get_scope_name(conf, message, metric_config.service_identifier or conf.service_identifier_default)
      end

      if metric then
        metric(name, message, metric_config, logger, conf, tags)

      else
        local stat_name = stat_name[metric_config_name]
        local stat_value = stat_value[metric_config_name]

        if stat_value ~= nil and stat_value ~= -1 then
          if conf.tag_style then
            logger:send_statsd(stat_name, stat_value,
              logger.stat_types[metric_config.stat_type],
              metric_config.sample_rate, tags, conf.tag_style)

          else
            logger:send_statsd(name .. "." .. stat_name, stat_value,
              logger.stat_types[metric_config.stat_type],
              metric_config.sample_rate)
          end
        end
      end
    end
  end

  logger:close_socket()
  return true
end



function _M.execute(conf)
  if not is_in_range(conf.allow_status_codes, ngx.status) then
    return
  end

  kong.log.debug("Status code is within given status code ranges")

  if not worker_id then
    worker_id = ngx.worker.id() or -1
  end

  conf._prefix = conf.prefix

  if conf.hostname_in_prefix and conf.tag_style == nil then
    conf._prefix = conf._prefix .. ".node." .. hostname
  end

  local message = kong.log.serialize({ngx = ngx, kong = kong, })
  message.cache_metrics = ngx.ctx.cache_metrics

  local ok, err = Queue.enqueue(
    Queue.get_plugin_params("statsd", conf),
    send_entries_to_upstream_server,
    conf,
    message
  )
  if not ok then
    kong.log.err("Failed to enqueue log entry to StatsD server: ", err)
  end
end

-- only for test
_M.is_in_range = is_in_range

return _M