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

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 50

def configure(conf)
  super
  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 



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

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) 


46
47
48
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 46

def multi_workers_ready?
  true
end

#package_record(record, timestamp) ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 69

def package_record(record, timestamp)
  packaged = {
    '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 



128
129
130
131
132
133
134
135
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 128

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 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/fluent/plugin/out_newrelic.rb', line 93

def write(chunk)
  payload = {
    'common' => {
      'attributes' => {
        'plugin' => {
          'type' => 'fluentd',
          'version' => NewrelicFluentdOutput::VERSION,
        }
      }
    },
    'logs' => []
  }
  chunk.msgpack_each do |ts, record|
    next unless record.is_a? Hash
    next if record.empty?
    payload['logs'].push(package_record(record, ts))
  end
  io = StringIO.new
  gzip = Zlib::GzipWriter.new(io)

  # Fluentd can run with a version of Ruby (2.1.0) whose to_json method doesn't support non-ASCII characters.
  # So we use Yajl, which can handle all Unicode characters. Apparently this library is what Fluentd uses
  # internally, so it is installed by default with td-agent.
  # See https://github.com/fluent/fluentd/issues/215
  gzip << Yajl.dump([payload])
  gzip.close
  send_payload(io.string)
end