Class: WideEvent::Store::Sender

Inherits:
Object
  • Object
show all
Defined in:
lib/wide_event/store/sender.rb

Overview

Owns the per-process bounded queue, batching, retry loop, and shutdown flush for delivering canonical event JSON to the store. Application threads only ever call #enqueue, which appends to an in-memory array under a mutex and never performs network I/O; a single worker thread seals batches, gzips them, and calls Client#ingest.

PID-aware: every enqueue compares Process.pid (or the injected process_id) against the PID recorded at construction/last reset. A mismatch (post-fork) discards the inherited queue and worker thread and starts fresh in the child before admitting the new event.

Defined Under Namespace

Classes: WireBatch

Constant Summary collapse

MAX_QUEUE_EVENTS = 
1000
MAX_QUEUE_BYTES = 
8 * 1024 * 1024
MAX_BATCH_EVENTS = 
100
MAX_BATCH_BYTES = 
1 * 1024 * 1024
MAX_COMPRESSED_BYTES = 
512 * 1024
FLUSH_INTERVAL = 
1.0
BASE_BACKOFF = 
1.0
MAX_BACKOFF = 
30.0
MAX_RETRY_AGE = 
24 * 60 * 60
WARNING_INTERVAL = 
60.0
SHUTDOWN_TIMEOUT = 
2.0

Instance Method Summary collapse

Constructor Details

#initialize(client:, service:, environment:, max_queue_events: MAX_QUEUE_EVENTS, max_queue_bytes: MAX_QUEUE_BYTES, max_batch_events: MAX_BATCH_EVENTS, max_batch_bytes: MAX_BATCH_BYTES, max_compressed_bytes: MAX_COMPRESSED_BYTES, flush_interval: FLUSH_INTERVAL, base_backoff: BASE_BACKOFF, max_backoff: MAX_BACKOFF, max_retry_age: MAX_RETRY_AGE, warning_interval: WARNING_INTERVAL, clock: -> { Time.now }, uuid: -> { SecureRandom.uuid }, process_id: -> { Process.pid }, pid: nil, autostart: true) ⇒ Sender

Returns a new instance of Sender.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/wide_event/store/sender.rb', line 33

def initialize(client:, service:, environment:,
                max_queue_events: MAX_QUEUE_EVENTS, max_queue_bytes: MAX_QUEUE_BYTES,
                max_batch_events: MAX_BATCH_EVENTS, max_batch_bytes: MAX_BATCH_BYTES,
                max_compressed_bytes: MAX_COMPRESSED_BYTES,
                flush_interval: FLUSH_INTERVAL, base_backoff: BASE_BACKOFF,
                max_backoff: MAX_BACKOFF, max_retry_age: MAX_RETRY_AGE,
                warning_interval: WARNING_INTERVAL,
                clock: -> { Time.now }, uuid: -> { SecureRandom.uuid },
                process_id: -> { Process.pid }, pid: nil, autostart: true)
  @client = client
  @service = service
  @environment = environment
  @max_queue_events = max_queue_events
  @max_queue_bytes = max_queue_bytes
  @max_batch_events = max_batch_events
  @max_batch_bytes = max_batch_bytes
  @max_compressed_bytes = max_compressed_bytes
  @flush_interval = flush_interval
  @base_backoff = base_backoff
  @max_backoff = max_backoff
  @max_retry_age = max_retry_age
  @warning_interval = warning_interval
  @clock = clock
  @uuid = uuid
  @process_id = process_id
  @autostart = autostart
  @max_event_bytes_per_batch = [ max_batch_bytes - envelope_overhead_bytes, 1 ].max

  @mutex = Mutex.new
  @cv = ConditionVariable.new
  reset_state_locked
  @pid = pid || @process_id.call
  @thread = nil
  @shutdown = false
  @shutdown_deadline = nil

  start_worker if @autostart
end

Instance Method Details

#__queue_for_test ⇒ Object

Test-only introspection of the raw pending queue.



118
119
120
 
# File 'lib/wide_event/store/sender.rb', line 118

def __queue_for_test
  @mutex.synchronize { @queue.dup }
end

#enqueue(event_json) ⇒ Object

Appends one canonical event JSON string to the bounded queue. Never performs network I/O and never raises into the caller.



74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/wide_event/store/sender.rb', line 74

def enqueue(event_json)
  @mutex.synchronize do
    reset_for_pid_change_locked
    admit_locked(event_json)
  end
  @cv.broadcast
  true
rescue StandardError => e
  WideEvent.handle_error(e, "store_sender_enqueue")
  false
end

#flush_once ⇒ Object

Performs exactly one send attempt of the current (or newly sealed) batch, ignoring the retry backoff schedule. Test-only synchronous hook; the worker thread drives normal delivery.



93
94
95
96
 
# File 'lib/wide_event/store/sender.rb', line 93

def flush_once
  tick(force: true)
  nil
end

#pending_drops ⇒ Object 



86
87
88
 
# File 'lib/wide_event/store/sender.rb', line 86

def pending_drops
  @mutex.synchronize { @pending_drops }
end

#shutdown(timeout: SHUTDOWN_TIMEOUT) ⇒ Object

Best-effort flush: keeps sending until nothing remains or the timeout elapses, then stops the worker thread.



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/wide_event/store/sender.rb', line 100

def shutdown(timeout: SHUTDOWN_TIMEOUT)
  @mutex.synchronize do
    @shutdown = true
    @shutdown_deadline = monotonic_now + timeout
  end
  @cv.broadcast
  if @thread
    @thread.join(timeout + 0.5)
    @thread.kill if @thread.alive?
  end
  @thread = nil
  nil
rescue StandardError => e
  WideEvent.handle_error(e, "store_sender_shutdown")
  nil
end