Class: Fluent::HTTPOutput

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

Instance Method Summary collapse

Constructor Details

#initialize ⇒ HTTPOutput

Returns a new instance of HTTPOutput.



67
68
69
70
71
72
73
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 67

def initialize
  super
  require 'net/http'
  require 'uri'
  require 'yajl'
  require 'set'
end

Instance Method Details

#configure(conf) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 106

def configure(conf)
  super

  serializers = [:json, :form]
  @serializer = if serializers.include? @serializer.intern
                  @serializer.intern
                else
                  :form
                end

  http_methods = [:get, :put, :post, :delete]
  @http_method = if http_methods.include? @http_method.intern
                  @http_method.intern
                else
                  :post
                end

  @ignore_http_status_code = if @ignore_http_status_code.nil?
                        [].to_set
                      else
                        StatusCodeParser.convert(@ignore_http_status_code)
                      end

  @auth = case @authentication
          when 'basic' then :basic
          else
            :none
          end
  @headers = {}
  conf.elements.each do |element|
    if element.name == 'headers'
      @headers = element.to_hash
    end
  end
end

#create_request(tag, time, record) ⇒ Object 



186
187
188
189
190
191
192
193
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 186

def create_request(tag, time, record)
  url = format_url(tag, time, record)
  uri = URI.parse(url)
  req = Net::HTTP.const_get(@http_method.to_s.capitalize).new(uri.path)
  set_body(req, tag, time, record)
  set_header(req, tag, time, record)
  return req, uri
end

#emit(tag, es, chain) ⇒ Object 



252
253
254
255
256
257
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 252

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 



150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 150

def format_url(tag, time, record)
  '''
  replace format string to value
  example
    /test/<data> =(use {data: 1})> /test/1
    /test/<hash.data> =(use {hash:{data:2}})> /test/2
  '''
  result_url = @endpoint_url
  return result_url unless record.is_a? Hash
  record.each_deep do |key_dir, value|
    result_url = result_url.gsub(/<#{key_dir.join(".")}>/, value.to_s)
  end
  return result_url
end

#handle_record(tag, time, record) ⇒ Object

end send_request



247
248
249
250
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 247

def handle_record(tag, time, record)
  req, uri = create_request(tag, time, record)
  send_request(req, uri)
end

#send_request(req, uri) ⇒ Object 



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 195

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
    client = Net::HTTP.new(uri.host, uri.port)
    if @use_ssl
      client.use_ssl = true
      client.ca_file = OpenSSL::X509::DEFAULT_CERT_FILE
      unless @verify_ssl
        client.verify_mode = OpenSSL::SSL::VERIFY_NONE
      end
    end
    res = client.start {|http|
      http.open_timeout = @open_timeout
      http.read_timeout = @read_timeout
      http.request(req)
    }
  rescue => e # rescue all StandardErrors
    # server didn't respond
    $log.warn "Net::HTTP.#{req.method.capitalize} raises exception: #{e.class}, '#{e.message}'"
    raise e if @raise_on_error
  else
     unless res and res.is_a?(Net::HTTPSuccess)
        res_summary = if res
                         "#{res.code} #{res.message} #{res.body}"
                      else
                         "res=nil"
                      end
        warning = "failed to #{req.method} #{uri} (#{res_summary})"
        $log.warn warning
        if @raise_on_http_failure
          unless @ignore_http_status_code.include?(res.code.to_i)
            raise warning
          else
            $log.debug "ignore http status code #{req.method}"
          end
        end

     end #end unless
  end # end begin
end

#set_body(req, tag, time, record) ⇒ Object 



165
166
167
168
169
170
171
172
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 165

def set_body(req, tag, time, record)
  if @serializer == :json
    set_json_body(req, record)
  else
    req.set_form_data(record)
  end
  req
end

#set_header(req, tag, time, record) ⇒ Object 



174
175
176
177
178
179
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 174

def set_header(req, tag, time, record)
  @headers.each do |key, value|
    req[key] = value
  end
  req
end

#set_json_body(req, data) ⇒ Object 



181
182
183
184
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 181

def set_json_body(req, data)
  req.body = Yajl.dump(data)
  req['Content-Type'] = 'application/json'
end

#shutdown ⇒ Object 



146
147
148
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 146

def shutdown
  super
end

#start ⇒ Object 



142
143
144
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 142

def start
  super
end