/kong/db/dao/targets.lua

local balancer = require "kong.runloop.balancer"
local cjson = require "cjson"
local workspaces   = require "kong.workspaces"
local tools_ip = require "kong.tools.ip"


local setmetatable = setmetatable
local tostring = tostring
local ipairs = ipairs
local table = table
local type = type
local min = math.min
local table_merge = require("kong.tools.table").table_merge


local _TARGETS = {}
local DEFAULT_PORT = 8000


local function sort_targets(a, b)
  if a.created_at < b.created_at then
    return true
  end
  if a.created_at == b.created_at then
    return a.id < b.id
  end
  return false
end


local function format_target(target)
  local p = tools_ip.normalize_ip(target)
  if not p then
    return false, "Invalid target; not a valid hostname or ip address"
  end
  return tools_ip.format_host(p, DEFAULT_PORT)
end


function _TARGETS:insert(entity, options)
  if entity.target then
    local formatted_target, err = format_target(entity.target)
    if not formatted_target then
      local err_t = self.errors:schema_violation({ target = err })
      return nil, tostring(err_t), err_t
    end
    entity.target = formatted_target
  end

  return self.super.insert(self, entity, options)
end


function _TARGETS:upsert(pk, entity, options)
  entity.id = pk.id

  if entity.target then
    local formatted_target, err = format_target(entity.target)
    if not formatted_target then
      local err_t = self.errors:schema_violation({ target = err })
      return nil, tostring(err_t), err_t
    end
    entity.target = formatted_target
  end

  -- backward compatibility with Kong older than 2.2.0
  local workspace = workspaces.get_workspace_id()
  local opts = { nulls = true, workspace = workspace }
  for existent, err, err_t in self:each_for_upstream(entity.upstream, nil, opts) do
    if not existent then
      return nil, err, err_t
    end
    if existent.target == entity.target then
      -- if the upserting entity is newer, update
      if entity.created_at > existent.created_at then
        local ok, err, err_t = self.super.delete(self, existent, opts)
        if ok then
          return self.super.insert(self, entity, options)
        end

        return ok, err, err_t
      end
      -- if upserting entity is older, keep the existent entity
      return true
    end
  end

  return self.super.insert(self, entity, options)
end


function _TARGETS:upsert_by_target(unique_key, entity, options)
  entity.target = unique_key
  return self:insert(entity, options)
end


function _TARGETS:select(pk, options)
  local target, err, err_t = self.super.select(self, pk, options)
  if err then
    return nil, err, err_t
  end

  if target then
    local formatted_target, err = format_target(target.target)
    if not formatted_target then
      local err_t = self.errors:schema_violation({ target = err })
      return nil, tostring(err_t), err_t
    end
    target.target = formatted_target
  end
  return target
end


function _TARGETS:delete_by_target(tgt, options)
  local target, err, err_t = self:select_by_target(tgt)
  if err then
    return nil, err, err_t
  end

  return self.super.delete(self, target, options)
end


-- Paginate through the targets for an upstream, including those with
-- weight=0 (i.e. the "raw" representation of targets in the database)
function _TARGETS:page_for_upstream_raw(upstream_pk, ...)
  local page, err, err_t, offset =
    self.super.page_for_upstream(self, upstream_pk, ...)
  if err then
    return nil, tostring(err), err_t
  end

  for _, target in ipairs(page) do
    local formatted_target, err = format_target(target.target)
    if not formatted_target then
      local err_t = self.errors:schema_violation({ target = err })
      return nil, tostring(err_t), err_t
    end
    target.target = formatted_target
  end

  return page, nil, nil, offset
end


-- Return the entire list of targets for an upstream, including entries with
-- weight=0 (i.e. the "raw" representation of targets in the database)
function _TARGETS:select_by_upstream_raw(upstream_pk, options)
  local targets = {}

  -- Note that each_for_upstream is not overridden, so it returns "raw".
  for target, err, err_t in self:each_for_upstream(upstream_pk, nil, options) do
    if not target then
      return nil, err, err_t
    end
    local formatted_target, err = format_target(target.target)
    if not formatted_target then
      local err_t = self.errors:schema_violation({ target = err })
      return nil, tostring(err_t), err_t
    end
    target.target = formatted_target

    table.insert(targets, target)
  end

  table.sort(targets, sort_targets)

  return targets
end


-- Paginate through targets for an upstream, returning only the active
-- (weight>0) target.
function _TARGETS:page_for_upstream(upstream_pk, size, offset, options)
  -- Read all targets
  local targets, err, err_t = self:select_by_upstream_raw(upstream_pk, options)
  if not targets then
    return nil, err, err_t
  end

  local all_active_targets = {}
  local seen = {}
  local len = 0

  for i = #targets, 1, -1 do
    local entry = targets[i]
    if not seen[entry.target] then
      -- add what we want to send to the client in our array
      len = len + 1
      all_active_targets[len] = entry

      -- track that we found this host:port so we only show
      -- the most recent active one
      seen[entry.target] = true
    end
  end

  local pagination = self.pagination

  if type(options) == "table" and type(options.pagination) == "table" then
    pagination = table_merge(pagination, options.pagination)
  end

  if not size then
    size = pagination.page_size
  end

  size = min(size, pagination.max_page_size)
  offset = offset or 0

  -- Extract the requested page
  local page = setmetatable({}, cjson.array_mt)
  for i = 1 + offset, size + offset do
    local target = all_active_targets[i]
    if not target then
      break
    end
    table.insert(page, target)
  end

  local next_offset
  if all_active_targets[size + offset + 1] then
    next_offset = tostring(size + offset)
  end

  return page, nil, nil, next_offset
end


-- Paginate through targets for an upstream, returning only the
-- latest state of each active (weight>0) target, and include
-- health information to the returned records.
function _TARGETS:page_for_upstream_with_health(upstream_pk, ...)
  local targets, err, err_t, next_offset = self:page_for_upstream(upstream_pk, ...)
  if not targets then
    return nil, err, err_t
  end

  local health_info
  health_info, err = balancer.get_upstream_health(upstream_pk.id)
  if err then
    ngx.log(ngx.ERR, "failed getting upstream health: ", err)
  end

  if health_info then
    for _, target in ipairs(targets) do
      -- In case of DNS errors when registering a target,
      -- that error happens inside lua-resty-dns-client
      -- and the end-result is that it just doesn't launch the callback,
      -- which means kong.runloop.balancer and healthchecks don't get
      -- notified about the target at all. We extrapolate the DNS error
      -- out of the fact that the target is missing from the balancer.
      -- Note that lua-resty-dns-client does retry by itself,
      -- meaning that if DNS is down and it eventually resumes working, the
      -- library will issue the callback and the target will change state.
      if health_info[target.target] ~= nil and
        #health_info[target.target].addresses > 0 then
        target.health = "HEALTHCHECKS_OFF"
        -- If any of the target addresses are healthy, then the target is
        -- considered healthy.
        for _, address in ipairs(health_info[target.target].addresses) do
          if address.health == "HEALTHY" then
            target.health = "HEALTHY"
            break
          elseif address.health == "UNHEALTHY" then
            target.health = "UNHEALTHY"
          end

        end
      else
        target.health = "DNS_ERROR"
      end
      target.data = health_info[target.target]
    end

  end

  return targets, nil, nil, next_offset
end


function _TARGETS:select_by_upstream_filter(upstream_pk, filter, options)
  local targets, err, err_t = self:select_by_upstream_raw(upstream_pk, options)
  if not targets then
    return nil, err, err_t
  end

  for i = #targets, 1, -1 do
    local t = targets[i]
    if t.id == filter.id or t.target == filter.target then
      return t
    end
  end
end


function _TARGETS:post_health(upstream_pk, target, address, is_healthy)
  local upstream = balancer.get_upstream_by_id(upstream_pk.id)
  local host_addr = tools_ip.normalize_ip(target.target)
  local hostname = tools_ip.format_host(host_addr.host)
  local ip
  local port

  if address ~= nil then
    local addr = tools_ip.normalize_ip(address)
    ip = addr.host
    if addr.port then
      port = addr.port
    else
      port = DEFAULT_PORT
    end
  else
    ip = nil
    port = host_addr.port
  end

  local _, err = balancer.post_health(upstream, hostname, ip, port, is_healthy)
  if err then
    return nil, err
  end

  local health = is_healthy and 1 or 0
  local packet = ("%s|%s|%d|%d|%s|%s"):format(hostname, ip or "", port, health,
                                           upstream.id,
                                           upstream.name)

  kong.cluster_events:broadcast("balancer:post_health", packet)

  return true
end


function _TARGETS:get_balancer_health(upstream_pk)
  local health_info, err = balancer.get_balancer_health(upstream_pk.id)
  if err then
    ngx.log(ngx.ERR, "failed getting upstream health: ", err)
  end

  return health_info
end


return _TARGETS