class Fluent::Plugin::LogIntelligenceOutput
Public Class Methods
new()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 37 def initialize super require 'http' require 'uri' end
Public Instance Methods
configure(conf)
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 153 def configure(conf) super validate_uri(@endpoint_url) @statuses = @http_retry_statuses.split(',').map { |status| status.to_i } @statuses = [] if @statuses.nil? @headers = retrieve_headers(conf) @http_client = Fluent::Plugin::HttpClient.new( @endpoint_url, @verify_ssl, @headers, @statuses, @open_timeout, @read_timeout, @log) end
create_lint_event(record)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 83 def create_lint_event(record) flattened_records = {} merged_records = {} if @flatten_hashes flattened_records = flatten_record(record, []) else flattened_records = record end keys = [] log = '' flattened_records.each do |key, value| begin next if value.nil? # LINT doesn't support duplicate fields, make unique names by appending underscore key = shorten_key(key) if keys.include?(key) value = merged_records[key] + " " + value end keys.push(key) key.force_encoding("utf-8") if value.is_a?(String) value.force_encoding("utf-8") end end if @log_text_keys.include?(key) if log != "#{value}" if log.empty? log = "#{value}" else log += " #{value}" end end else merged_records[key] = value end end merged_records["text"] = log if log == "\\n" {} else merged_records end end
flatten_record(record, prefix=[])
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 131 def flatten_record(record, prefix=[]) ret = {} case record when Hash record.each do |key, value| if @log_text_keys.include?(key) ret.merge!({key.to_s => value}) else ret.merge! flatten_record(value, prefix + [key.to_s]) end end when Array record.each do |value| ret.merge! flatten_record(value, prefix) end else return {prefix.join(@flatten_hashes_separator) => record} end ret end
retrieve_headers(conf)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 55 def retrieve_headers(conf) headers = {} conf.elements.each do |element| if @http_compress set_gzip_header(element) end if element.name == 'headers' if @bearer_token != '' element['Authorization'] = 'Bearer ' + @bearer_token end headers = element.to_hash end end headers end
set_gzip_header(element)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 71 def set_gzip_header(element) element['Content-Encoding'] = 'gzip' element end
shorten_key(key)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 76 def shorten_key(key) # LINT doesn't allow some characters in field 'name' # like '/', '-', '\', '.', etc. so replace them with @flatten_hashes_separator key = key.gsub(/[\/\.\-\\]/,@flatten_hashes_separator).downcase key end
shutdown()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 171 def shutdown super begin @http_client.close if @http_client rescue end end
start()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 167 def start super end
validate_uri(uri_string)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 43 def validate_uri(uri_string) unless uri_string =~ /^#{URI.regexp}$/ fail Fluent::ConfigError, 'endpoint_url invalid' end begin @uri = URI.parse(uri_string) rescue URI::InvalidURIError raise Fluent::ConfigError, 'endpoint_url invalid' end end
write(chunk)
click to toggle source
# File lib/fluent/plugin/out_vmware_log_intelligence.rb, line 179 def write(chunk) 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 data = [] chunk.each do |time, record| data << create_lint_event(record) end if @http_compress gzip_body = Zlib::GzipWriter.new(StringIO.new) gzip_body << data.to_json @http_client.post(gzip_body.close.string) else @last_request_time = Time.now.to_f @http_client.post(JSON.dump(data)) end end