Class: Aikido::Zen::WorkerProcess::Agent::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/aikido/zen/worker_process/agent/client.rb

Constant Summary collapse

KEEPALIVE_INTERVAL =

The keepalive interval must be less than the RPC server read timeout.

4

Instance Method Summary collapse

Constructor Details

#initialize(host, port, secret: Aikido::Zen.secret, config: Aikido::Zen.config, worker: Aikido::Zen::Worker.new(config: config), keepalive_worker: Aikido::Zen::Worker.new(config: config), polling_interval: config.worker_process_polling_interval, heartbeat_interval: config.worker_process_heartbeat_interval, collector: Aikido::Zen.collector) ⇒ Client

Returns a new instance of Client.



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 11

def initialize(
  host,
  port,
  secret: Aikido::Zen.secret,
  config: Aikido::Zen.config,
  worker: Aikido::Zen::Worker.new(config: config),
  keepalive_worker: Aikido::Zen::Worker.new(config: config),
  polling_interval: config.worker_process_polling_interval,
  heartbeat_interval: config.worker_process_heartbeat_interval,
  collector: Aikido::Zen.collector
)
  @config = config
  @worker = worker
  @keepalive_worker = keepalive_worker
  @polling_interval = polling_interval
  @heartbeat_interval = heartbeat_interval
  @collector = collector

  @rpc_client = Aikido::Zen::RPC::Client.new(secret, host, port, reconnect: true)
end

Instance Method Details

#calculate_rate_limits(request) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 61

def calculate_rate_limits(request)
  result = @rpc_client.invoke(
    "calculate_rate_limits", Aikido::Zen::IPC::READ_TIMEOUT,
    request.route.as_json, request.client_ip, request.actor&.as_json
  )

  Aikido::Zen::RateLimiter::Result.from_json(result) if result
rescue => err
  @config.logger.error("Forked worker process #{Process.pid}: failed to get rate limits from parent: #{err.message}")
  raise
end

#close ⇒ Object 



32
33
34
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 32

def close
  @rpc_client.close
end

#send_collector_events ⇒ Object 



54
55
56
57
58
59
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 54

def send_collector_events
  events_data = @collector.flush_events.map(&:as_json)
  @rpc_client.invoke("send_collector_events", KEEPALIVE_INTERVAL, events_data)
rescue => err
  @config.logger.error("Forked worker process #{Process.pid}: failed to send collector events to parent: #{err.message}")
end

#start ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 36

def start
  @rpc_client.start do
    @worker.perform do
      update_settings(updated_settings)
    rescue => err
      @config.logger.error("Forked worker process #{Process.pid}: failed to get settings after connecting to parent: #{err.message}")
    end
  end

  schedule_tasks
end

#stop ⇒ Object 



48
49
50
51
52
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 48

def stop
  @keepalive_worker.shutdown
  @worker.shutdown
  @rpc_client.stop
end