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 



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
141
142
143
144
145
146
147
148
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 108

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

  @formatter = nil
  unless @format.empty?
    @formatter = Fluent::Plugin.new_formatter(@format)
    @formatter.configure(conf)
  end
end

#create_request(tag, time, record) ⇒ Object 



194
195
196
197
198
199
200
201
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 194

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 



260
261
262
263
264
265
266
267
268
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 260

def emit(tag, es, chain)
  es.each do |time, record|
    if @formatter
      record = @formatter.format(tag, time, record)
    end
    handle_record(tag, time, record)
  end
  chain.next
end

#format_url(tag, time, record) ⇒ Object 



158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 158

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



255
256
257
258
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 255

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

#send_request(req, uri) ⇒ Object 



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
246
247
248
249
250
251
252
253
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 203

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 



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

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 



182
183
184
185
186
187
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 182

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

#set_json_body(req, data) ⇒ Object 



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

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

#shutdown ⇒ Object 



154
155
156
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 154

def shutdown
  super
end

#start ⇒ Object 



150
151
152
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 150

def start
  super
end