Class: Fluent::FalconOutput
- Inherits:
-
Output
- Object
- Output
- Fluent::FalconOutput
- Defined in:
- lib/fluent/plugin/out_falcon.rb
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #create_request(tag, time, record) ⇒ Object
- #emit(tag, es, chain) ⇒ Object
- #format_url(tag, time, record) ⇒ Object
-
#handle_record(tag, time, record) ⇒ Object
end send_request.
-
#initialize ⇒ FalconOutput
constructor
A new instance of FalconOutput.
- #send_request(req, uri) ⇒ Object
- #set_body(req, tag, time, record) ⇒ Object
- #set_header(req, tag, time, record) ⇒ Object
- #shutdown ⇒ Object
- #start ⇒ Object
Constructor Details
#initialize ⇒ FalconOutput
Returns a new instance of FalconOutput.
4 5 6 7 8 9 |
# File 'lib/fluent/plugin/out_falcon.rb', line 4 def initialize super require 'net/http' require 'uri' require 'yajl' end |
Instance Method Details
#configure(conf) ⇒ Object
28 29 30 31 32 33 34 35 36 |
# File 'lib/fluent/plugin/out_falcon.rb', line 28 def configure(conf) super @auth = case @authentication when 'basic' then :basic else :none end end |
#create_request(tag, time, record) ⇒ Object
60 61 62 63 64 65 66 67 |
# File 'lib/fluent/plugin/out_falcon.rb', line 60 def create_request(tag, time, record) url = format_url(tag, time, record) uri = URI.parse(url) req = Net::HTTP.const_get('Post').new(uri.path) set_body(req, tag, time, record) set_header(req, tag, time, record) return req, uri end |
#emit(tag, es, chain) ⇒ Object
105 106 107 108 109 110 |
# File 'lib/fluent/plugin/out_falcon.rb', line 105 def emit(tag, es, chain) es.each do |time, record| handle_record(tag, time, record) end chain.next end |
#format_url(tag, time, record) ⇒ Object
46 47 48 |
# File 'lib/fluent/plugin/out_falcon.rb', line 46 def format_url(tag, time, record) @endpoint_url end |
#handle_record(tag, time, record) ⇒ Object
end send_request
100 101 102 103 |
# File 'lib/fluent/plugin/out_falcon.rb', line 100 def handle_record(tag, time, record) req, uri = create_request(tag, time, record) send_request(req, uri) end |
#send_request(req, uri) ⇒ Object
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 94 95 96 97 98 |
# File 'lib/fluent/plugin/out_falcon.rb', line 69 def send_request(req, uri) is_rate_limited = (@rate_limit_msec != 0 and not @last_request_time.nil?) if is_rate_limited and ((Time.now.to_f - @last_request_time) * 1000.0 < @rate_limit_msec) $log.info('Dropped request due to rate limiting') return end res = nil begin if @auth and @auth == :basic req.basic_auth(@username, @password) end @last_request_time = Time.now.to_f res = Net::HTTP.new(uri.host, uri.port).start {|http| http.request(req) } rescue => e # rescue all StandardErrors # server didn't respond $log.warn "Net::HTTP.#{req.method.capitalize} raises exception: #{e.class}, '#{e.}'" raise e if @raise_on_error else unless res and res.is_a?(Net::HTTPSuccess) res_summary = if res "#{res.code} #{res.} #{res.body}" else "res=nil" end $log.warn "failed to #{req.method} #{uri} (#{res_summary})" end #end unless end # end begin end |
#set_body(req, tag, time, record) ⇒ Object
50 51 52 53 54 |
# File 'lib/fluent/plugin/out_falcon.rb', line 50 def set_body(req, tag, time, record) req.body = Yajl.dump(eval(@records)) req['Content-Type'] = 'application/json' req end |
#set_header(req, tag, time, record) ⇒ Object
56 57 58 |
# File 'lib/fluent/plugin/out_falcon.rb', line 56 def set_header(req, tag, time, record) req end |
#shutdown ⇒ Object
42 43 44 |
# File 'lib/fluent/plugin/out_falcon.rb', line 42 def shutdown super end |
#start ⇒ Object
38 39 40 |
# File 'lib/fluent/plugin/out_falcon.rb', line 38 def start super end |