Class: Fluent::HttpBufferedOutput
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::HttpBufferedOutput
- Defined in:
- lib/fluent/plugin/out_http_buffered.rb
Overview
Main Output plugin class
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #format(tag, time, record) ⇒ Object
-
#initialize ⇒ HttpBufferedOutput
constructor
A new instance of HttpBufferedOutput.
- #shutdown ⇒ Object
- #start ⇒ Object
- #write(chunk) ⇒ Object
Constructor Details
#initialize ⇒ HttpBufferedOutput
Returns a new instance of HttpBufferedOutput.
8 9 10 11 12 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 8 def initialize super require 'net/http' require 'uri' end |
Instance Method Details
#configure(conf) ⇒ Object
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 26 def configure(conf) super # Check if endpoint URL is valid unless @endpoint_url =~ /^#{URI.regexp}$/ fail Fluent::ConfigError, 'endpoint_url invalid' end begin @uri = URI.parse(@endpoint_url) rescue URI::InvalidURIError raise Fluent::ConfigError, 'endpoint_url invalid' end # Parse http statuses @statuses = @http_retry_statuses.split(',').map { |status| status.to_i } @statuses = [] if @statuses.nil? @http = Net::HTTP.new(@uri.host, @uri.port) @http.read_timeout = @http_read_timeout @http.open_timeout = @http_open_timeout end |
#format(tag, time, record) ⇒ Object
62 63 64 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 62 def format(tag, time, record) [tag, time, record].to_msgpack end |
#shutdown ⇒ Object
54 55 56 57 58 59 60 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 54 def shutdown super begin @http.finish rescue end end |
#start ⇒ Object
50 51 52 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 50 def start super end |
#write(chunk) ⇒ Object
66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
# File 'lib/fluent/plugin/out_http_buffered.rb', line 66 def write(chunk) data = [] chunk.msgpack_each do |(tag, time, record)| data << [tag, time, record] end request = create_request(data) begin response = @http.start do |http| request = create_request(data) http.request request end if @statuses.include? response.code.to_i # Raise an exception so that fluent retries fail "Server returned bad status: #{response.code}" end rescue IOError, EOFError, SystemCallError => e # server didn't respond $log.warn "Net::HTTP.#{request.method.capitalize} raises exception: #{e.class}, '#{e.}'" ensure begin @http.finish rescue end end end |