Class: PromMultiProc::Writer

Inherits:
Object
  • Object
show all
Defined in:
lib/prom_multi_proc/writer.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket:, batch_size: nil, batch_timeout: nil, validate: false) ⇒ Writer

Returns a new instance of Writer.



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/prom_multi_proc/writer.rb', line 7

def initialize(socket:, batch_size: nil, batch_timeout: nil, validate: false)
  batch_size = 1 if batch_size.nil?
  batch_timeout = 3 if batch_timeout.nil?
  unless batch_size.is_a?(Integer) && batch_size > 0
    raise PromMultiProcError.new("Invalid batch size: #{batch_size}")
  end
  unless batch_timeout.is_a?(Integer) && batch_timeout > 0
    raise PromMultiProcError.new("Invalid batch timeout: #{batch_timeout}")
  end

  @batch_size = batch_size
  @batch_timeout = batch_timeout
  @validate = validate
  @socket = socket

  @messages = []
  @lock = Mutex.new
  @thread = Thread.new do
    loop do
      flush(force: true)
      sleep(@batch_timeout)
    end
  end
end

Instance Attribute Details

#batch_size ⇒ Object (readonly)

Returns the value of attribute batch_size.



5
6
7
 
# File 'lib/prom_multi_proc/writer.rb', line 5

def batch_size
  @batch_size
end

#batch_timeout ⇒ Object (readonly)

Returns the value of attribute batch_timeout.



5
6
7
 
# File 'lib/prom_multi_proc/writer.rb', line 5

def batch_timeout
  @batch_timeout
end

#socket ⇒ Object (readonly)

Returns the value of attribute socket.



5
6
7
 
# File 'lib/prom_multi_proc/writer.rb', line 5

def socket
  @socket
end

Instance Method Details

#flush(force: false) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/prom_multi_proc/writer.rb', line 65

def flush(force: false)
  @lock.synchronize do
    if (force && @messages.length > 0) || (@messages.length >= batch_size)
      begin
        write_socket(JSON.generate(@messages))
      rescue StandardError => e
        # Never raise into host app; drop the batch
        warn("prom_multi_proc_rb - flush: Failed to write batch to socket #{e}")
        false
      ensure
        @messages.clear
      end
    else
      true
    end
  end
end

#shutdown ⇒ Object 



36
37
38
39
40
41
 
# File 'lib/prom_multi_proc/writer.rb', line 36

def shutdown
  if @thread&.alive?
    flush
    @thread.kill
  end
end

#socket? ⇒ Boolean

Returns:

  • (Boolean) 


83
84
85
86
87
 
# File 'lib/prom_multi_proc/writer.rb', line 83

def socket?
  File.socket?(@socket) && File.writable?(@socket)
rescue StandardError
  false
end

#validate? ⇒ Boolean

Returns:

  • (Boolean) 


32
33
34
 
# File 'lib/prom_multi_proc/writer.rb', line 32

def validate?
  @validate
end

#write(metric, method, value, labels) ⇒ Object 



43
44
45
 
# File 'lib/prom_multi_proc/writer.rb', line 43

def write(metric, method, value, labels)
  write_multi([[metric, method, value, labels]])
end

#write_multi(metrics) ⇒ Object

array of arrays where inner array is length 4 matching arguments for signature of #write



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/prom_multi_proc/writer.rb', line 49

def write_multi(metrics)
  if validate?
    metrics.each do |m, method, value, labels|
      m.validate!(method, value, labels)
    end
  end

  @lock.synchronize do
    metrics.each do |m, method, value, labels|
      @messages << m.to_msg(method, value, labels)
    end
  end

  flush
end