Class: Fluent::Plugin::HttpInput
- Defined in:
- lib/fluent/plugin/in_http.rb
Defined Under Namespace
Classes: Handler, KeepaliveManager
Constant Summary collapse
- EMPTY_GIF_IMAGE =
"GIF89a\u0001\u0000\u0001\u0000\x80\xFF\u0000\xFF\xFF\xFF\u0000\u0000\u0000,\u0000\u0000\u0000\u0000\u0001\u0000\u0001\u0000\u0000\u0002\u0002D\u0001\u0000;".force_encoding("UTF-8")
- EVENT_RECORD_PARAMETER =
'_event_record'
- RES_TEXT_HEADER =
{'Content-Type' => 'text/plain'}.freeze
- RESPONSE_200 =
["200 OK".freeze, RES_TEXT_HEADER, "".freeze].freeze
- RESPONSE_204 =
["204 No Content".freeze, {}.freeze].freeze
- RESPONSE_IMG =
["200 OK".freeze, {'Content-Type'=>'image/gif; charset=utf-8'}.freeze, EMPTY_GIF_IMAGE].freeze
- RES_400_STATUS =
"400 Bad Request".freeze
- RES_500_STATUS =
"500 Internal Server Error".freeze
Constants included from Configurable
Configurable::CONFIG_TYPE_REGISTRY
Instance Attribute Summary
Attributes included from Fluent::PluginLoggerMixin
Attributes inherited from Base
Instance Method Summary collapse
- #close ⇒ Object
- #configure(conf) ⇒ Object
-
#initialize ⇒ HttpInput
constructor
A new instance of HttpInput.
- #multi_workers_ready? ⇒ Boolean
- #on_request(path_info, params) ⇒ Object
- #start ⇒ Object
Methods included from Fluent::PluginHelper::Mixin
Methods included from Fluent::PluginLoggerMixin
Methods included from Fluent::PluginId
#plugin_id, #plugin_id_configured?, #plugin_id_for_test?, #plugin_root_dir, #stop
Methods inherited from Base
#after_shutdown, #after_shutdown?, #after_start, #after_started?, #before_shutdown, #before_shutdown?, #called_in_test?, #closed?, #configured?, #context_router, #context_router=, #fluentd_worker_id, #has_router?, #inspect, #plugin_root_dir, #reloadable_plugin?, #shutdown, #shutdown?, #started?, #stop, #stopped?, #string_safe_encoding, #terminate, #terminated?
Methods included from SystemConfig::Mixin
#system_config, #system_config_override
Methods included from Configurable
#config, #configure_proxy_generate, #configured_section_create, included, lookup_type, register_type
Constructor Details
#initialize ⇒ HttpInput
Returns a new instance of HttpInput.
92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
# File 'lib/fluent/plugin/in_http.rb', line 92 def initialize super @km = nil @format_name = nil @parser_time_key = nil # default parsers @parser_msgpack = nil @parser_json = nil @default_time_parser = nil @default_keep_time_key = nil @float_time_parser = nil # <parse> configured parser @custom_parser = nil end |
Instance Method Details
#close ⇒ Object
182 183 184 185 |
# File 'lib/fluent/plugin/in_http.rb', line 182 def close server_wait_until_stop super end |
#configure(conf) ⇒ Object
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 |
# File 'lib/fluent/plugin/in_http.rb', line 110 def configure(conf) compat_parameters_convert(conf, :parser) super m = if @parser_configs.first['@type'] == 'in_http' @parser_msgpack = parser_create(usage: 'parser_in_http_msgpack', type: 'msgpack') @parser_msgpack.time_key = nil @parser_msgpack.estimate_current_event = false @parser_json = parser_create(usage: 'parser_in_http_json', type: 'json') @parser_json.time_key = nil @parser_json.estimate_current_event = false default_parser = parser_create(usage: '') @format_name = 'default' @parser_time_key = default_parser.time_key @default_time_parser = default_parser.get_time_parser @default_keep_time_key = default_parser.keep_time_key method(:parse_params_default) else @custom_parser = parser_create @format_name = @parser_configs.first['@type'] @parser_time_key = @custom_parser.time_key method(:parse_params_with_parser) end self.singleton_class.module_eval do define_method(:parse_params, m) end end |
#multi_workers_ready? ⇒ Boolean
164 165 166 |
# File 'lib/fluent/plugin/in_http.rb', line 164 def multi_workers_ready? true end |
#on_request(path_info, params) ⇒ Object
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 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 |
# File 'lib/fluent/plugin/in_http.rb', line 194 def on_request(path_info, params) begin path = path_info[1..-1] # remove / tag = path.split('/').join('.') record_time, record = parse_params(params) # Skip nil record if record.nil? log.debug { "incoming event is invalid: path=#{path_info} params=#{params.to_json}" } if @respond_with_empty_img return RESPONSE_IMG else if @use_204_response return RESPONSE_204 else return RESPONSE_200 end end end mes = nil # Support batched requests if record.is_a?(Array) mes = Fluent::MultiEventStream.new record.each do |single_record| add_params_to_record(single_record, params) if param_time = params['time'] param_time = param_time.to_f single_time = param_time.zero? ? Fluent::EventTime.now : @float_time_parser.parse(param_time) elsif @custom_parser single_time = @custom_parser.parse_time(single_record) single_time, single_record = @custom_parser.convert_values(single_time, single_record) else single_time = convert_time_field(single_record) end mes.add(single_time, single_record) end else add_params_to_record(record, params) time = if param_time = params['time'] param_time = param_time.to_f param_time.zero? ? Fluent::EventTime.now : @float_time_parser.parse(param_time) else if record_time.nil? convert_time_field(record) else record_time end end end rescue => e if @dump_error_log log.error "failed to process request", error: e end return [RES_400_STATUS, RES_TEXT_HEADER, "400 Bad Request\n#{e}\n"] end # TODO server error begin if mes router.emit_stream(tag, mes) else router.emit(tag, time, record) end rescue => e if @dump_error_log log.error "failed to emit data", error: e end return [RES_500_STATUS, RES_TEXT_HEADER, "500 Internal Server Error\n#{e}\n"] end if @respond_with_empty_img return RESPONSE_IMG else if @use_204_response return RESPONSE_204 else return RESPONSE_200 end end end |
#start ⇒ Object
168 169 170 171 172 173 174 175 176 177 178 179 180 |
# File 'lib/fluent/plugin/in_http.rb', line 168 def start @_event_loop_run_timeout = @blocking_timeout super log.debug "listening http", bind: @bind, port: @port @km = KeepaliveManager.new(@keepalive_timeout) event_loop_attach(@km) server_create_connection(:in_http, @port, bind: @bind, backlog: @backlog, &method(:on_server_connect)) @float_time_parser = Fluent::NumericTimeParser.new(:float) end |