b0VIM 7.4 `Ul z ubuntu ip-172-31-28-228 ~ubuntu/fluent-plugin-http-record-modfier/lib/fluent/plugin/filter_http_record_modifier.rb
3210 #"! U tp p x q 8 ad
p j U < ( c /
m
4
V U = 1 d
P
,
[ Q P < e F W . v n V 6 L ' o ? p O & placeholders = { tag_suffix = tag_suffix(tag_parts) tag_prefix = tag_prefix(tag_parts) tag_parts = tag.split('.') def format_params(tag, time, record) end clean clean['body'] = body end end clean = Yajl.load(res.body) if body.is_a? Hash body = Yajl.load(body) if res.content_type == 'application/json' body = res.body clean = {} def deserialize_body(res) end log.debug "map:#{@map} record:#{last_record} placeholders:#{placeholders}" log.warn_backtrace log.warn "failed to reform records", :error_class => e.class, :error => e.message rescue => e new_es end new_es.add(time, new_record) end time = new_record[@renew_time_key].to_i if @renew_time_key && new_record.has_key?(@renew_time_key) new_record = reform(time, record, placeholders, body) body = deserialize_body(res) res = send_request(req, uri) req, uri = create_request(tag, time, record) last_record = record # for debug log es.each do |time, record| last_record = nil } 'hostname' => @hostname, 'tag_suffix' => tag_suffix, 'tag_prefix' => tag_prefix, 'tag_parts' => tag_parts, 'tag' => tag, placeholders = { tag_suffix = tag_suffix(tag_parts) tag_prefix = tag_prefix(tag_parts) tag_parts = tag.split('.') new_es = MultiEventStream.new def filter_stream(tag, es) end @hostname = Socket.gethostname }) :auto_typecast => @auto_typecast, :log => log, @placeholder_expander = PlaceholderExpander.new({ end @keep_keys = @keep_keys.split(',') raise Fluent::ConfigError, "`renew_record` must be true to use `keep_keys`" unless @renew_record if @keep_keys end @remove_keys = @remove_keys.split(',') if @remove_keys end end @maped_params[k] = parse_value(v) element.has_key?(k) # to suppress unread configuration warning element.each_pair do |k, v| conf.elements.select { |element| element.name == 'params' }.each do |element| # directive @maped_params = {} end end @map[k] = parse_value(v) element.has_key?(k) # to suppress unread configuration warning element.each_pair do |k, v| conf.elements.select { |element| element.name == 'record' }.each do |element| # directive @map = {} @method ||= conf['method'] super def configure(conf) config_param :raise_on_error, :bool, :default => true config_param :password, :string, :default => '' config_param :username, :string, :default => '' config_param :authentication, :string, :default => nil config_param :serializer, :string, :default => :form config_param :endpoint_url, :string config_param :auto_typecast, :bool, :default => false # false for lower version compatibility config_param :rene_record, :bool, :default => false config_param :method, :string, :default => :get config_param :keep_keys, :string, :default => nil config_param :remove_keys, :string, :default => nil #Based on Filter Record Tranformer that is built-in on Fluent end super require 'uri' require 'net/http' require 'yajl' require 'socket' def initialize Plugin.register_filter('http_record_modifier', self) class HttpRecordModifier < Filter module Fluent ad k g 8 U 7
v @
}
g
D
_ V %
n
F
p T J @ ? 1 y q k g f end end end end end log.warn "unknown placeholder `#{placeholder}` found" unless @placeholders.include?(placeholder) def log_unknown_placeholder(placeholder) private end } @placeholders[$1] log_unknown_placeholder($1) str.gsub(/(\${[^}]+}|__[A-Z_]+__)/) { end end return @placeholders[single_placeholder_matched[1]] log_unknown_placeholder($1) if single_placeholder_matched single_placeholder_matched = str.match(/\A(\${[^}]+}|__[A-Z_]+__)\z/) if @auto_typecast and !force_stringify def expand(str, force_stringify=false) end placeholder.store("${#{before}}", value) # string, interger, float, and others? end end } crawl_placeholder(v, placeholder, "#{before}[#{idx-size}]", limit - 1) #suport [-1] crawl_placeholder(v, placeholder, "#{before}[#{idx}]", limit - 1) value.each_with_index { |v, idx| size = value.size elsif value.kind_of?(Array) # tag_parts, etc value.each {|key, v| crawl_placeholder(v, placeholder, "#{before}.#{key}", limit - 1)} if value.kind_of?(Hash) if limit >= 0 def crawl_placeholder (value, placeholder, before, limit = 50) end @placeholders = placeholders end end placeholders.store("${#{key}}", value) else # string, interger, float, and others? } placeholders.store("${#{key}[#{idx-size}]}", v) # support [-1] placeholders.store("${#{key}[#{idx}]}", v) value.each_with_index { |v, idx| size = value.size if value.kind_of?(Array) # tag_parts, etc opts.each do |key, value| } record.each {|key, value| crawl_placeholder(value, placeholders, "#{key}")