Class: Kamal::OtelShipper

Inherits:
Object
  • Object
show all
Defined in:
lib/kamal/otel_shipper.rb

Constant Summary collapse

BATCH_SIZE = 
100
FLUSH_INTERVAL = 
5.seconds
OTEL_ATTRIBUTE_KEYS = 
{
  service: "service.namespace",
  version: "kamal.deploy_version",
  performer: "kamal.performer",
  destination: "deployment.environment.name"
}
SEVERITIES = 
{
  debug: { severityNumber: 5,  severityText: "DEBUG" },
  info:  { severityNumber: 9,  severityText: "INFO" },
  warn:  { severityNumber: 13, severityText: "WARN" },
  error: { severityNumber: 17, severityText: "ERROR" },
  fatal: { severityNumber: 21, severityText: "FATAL" }
}
LOGGER_SEVERITIES = 
{
  Logger::DEBUG => :debug,
  Logger::INFO  => :info,
  Logger::WARN  => :warn,
  Logger::ERROR => :error,
  Logger::FATAL => :fatal
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(endpoint:, tags:) ⇒ OtelShipper

Returns a new instance of OtelShipper.



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/kamal/otel_shipper.rb', line 36

def initialize(endpoint:, tags:)
  @endpoint = URI("#{endpoint.chomp('/')}/v1/logs")
  @run_id = SecureRandom.uuid
  @resource_attributes = [
    { key: "service.name", value: { stringValue: "kamal" } },
    { key: "service.version", value: { stringValue: Kamal::VERSION } },
    { key: "kamal.run_id", value: { stringValue: @run_id } },
    *tags.tags.map do |key, value|
      otel_key = OTEL_ATTRIBUTE_KEYS.fetch(key, "kamal.#{key}")
      { key: otel_key, value: { stringValue: value.to_s } }
    end
  ]
  @buffer = Queue.new
  @flush_mutex = Mutex.new
  @running = true
  @signal = Queue.new
  @thread = start_flush_thread
end

Instance Attribute Details

#run_id ⇒ Object (readonly)

Returns the value of attribute run_id.



34
35
36
 
# File 'lib/kamal/otel_shipper.rb', line 34

def run_id
  @run_id
end

Instance Method Details

#<<(str) ⇒ Object 



55
56
57
 
# File 'lib/kamal/otel_shipper.rb', line 55

def <<(str)
  append(str)
end

#append(str, host: nil, iostream: nil, severity: nil) ⇒ Object 



59
60
61
62
63
64
65
66
67
 
# File 'lib/kamal/otel_shipper.rb', line 59

def append(str, host: nil, iostream: nil, severity: nil)
  otel_severity = LOGGER_SEVERITIES.fetch(severity, :info)
  extra = build_context_attributes(host: host, iostream: iostream)
  str.to_s.each_line do |line|
    enqueue build_record(line.chomp, severity: otel_severity, attributes: extra)
  end

  self
end

#event(name, severity: :info, **attributes) ⇒ Object 



69
70
71
72
73
74
 
# File 'lib/kamal/otel_shipper.rb', line 69

def event(name, severity: :info, **attributes)
  attrs = attributes.map { |k, v| { key: k.to_s, value: typed_value(v) } }
  enqueue build_record(name, severity: severity, event_name: name, attributes: attrs)

  self
end

#flush ⇒ Object 



76
77
78
79
80
81
 
# File 'lib/kamal/otel_shipper.rb', line 76

def flush
  @flush_mutex.synchronize do
    lines = drain_buffer
    ship(lines) if lines.any?
  end
end

#shutdown ⇒ Object 



83
84
85
86
87
88
 
# File 'lib/kamal/otel_shipper.rb', line 83

def shutdown
  @running = false
  @signal << true
  @thread&.join(FLUSH_INTERVAL + 1.second)
  flush
end