<commit_before>--[[
Copyright 2012 Rackspace

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS-IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
--]]

local Emitter = require('core').Emitter
local math = require('math')
local timer = require('timer')
local fmt = require('string').format

local async = require('async')

local Scheduler = require('../schedule').Scheduler
local AgentClient = require('./client').AgentClient
local ConnectionMessages = require('./connection_messages').ConnectionMessages
local logging = require('logging')
local consts = require('../util/constants')
local misc = require('../util/misc')
local vtime = require('virgo-time')

local ConnectionStream = Emitter:extend()
function ConnectionStream:initialize(id, token, options)
  self._id = id
  self._token = token
  self._clients = {}
  self._unauthedClients = {}
  self._delays = {}
  self._messages = ConnectionMessages:new(self)
  self._activeTimeSyncClient = nil
  self._options = options or {}
end

--[[
Create and establish a connection to the multiple endpoints.

addresses - An Array of ip:port pairs
callback - Callback called with (err) when all the connections have been
established.
--]]
function ConnectionStream:createConnections(addresses, callback)
  async.series({
    function(callback)
      self._scheduler = Scheduler:new('scheduler.state', {}, callback)
      self._scheduler:on('check', function(check, checkResult)
        self:_sendMetrics(check, checkResult)
      end)
    end,
    function(callback)
      self._scheduler:start()
      callback()
    end,
    -- connect
    function(callback)
      async.forEach(addresses, function(address, callback)
        local split, client, options
        split = misc.splitAddress(address)
        options = misc.merge({
          host = split[1],
          port = split[2],
          datacenter = address
        }, self._options)
        self:createConnection(options, callback)
      end)
    end
  }, callback)
end

function ConnectionStream:_sendMetrics(check, checkResults)
  local client = self:getClient()
  if client then
    client.protocol:request('check_metrics.post', check, checkResults)
  end
end

function ConnectionStream:_setDelay(datacenter)
  local maxDelay = consts.DATACENTER_MAX_DELAY
  local jitter = consts.DATACENTER_MAX_DELAY_JITTER
  local previousDelay = self._delays[datacenter]
  local delay

  if previousDelay == nil then
    self._delays[datacenter] = 0
    previousDelay = 0
  end

  delay = math.min(previousDelay, maxDelay) + (jitter * math.random())
  self._delays[datacenter] = delay

  return delay
end

--[[
Retry a connection to the endpoint.

options - datacenter, host, port
  datacenter - Datacenter name / host alias.
  host - Hostname.
  port - Port.
callback - Callback called with (err)
]]--
function ConnectionStream:reconnect(options, callback)
  local datacenter = options.datacenter
  local delay = self:_setDelay(datacenter)

  logging.infof('%s:%d -> Retrying connection in %dms', options.host, options.port, delay)
  timer.setTimeout(delay, function()
    self:createConnection(options, callback)
  end)
end

function ConnectionStream:getClient()
  local client
  local latency
  local min_latency = 2147483647
  for k, v in pairs(self._clients) do
    latency = self._clients[k]:getLatency()
    if latency == nil then
      client = self._clients[k]
    elseif min_latency > latency then
      client = self._clients[k]
      min_latency = latency
    end
  end
  return client
end

--[[
The algorithm for syncing time follows:

Note: Promoted clients have been handshake accepted to the endpoint.

1. On promotion, attach a time_sync event to the client
2. If a client disconnects and it is the time sync client then find
   a new client to perform time syncs
]]--
function ConnectionStream:_attachTimeSyncEvent(client)
  if not client then
    self._activeTimeSyncClient = nil
    return
  end
  if self._activeTimeSyncClient then
    -- client already attached
    return
  end
  self._activeTimeSyncClient = client
  client:on('time_sync', function(timeObj)
    logging.info('Syncing time')
    vtime.timesync(timeObj.agent_send_timestamp, timeObj.server_receive_timestamp,
                   timeObj.server_response_timestamp, timeObj.agent_recv_timestamp)
  end)
end

--[[
Move an unauthenticated client to the list of clients that have been authenticated.
client - the client.
]]--
function ConnectionStream:_promoteClient(client)
  local datacenter = client:getDatacenter()
  client:log(logging.INFO, fmt('Connection has been authenticated to %s', datacenter))
  self._clients[datacenter] = client
  self._unauthedClients[datacenter] = nil
  self:_attachTimeSyncEvent(client)
  self:emit('promote')
end

--[[
Create and establish a connection to the endpoint.

datacenter - Datacenter name / host alias.
host - Hostname.
port - Port.
callback - Callback called with (err)
]]--
function ConnectionStream:createConnection(options, callback)
  local opts = misc.merge({<commit_msg>monitoring: connection_stream: remove syncing time info<commit_after>--[[
Copyright 2012 Rackspace

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS-IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
--]]

local Emitter = require('core').Emitter
local math = require('math')
local timer = require('timer')
local fmt = require('string').format

local async = require('async')

local Scheduler = require('../schedule').Scheduler
local AgentClient = require('./client').AgentClient
local ConnectionMessages = require('./connection_messages').ConnectionMessages
local logging = require('logging')
local consts = require('../util/constants')
local misc = require('../util/misc')
local vtime = require('virgo-time')

local ConnectionStream = Emitter:extend()
function ConnectionStream:initialize(id, token, options)
  self._id = id
  self._token = token
  self._clients = {}
  self._unauthedClients = {}
  self._delays = {}
  self._messages = ConnectionMessages:new(self)
  self._activeTimeSyncClient = nil
  self._options = options or {}
end

--[[
Create and establish a connection to the multiple endpoints.

addresses - An Array of ip:port pairs
callback - Callback called with (err) when all the connections have been
established.
--]]
function ConnectionStream:createConnections(addresses, callback)
  async.series({
    function(callback)
      self._scheduler = Scheduler:new('scheduler.state', {}, callback)
      self._scheduler:on('check', function(check, checkResult)
        self:_sendMetrics(check, checkResult)
      end)
    end,
    function(callback)
      self._scheduler:start()
      callback()
    end,
    -- connect
    function(callback)
      async.forEach(addresses, function(address, callback)
        local split, client, options
        split = misc.splitAddress(address)
        options = misc.merge({
          host = split[1],
          port = split[2],
          datacenter = address
        }, self._options)
        self:createConnection(options, callback)
      end)
    end
  }, callback)
end

function ConnectionStream:_sendMetrics(check, checkResults)
  local client = self:getClient()
  if client then
    client.protocol:request('check_metrics.post', check, checkResults)
  end
end

function ConnectionStream:_setDelay(datacenter)
  local maxDelay = consts.DATACENTER_MAX_DELAY
  local jitter = consts.DATACENTER_MAX_DELAY_JITTER
  local previousDelay = self._delays[datacenter]
  local delay

  if previousDelay == nil then
    self._delays[datacenter] = 0
    previousDelay = 0
  end

  delay = math.min(previousDelay, maxDelay) + (jitter * math.random())
  self._delays[datacenter] = delay

  return delay
end

--[[
Retry a connection to the endpoint.

options - datacenter, host, port
  datacenter - Datacenter name / host alias.
  host - Hostname.
  port - Port.
callback - Callback called with (err)
]]--
function ConnectionStream:reconnect(options, callback)
  local datacenter = options.datacenter
  local delay = self:_setDelay(datacenter)

  logging.infof('%s:%d -> Retrying connection in %dms', options.host, options.port, delay)
  timer.setTimeout(delay, function()
    self:createConnection(options, callback)
  end)
end

function ConnectionStream:getClient()
  local client
  local latency
  local min_latency = 2147483647
  for k, v in pairs(self._clients) do
    latency = self._clients[k]:getLatency()
    if latency == nil then
      client = self._clients[k]
    elseif min_latency > latency then
      client = self._clients[k]
      min_latency = latency
    end
  end
  return client
end

--[[
The algorithm for syncing time follows:

Note: Promoted clients have been handshake accepted to the endpoint.

1. On promotion, attach a time_sync event to the client
2. If a client disconnects and it is the time sync client then find
   a new client to perform time syncs
]]--
function ConnectionStream:_attachTimeSyncEvent(client)
  if not client then
    self._activeTimeSyncClient = nil
    return
  end
  if self._activeTimeSyncClient then
    -- client already attached
    return
  end
  self._activeTimeSyncClient = client
  client:on('time_sync', function(timeObj)
    vtime.timesync(timeObj.agent_send_timestamp, timeObj.server_receive_timestamp,
                   timeObj.server_response_timestamp, timeObj.agent_recv_timestamp)
  end)
end

--[[
Move an unauthenticated client to the list of clients that have been authenticated.
client - the client.
]]--
function ConnectionStream:_promoteClient(client)
  local datacenter = client:getDatacenter()
  client:log(logging.INFO, fmt('Connection has been authenticated to %s', datacenter))
  self._clients[datacenter] = client
  self._unauthedClients[datacenter] = nil
  self:_attachTimeSyncEvent(client)
  self:emit('promote')
end

--[[
Create and establish a connection to the endpoint.

datacenter - Datacenter name / host alias.
host - Hostname.
port - Port.
callback - Callback called with (err)
]]--
function ConnectionStream:createConnection(options, callback)
  local opts = misc.merge({
{'fraction_non_alphanumeric': 0.09357426016410067, 'fraction_numerical': 0.0044251866875633815, 'mean_word_length': 4.024548402037981, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4550929', 'n_tokens_mistral': 3136, 'n_tokens_neox': 3061, 'n_words': 964}