revert reports changes

lua-tree/share/lua/5.1/kong/reports.lua

@@ -6,12 +6,21 @@ local knode = (kong and kong.node) and kong.node or
               require "kong.pdk.node".new()
 
 
+local kong_dict = ngx.shared.kong
+local ngx = ngx
+local tcp_sock = ngx.socket.tcp
+local timer_at = ngx.timer.at
+local ngx_log = ngx.log
+local var = ngx.var
+local subsystem = ngx.config.subsystem
 local concat = table.concat
 local tostring = tostring
 local lower = string.lower
 local pairs = pairs
 local error = error
 local type = type
+local WARN = ngx.WARN
+local DEBUG = ngx.DEBUG
 local sub = string.sub
 
 
@@ -77,13 +86,21 @@ do
 end
 
 
+local function log(lvl, ...)
+  ngx_log(lvl, "[reports] ", ...)
+end
+
+
 local function serialize_report_value(v)
   if type(v) == "function" then
     v = v()
   end
 
   if type(v) == "table" then
     local json, err = cjson.encode(v)
+    if err then
+      log(WARN, "could not JSON encode given table entity: ", err)
+    end
 
     v = json
   end
@@ -125,6 +142,9 @@ local function send_report(signal_type, t, host, port)
     end
   end
 
+  local sock = tcp_sock()
+  sock:settimeouts(30000, 30000, 30000)
+
   -- errors are not logged to avoid false positives for users
   -- who run Kong in an air-gapped environments
 
@@ -135,6 +155,7 @@ local function send_report(signal_type, t, host, port)
 
   local hs_ok, err = sock:sslhandshake(_ssl_session, nil, _ssl_verify)
   if not hs_ok then
+    log(DEBUG, "failed to complete SSL handshake for reports: ", err)
     return
   end
 
@@ -147,6 +168,30 @@ end
 
 -- ping timer handler
 
+
+-- Hold a lock for the whole interval (exptime) to prevent multiple
+-- worker processes from sending the test request simultaneously.
+-- Other workers do not need to wait until this lock is released,
+-- and can ignore the event, knowing another worker is handling it.
+-- We subtract 1ms to the exp time to prevent a race condition
+-- with the next timer event.
+local function get_lock(key, exptime)
+  local ok, err = kong_dict:safe_add(key, true, exptime - 0.001)
+  if not ok and err ~= "exists" then
+    log(WARN, "could not get lock from 'kong' shm: ", err)
+  end
+
+  return ok
+end
+
+
+local function create_timer(...)
+  local ok, err = timer_at(...)
+  if not ok then
+    log(WARN, "could not create ping timer: ", err)
+  end
+end
+
 -- @param interval exposed for unit test only
 local function create_counter(interval)
   local err
@@ -158,17 +203,87 @@ end
 
 local function get_counter(key)
   local count, err = report_counter:get(key)
+  if err then
+    log(WARN, "could not get ", key, " from 'kong' shm: ", err)
+  end
   return count or 0
 end
 
 
 local function reset_counter(key, amount)
   local ok, err = report_counter:reset(key, amount)
+  if not ok then
+    log(WARN, "could not reset ", key, " in 'kong' shm: ", err)
+  end
 end
 
 
 local function incr_counter(key)
   local ok, err = report_counter:incr(key, 1)
+  if not ok then
+    log(WARN, "could not increment ", key, " in 'kong' shm: ", err)
+  end
+end
+
+
+-- returns a string indicating the "kind" of the current request/stream:
+-- "http", "https", "h2c", "h2", "grpc", "grpcs", "ws", "wss", "tcp", "tls", "udp"
+-- or nil + error message if the suffix could not be determined
+local get_current_suffix
+
+if subsystem == "http" then
+function get_current_suffix(ctx)
+  local scheme = var.scheme
+  local proxy_mode = var.kong_proxy_mode
+  if scheme == "http" or scheme == "https" then
+    if proxy_mode == "http" or proxy_mode == "unbuffered" then
+      local http_upgrade = var.http_upgrade
+      if http_upgrade and lower(http_upgrade) == "websocket" then
+        if scheme == "http" then
+          return "ws"
+        end
+
+        return "wss"
+      end
+
+      if ngx.req.http_version() == 2 then
+        if scheme == "http" then
+          return "h2c"
+        end
+
+        return "h2"
+      end
+
+        return scheme -- http/https
+      end
+
+    if proxy_mode == "grpc" then
+      if scheme == "http" then
+        return "grpc"
+      end
+
+      if scheme == "https" then
+        return "grpcs"
+      end
+    end
+  end
+
+  if ctx.KONG_UNEXPECTED then
+    return nil
+  end
+
+  log(WARN, "could not determine log suffix (scheme=", tostring(scheme),
+            ", proxy_mode=", tostring(proxy_mode), ")")
+end
+
+else -- subsystem == "stream"
+  function get_current_suffix(ctx)
+    if var.ssl_protocol then
+      return "tls"
+    end
+
+    return lower(var.protocol)
+  end
 end
 
 
@@ -179,6 +294,28 @@ local function send_ping(host, port)
     _ping_infos.cluster_id = kong.cluster.get_id()
   end
 
+  if subsystem == "stream" then
+    _ping_infos.streams     = get_counter(STREAM_COUNT_KEY)
+    _ping_infos.tcp_streams = get_counter(TCP_STREAM_COUNT_KEY)
+    _ping_infos.udp_streams = get_counter(UDP_STREAM_COUNT_KEY)
+    _ping_infos.tls_streams = get_counter(TLS_STREAM_COUNT_KEY)
+    _ping_infos.go_plugin_reqs = get_counter(GO_PLUGINS_REQUEST_COUNT_KEY)
+
+    _ping_infos.stream_route_cache_hit_pos = get_counter(STEAM_ROUTE_CACHE_HITS_KEY_POS)
+    _ping_infos.stream_route_cache_hit_neg = get_counter(STEAM_ROUTE_CACHE_HITS_KEY_NEG)
+
+    send_report("ping", _ping_infos, host, port)
+
+    reset_counter(STREAM_COUNT_KEY, _ping_infos.streams)
+    reset_counter(TCP_STREAM_COUNT_KEY, _ping_infos.tcp_streams)
+    reset_counter(UDP_STREAM_COUNT_KEY, _ping_infos.udp_streams)
+    reset_counter(TLS_STREAM_COUNT_KEY, _ping_infos.tls_streams)
+    reset_counter(GO_PLUGINS_REQUEST_COUNT_KEY, _ping_infos.go_plugin_reqs)
+    reset_counter(STEAM_ROUTE_CACHE_HITS_KEY_POS, _ping_infos.stream_route_cache_hit_pos)
+    reset_counter(STEAM_ROUTE_CACHE_HITS_KEY_NEG, _ping_infos.stream_route_cache_hit_neg)
+    return
+  end
+
   _ping_infos.requests       = get_counter(REQUEST_COUNT_KEY)
   _ping_infos.http_reqs      = get_counter(HTTP_REQUEST_COUNT_KEY)
   _ping_infos.https_reqs     = get_counter(HTTPS_REQUEST_COUNT_KEY)
@@ -277,6 +414,7 @@ do
     local res, err = red:info("server")
     if type(res) ~= "string" then
       -- could be nil or ngx.null
+      ngx_log(WARN, "failed to retrieve Redis version: ", err)
 
     else
       -- retrieve first 2 digits only
@@ -314,9 +452,24 @@ return {
       return
     end
 
+    local count_key = subsystem == "stream" and STREAM_COUNT_KEY
+                                             or REQUEST_COUNT_KEY
+    incr_counter(count_key)
+
     if ctx.ran_go_plugin then
       incr_counter(GO_PLUGINS_REQUEST_COUNT_KEY)
     end
+
+    local suffix = get_current_suffix(ctx)
+    if suffix then
+      incr_counter(count_key .. ":" .. suffix)
+    end
+
+    local route_match_cached = ctx.route_match_cached
+
+    if route_match_cached then
+      incr_counter(count_key .. ":" .. ROUTE_CACHE_HITS_KEY .. ":" .. route_match_cached)
+    end
   end,
 
   -- custom methods