Class: Fluent::Plugin::NewrelicOutput

Inherits:
Output
  • Object
show all
Defined in:
lib/fluent/plugin/out_newrelic.rb

Defined Under Namespace

Classes: ConnectionFailure

Constant Summary collapse

DEFAULT_BUFFER_TYPE = 
'memory'.freeze
DEFAULT_TIMEKEY = 
5
DEFAULT_TIMEKEY_WAIT = 
0
MAX_PAYLOAD_SIZE =

bytes

1000000

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 55

def configure(conf)
  super

  @api_key ||= ENV["NEW_RELIC_API_KEY"]
  @license_key ||= ENV["NEW_RELIC_LICENSE_KEY"]
  if @api_key.nil? && @license_key.nil?
    raise Fluent::ConfigError.new("'api_key' or 'license_key' parameter is required")
  end

  # create initial sockets hash and socket based on config param
  @end_point = URI.parse(@base_uri)
  auth = {
    @api_key.nil? ? 'X-License-Key' : 'X-Insert-Key' =>
    @api_key.nil? ? @license_key : @api_key
  }
  @header = {
      'X-Event-Source' => 'logs',
      'Content-Encoding' => 'gzip'
  }.merge(auth)
  .freeze
end

#handle_response(response) ⇒ Object 



114
115
116
117
118
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 114

def handle_response(response)
  if !(200 <= response.code.to_i && response.code.to_i < 300)
    log.error("Response was " + response.code + " " + response.body)
  end
end

#multi_workers_ready? ⇒ Boolean

This tells Fluentd that it can run this output plugin in multiple workers. Our plugin has no interactions with other processes

Returns:

  • (Boolean) 


51
52
53
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 51

def multi_workers_ready?
  true
end

#package_record(record, timestamp) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 77

def package_record(record, timestamp)
  packaged = {
    'timestamp' => resolveTimestamp(record['timestamp'], timestamp),
    # non-intrinsic attributes get put into 'attributes'
    'attributes' => record
  }

  # intrinsic attributes go at the top level
  if record.has_key?('message')
    packaged['message'] = record['message']
    packaged['attributes'].delete('message')
  end

  # Kubernetes logging puts the message field in the 'log' attribute, we'll use that
  # as the 'message' field if it exists. We do the same in the Fluent Bit output plugin.
  # See https://docs.docker.com/config/containers/logging/fluentd/
  if record.has_key?('log')
    packaged['message'] = record['log']
    packaged['attributes'].delete('log')
  end

  packaged
end

#send_payload(payload) ⇒ Object 



120
121
122
123
124
125
126
127
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 120

def send_payload(payload)
  http = Net::HTTP.new(@end_point.host, 443)
  http.use_ssl = true
  http.verify_mode = OpenSSL::SSL::VERIFY_PEER
  request = Net::HTTP::Post.new(@end_point.request_uri, @header)
  request.body = payload
  handle_response(http.request(request))
end

#write(chunk) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 101

def write(chunk)
  logs = []
  chunk.msgpack_each do |ts, record|
    next unless record.is_a? Hash
    next if record.empty?
    logs.push(package_record(record, ts))
  end


  payloads = get_compressed_payloads(logs)
  payloads.each { |payload| send_payload(payload) }
end