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 



105
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
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 105

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 



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

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 



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

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 



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

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
  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



242
243
244
245
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 242

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

#send_request(req, uri) ⇒ Object 



193
194
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
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 193

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
    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 



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

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 



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

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

#set_json_body(req, data) ⇒ Object 



179
180
181
182
 
# File 'lib/fluent/plugin/out_http_ext.rb', line 179

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

#shutdown ⇒ Object 



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

def shutdown
  super
end

#start ⇒ Object 



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

def start
  super
end