Class: NewRelic::Agent::LogEventAggregator

Inherits:
EventAggregator show all
Defined in:
lib/new_relic/agent/log_event_aggregator.rb

Constant Summary collapse

LEVEL_KEY =

Per-message keys

'level'.freeze
MESSAGE_KEY = 
'message'.freeze
TIMESTAMP_KEY = 
'timestamp'.freeze
PRIORITY_KEY = 
'priority'.freeze
LINES =

Metric keys

'Logging/lines'.freeze
DROPPED_METRIC = 
'Logging/Forwarding/Dropped'.freeze
SEEN_METRIC = 
'Supportability/Logging/Forwarding/Seen'.freeze
SENT_METRIC = 
'Supportability/Logging/Forwarding/Sent'.freeze
LOGGER_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Ruby/Logger/%s'.freeze
LOGSTASHER_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Ruby/LogStasher/%s'.freeze
LOGGING_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Ruby/Logging/%s'.freeze
METRICS_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Metrics/Ruby/%s'.freeze
FORWARDING_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Forwarding/Ruby/%s'.freeze
DECORATING_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/LocalDecorating/Ruby/%s'.freeze
LABELS_SUPPORTABILITY_FORMAT = 
'Supportability/Logging/Labels/Ruby/%s'.freeze
MAX_BYTES =

32 * 1024 bytes (32 kibibytes)

32768
OVERALL_ENABLED_KEY =

Config keys

:'application_logging.enabled'
METRICS_ENABLED_KEY = 
:'application_logging.metrics.enabled'
FORWARDING_ENABLED_KEY = 
:'application_logging.forwarding.enabled'
DECORATING_ENABLED_KEY = 
:'application_logging.local_decorating.enabled'
LABELS_ENABLED_KEY = 
:'application_logging.forwarding.labels.enabled'
LOG_LEVEL_KEY = 
:'application_logging.forwarding.log_level'
CUSTOM_ATTRIBUTES_KEY = 
:'application_logging.forwarding.custom_attributes'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from EventAggregator

#after_initialize, buffer_class, capacity_key, #enabled?, enabled_fn, enabled_keys, #has_metadata?, #merge!, named

Constructor Details

#initialize(events) ⇒ LogEventAggregator

Returns a new instance of LogEventAggregator.



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 49

def initialize(events)
  super(events)
  @counter_lock = Mutex.new
  @seen = 0
  @seen_by_severity = Hash.new(0)
  @high_security = NewRelic::Agent.config[:high_security]
  @instrumentation_logger_enabled = NewRelic::Agent::Instrumentation::Logger.enabled?
  @attributes = NewRelic::Agent::LogEventAttributes.new

  register_for_done_configuring(events)
end

Instance Attribute Details

#attributes ⇒ Object (readonly)

Returns the value of attribute attributes.



47
48
49
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 47

def attributes
  @attributes
end

Class Method Details

.payload_to_melt_format(data) ⇒ Object

Because our transmission format (MELT) is different than historical agent payloads, extract the munging here to keep the service focused on the general harvest + transmit instead of the format.

Payload shape matches the publicly documented MELT format. docs.newrelic.com/docs/logs/log-api/introduction-log-api

We have to keep the aggregated payloads in a separate shape, though, to work with the priority sampling buffers



256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 256

def self.payload_to_melt_format(data)
  common_attributes = LinkingMetadata.append_service_linking_metadata({})

  # To save on unnecessary data transmission, trim the entity.type
  # sent by classic logs-in-context
  common_attributes.delete(ENTITY_TYPE_KEY)
  aggregator = NewRelic::Agent.agent.log_event_aggregator
  common_attributes.merge!(aggregator.attributes.custom_attributes)
  common_attributes.merge!(aggregator.labels)

  _, items = data
  payload = [{
    common: {attributes: common_attributes},
    logs: items.map(&:last)
  }]

  return [payload, items.size]
end

Instance Method Details

#add_custom_attributes(custom_attributes) ⇒ Object 



239
240
241
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 239

def add_custom_attributes(custom_attributes)
  attributes.add_custom_attributes(custom_attributes)
end

#add_event_metadata(formatted_message, severity) ⇒ Object 



168
169
170
171
172
173
174
175
176
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 168

def add_event_metadata(formatted_message, severity)
  metadata = {
    LEVEL_KEY => severity,
    TIMESTAMP_KEY => Process.clock_gettime(Process::CLOCK_REALTIME) * 1000
  }
  metadata[MESSAGE_KEY] = formatted_message unless formatted_message.nil?

  LinkingMetadata.append_trace_linking_metadata(metadata)
end

#add_logging_event_attributes(event, log, mdc_data) ⇒ Object 



220
221
222
223
224
225
226
227
228
229
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 220

def add_logging_event_attributes(event, log, mdc_data)
  event['level_number'] = log.level # Logging always assigns a level number
  event['file'] = log.file unless log.file.empty?
  event['line'] = log.line unless (log.line.respond_to?(:empty?) && log.line.empty?) || log.line.nil?
  event['method_name'] = log.method_name unless log.method_name.nil? || log.method_name.empty?
  event['logger'] = log.logger unless log.logger.empty?
  add_mdc_data_to_event(event, mdc_data) if !mdc_data.empty?

  event
end

#add_logstasher_event_attributes(event, log) ⇒ Object 



202
203
204
205
206
207
208
209
210
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 202

def add_logstasher_event_attributes(event, log)
  log_copy = log.dup
  # Delete previously reported attributes
  log_copy.delete('message')
  log_copy.delete('level')
  log_copy.delete('@timestamp')

  event['attributes'] = log_copy
end

#add_mdc_data_to_event(event, mdc_data) ⇒ Object 



231
232
233
234
235
236
237
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 231

def add_mdc_data_to_event(event, mdc_data)
  mdc_data.each do |key, value|
    event["context.mdc.#{key}"] = value
  end
rescue => e
  NewRelic::Agent.logger.debug("Failed to add Logging MDC data to event: #{e.message}")
end

#capacity ⇒ Object 



61
62
63
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 61

def capacity
  @buffer.capacity
end

#create_event(priority, formatted_message, severity) ⇒ Object 



187
188
189
190
191
192
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 187

def create_event(priority, formatted_message, severity)
  formatted_message = truncate_message(formatted_message)
  event = add_event_metadata(formatted_message, severity)

  create_prioritized_event(priority, event)
end

#create_logging_event(priority, severity, log, mdc_data) ⇒ Object 



212
213
214
215
216
217
218
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 212

def create_logging_event(priority, severity, log, mdc_data)
  formatted_message = truncate_message(log.data)
  event = add_event_metadata(formatted_message, severity)
  add_logging_event_attributes(event, log, mdc_data)

  create_prioritized_event(priority, event)
end

#create_logstasher_event(priority, severity, log) ⇒ Object 



194
195
196
197
198
199
200
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 194

def create_logstasher_event(priority, severity, log)
  formatted_message = log['message'] ? truncate_message(log['message']) : nil
  event = add_event_metadata(formatted_message, severity)
  add_logstasher_event_attributes(event, log)

  create_prioritized_event(priority, event)
end

#create_prioritized_event(priority, event) ⇒ Object 



178
179
180
181
182
183
184
185
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 178

def create_prioritized_event(priority, event)
  [
    {
      PrioritySampledBuffer::PRIORITY_KEY => priority
    },
    event
  ]
end

#determine_severity(log) ⇒ Object 



141
142
143
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 141

def determine_severity(log)
  log['level'] ? log['level'].to_s.upcase : 'UNKNOWN'
end

#harvest! ⇒ Object 



275
276
277
278
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 275

def harvest!
  record_customer_metrics()
  super
end

#increment_event_counters(severity) ⇒ Object 



145
146
147
148
149
150
151
152
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 145

def increment_event_counters(severity)
  return unless NewRelic::Agent.config[METRICS_ENABLED_KEY]

  @counter_lock.synchronize do
    @seen += 1
    @seen_by_severity[severity] += 1
  end
end

#labels ⇒ Object 



243
244
245
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 243

def labels
  @labels ||= create_labels
end

#logger_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


289
290
291
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 289

def logger_enabled?
  @enabled && @instrumentation_logger_enabled
end

#logging_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


297
298
299
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 297

def logging_enabled?
  @enabled && NewRelic::Agent::Instrumentation::Logging::Logger.enabled?
end

#logstasher_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


293
294
295
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 293

def logstasher_enabled?
  @enabled && NewRelic::Agent::Instrumentation::LogStasher.enabled?
end

#monitoring_conditions_met?(severity) ⇒ Boolean

Returns:

  • (Boolean) 


137
138
139
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 137

def monitoring_conditions_met?(severity)
  !severity_too_low?(severity) && NewRelic::Agent.config[FORWARDING_ENABLED_KEY] && !@high_security
end

#record(formatted_message, severity) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 65

def record(formatted_message, severity)
  return unless logger_enabled?

  severity = 'UNKNOWN' if severity.nil? || severity.empty?
  increment_event_counters(severity)

  return if formatted_message.nil? || formatted_message.empty?
  return unless monitoring_conditions_met?(severity)

  txn = NewRelic::Agent::Transaction.tl_current
  priority = LogPriority.priority_for(txn)

  return txn.add_log_event(create_event(priority, formatted_message, severity)) if txn

  @lock.synchronize do
    @buffer.append(priority: priority) do
      create_event(priority, formatted_message, severity)
    end
  end
rescue
  nil
end

#record_batch(txn, logs) ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 154

def record_batch(txn, logs)
  # Ensure we have the same shared priority
  priority = LogPriority.priority_for(txn)
  logs.each do |log|
    log.first[PRIORITY_KEY] = priority
  end

  @lock.synchronize do
    logs.each do |log|
      @buffer.append(event: log)
    end
  end
end

#record_logging_event(log, mdc_data) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 114

def record_logging_event(log, mdc_data)
  return unless logging_enabled?
  return if log.data.nil? || log.data.empty?

  severity = ::Logging::LNAMES[log.level] || 'UNKNOWN'
  increment_event_counters(severity)

  return unless monitoring_conditions_met?(severity)

  txn = NewRelic::Agent::Transaction.tl_current
  priority = LogPriority.priority_for(txn)

  return txn.add_log_event(create_logging_event(priority, severity, log, mdc_data)) if txn

  @lock.synchronize do
    @buffer.append(priority: priority) do
      create_logging_event(priority, severity, log, mdc_data)
    end
  end
rescue
  nil
end

#record_logstasher_event(log) ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 88

def record_logstasher_event(log)
  return unless logstasher_enabled?

  # LogStasher logs do not inherently include a message key, so most logs are recorded.
  # But when the key exists, we should not record the log if the message value is nil or empty.
  return if log.key?('message') && (log['message'].nil? || log['message'].empty?)

  severity = determine_severity(log)
  increment_event_counters(severity)

  return unless monitoring_conditions_met?(severity)

  txn = NewRelic::Agent::Transaction.tl_current
  priority = LogPriority.priority_for(txn)

  return txn.add_log_event(create_logstasher_event(priority, severity, log)) if txn

  @lock.synchronize do
    @buffer.append(priority: priority) do
      create_logstasher_event(priority, severity, log)
    end
  end
rescue
  nil
end

#reset! ⇒ Object 



280
281
282
283
284
285
286
287
 
# File 'lib/new_relic/agent/log_event_aggregator.rb', line 280

def reset!
  @counter_lock.synchronize do
    @seen = 0
    @seen_by_severity.clear
  end

  super
end