class Fluent::Plugin::OutOracleOCILogAnalytics
Constants
- MAX_FILES_PER_ZIP
Public Class Methods
new()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 149 def initialize super end
Public Instance Methods
configure(conf)
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 253 def configure(conf) super initialize_logger initialize_loganalytics_client @@logger.error {"Error in config file : Buffer plugin must be of @type file."} unless buffer_config['@type'] == 'file' raise Fluent::ConfigError, "Error in config file : Buffer plugin must be of @type file." unless buffer_config['@type'] == 'file' is_mandatory_fields_valid,invalid_field_name = mandatory_field_validator if !is_mandatory_fields_valid @@logger.error {"Error in config file : invalid #{invalid_field_name}"} raise Fluent::ConfigError, "Error in config file : invalid #{invalid_field_name}" end # Get the chunk_limit_size from conf as it's not available in the buffer_config unless conf.elements(name: 'buffer').empty? buffer_conf = conf.elements(name: 'buffer').first chunk_limit_size_from_conf = buffer_conf['chunk_limit_size'] unless chunk_limit_size_from_conf.nil? log.debug "chunk limit size as per the configuration file is #{chunk_limit_size_from_conf}" case chunk_limit_size_from_conf.to_s when /([0-9]+)k/i chunk_limit_size_bytes = $~[1].to_i * 1024 when /([0-9]+)m/i chunk_limit_size_bytes = $~[1].to_i * (1024 ** 2) when /([0-9]+)g/i chunk_limit_size_bytes = $~[1].to_i * (1024 ** 3) when /([0-9]+)t/i chunk_limit_size_bytes = $~[1].to_i * (1024 ** 4) else raise Fluent::ConfigError, "error parsing chunk_limit_size" end log.debug "chunk limit size in bytes as per the configuration file is #{chunk_limit_size_bytes}" if !chunk_limit_size_bytes.between?(1048576, 2097152) raise Fluent::ConfigError, "chunk_limit_size must be between 1MB and 2MB" end end end if buffer_config.flush_interval < 10 raise Fluent::ConfigError, "flush_interval must be greater than or equal to 10sec" end @mutex = Mutex.new @num_flush_threads = Float(buffer_config.flush_thread_count) max_chunk_lifespan = (buffer_config.retry_type == :exponential_backoff) ? buffer_config.retry_wait * buffer_config.retry_exponential_backoff_base**(buffer_config.retry_max_times+1) - 1 : buffer_config.retry_wait * buffer_config.retry_max_times end
flatten(kubernetes_metadata)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 537 def flatten(kubernetes_metadata) kubernetes_metadata.each_with_object({}) do |(key, value), hash| hash[key] = value if value.is_a? Hash flatten(value).map do |hash_key, hash_value| hash["#{key}.#{hash_key}"] = hash_value end end end end
get_kubernetes_metadata(oci_la_metadata,record)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 548 def get_kubernetes_metadata(oci_la_metadata,record) if oci_la_metadata == nil oci_la_metadata = {} end kubernetes_metadata = flatten(record["kubernetes"]) kubernetes_metadata.each do |key, value| if kubernetes_metadata_keys_mapping.has_key?(key) if !is_valid(oci_la_metadata[kubernetes_metadata_keys_mapping[key]]) oci_la_metadata[kubernetes_metadata_keys_mapping[key]] = json_message_handler(value) end end end return oci_la_metadata rescue => ex @@logger.error {"Error occurred while getting kubernetes oci_la_metadata: error message: #{ex}"} return oci_la_metadata end
get_logSets_map_per_logGroupId(oci_la_log_group_id,records_per_logGroupId)
click to toggle source
Each oci_la_log_set will correspond to a separate file in the zip Only MAX_FILES_PER_ZIP
files are allowed per zip. Here we are grouping logSets so that if file_count reaches MAX_FILES_PER_ZIP
, these files will be considered for a separate zip file.
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 786 def get_logSets_map_per_logGroupId(oci_la_log_group_id,records_per_logGroupId) file_count = 0 oci_la_global_metadata = nil is_oci_la_global_metadata_assigned = false oci_la_log_set = nil records_per_logSet_map = Hash.new logSets_per_logGroupId_map = Hash.new records_per_logGroupId.group_by { |record| if !is_oci_la_global_metadata_assigned record_hash = record.keys.map {|x| [x,true]}.to_h if record_hash.has_key?("oci_la_global_metadata") oci_la_global_metadata = record['oci_la_global_metadata'] end is_oci_la_global_metadata_assigned = true end oci_la_log_set = record['oci_la_log_set'] (oci_la_log_set) }.map { |oci_la_log_set, records_per_logSet| if file_count % OutOracleOCILogAnalytics::MAX_FILES_PER_ZIP == 0 records_per_logSet_map = Hash.new end records_per_logSet_map[oci_la_log_set] = records_per_logSet file_count += 1 if file_count % OutOracleOCILogAnalytics::MAX_FILES_PER_ZIP == 0 logSets_per_logGroupId_map[file_count] = records_per_logSet_map end } logSets_per_logGroupId_map[file_count] = records_per_logSet_map return logSets_per_logGroupId_map,oci_la_global_metadata rescue => exc @@logger.error {"Error in mapping records to oci_la_log_set. oci_la_log_group_id: #{oci_la_log_group_id}, error message:#{exc}"} end
get_or_parse_logSet(unparsed_logSet, record, record_hash, is_tag_exists)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 303 def get_or_parse_logSet(unparsed_logSet, record, record_hash, is_tag_exists) oci_la_log_set = nil parsed_logSet = nil if !is_valid(unparsed_logSet) return nil end if record_hash.has_key?("oci_la_log_set_ext_regex") && is_valid(record["oci_la_log_set_ext_regex"]) parsed_logSet = unparsed_logSet.match(record["oci_la_log_set_ext_regex"]) #*******************************************TO-DO********************************************************** # Based on the observed behaviour, below cases are handled. We need to revisit this section. # When trying to apply regex on a String and getting a matched substring, observed couple of scenarios. # For oci_la_log_set_ext_regex value = '.*\\\\/([^\\\\.]{1,40}).*' this returns an array with both input string and matched pattern # For oci_la_log_set_ext_regex value = '[ \\\\w-]+?(?=\\\\.)' this returns an array with only matched pattern # For few cases, String is returned instead of an array. #*******************************************End of TO-DO*************************************************** if parsed_logSet!= nil # Based on the regex pattern, match is returning different outputs for same input. if parsed_logSet.is_a? String oci_la_log_set = parsed_logSet.encode("UTF-8") # When matched String is returned instead of an array. elsif parsed_logSet.length > 1 #oci_la_log_set_ext_regex '.*\\\\/([^\\\\.]{1,40}).*' this returns an array with both input string and matched pattern oci_la_log_set = parsed_logSet[1].encode("UTF-8") elsif parsed_logSet.length > 0 # oci_la_log_set_ext_regex '[ \\\\w-]+?(?=\\\\.)' this returns an array with only matched pattern oci_la_log_set = parsed_logSet[0].encode("UTF-8") #Encoding to handle escape characters else oci_la_log_set = nil end else oci_la_log_set = nil if is_tag_exists @@logger.error {"Error occurred while parsing oci_la_log_set : #{unparsed_logSet} with oci_la_log_set_ext_regex : #{record["oci_la_log_set_ext_regex"]}. Default oci_la_log_set will be assigned to all the records with tag : #{record["tag"]}."} else @@logger.error {"Error occurred while parsing oci_la_log_set : #{unparsed_logSet} with oci_la_log_set_ext_regex : #{record["oci_la_log_set_ext_regex"]}. Default oci_la_log_set will be assigned."} end end else oci_la_log_set = unparsed_logSet.force_encoding('UTF-8').encode("UTF-8") end return oci_la_log_set rescue => ex @@logger.error {"Error occurred while parsing oci_la_log_set : #{ex}. Default oci_la_log_set will be assigned."} return nil end
get_valid_metadata(oci_la_metadata)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 458 def get_valid_metadata(oci_la_metadata) if oci_la_metadata != nil if oci_la_metadata.is_a?(Hash) valid_metadata = Hash.new invalid_keys = [] oci_la_metadata.each do |key, value| if value != nil && !value.is_a?(Hash) && !value.is_a?(Array) if key != nil && !key.is_a?(Hash) && !key.is_a?(Array) valid_metadata[key] = value else invalid_keys << key end else invalid_keys << key end end if invalid_keys.length > 0 @@logger.warn {"Skipping the following oci_la_metadata/oci_la_global_metadata keys #{invalid_keys.compact.reject(&:empty?).join(',')} as the corresponding values are in invalid format."} end if valid_metadata.length > 0 return valid_metadata else return nil end else @@logger.warn {"Ignoring 'oci_la_metadata'/'oci_la_global_metadata' provided in the record_transformer filter as only key-value pairs are supported."} return nil end else return nil end end
get_zipped_stream(oci_la_log_group_id,oci_la_global_metadata,records_per_logSet_map)
click to toggle source
takes a fluentD chunk and converts it to an in-memory zipfile, populating metrics hash provided Any exception raised is passed into the metrics hash, to be re-thrown from write()
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 824 def get_zipped_stream(oci_la_log_group_id,oci_la_global_metadata,records_per_logSet_map) begin current, = Time.now current_f, current_s = current.to_f, current.strftime("%Y%m%dT%H%M%S%9NZ") number_of_records = 0 noOfFilesGenerated = 0 zippedstream = Zip::OutputStream.write_buffer { |zos| records_per_logSet_map.each do |oci_la_log_set,records_per_logSet| lrpes_for_logEvents = records_per_logSet.group_by { |record| [ record['oci_la_metadata'], record['oci_la_entity_id'], record['oci_la_entity_type'], record['oci_la_log_source_name'] , record['oci_la_log_path'] ]}.map { |lrpe_key, records_per_lrpe| number_of_records += records_per_lrpe.length LogEvents.new(lrpe_key, records_per_lrpe) } noOfFilesGenerated = noOfFilesGenerated +1 if is_valid(oci_la_log_set) then nextEntry = oci_la_log_group_id+ "_#{current_s}" +"_"+ noOfFilesGenerated.to_s + "_logSet=" + oci_la_log_set + ".json" #oci_la_log_group_id + ".json" else nextEntry = oci_la_log_group_id + "_#{current_s}" +"_"+ noOfFilesGenerated.to_s + ".json" end @@logger.debug {"Added entry #{nextEntry} for oci_la_log_set #{oci_la_log_set} into the zip."} zos.put_next_entry(nextEntry) logEventsJsonFinal = LogEventsJson.new(oci_la_global_metadata,lrpes_for_logEvents) zos.write logEventsJsonFinal.to_hash.to_json end } zippedstream.rewind if @dump_zip_file save_zip_to_local(oci_la_log_group_id,zippedstream,current_s) end #zippedstream.rewind if records.length > 0 #reposition buffer pointer to the beginning rescue => exc @@logger.error {"Error in generating payload. oci_la_log_group_id: #{oci_la_log_group_id}, error message:#{exc}"} end return zippedstream,number_of_records end
group_by_logGroupId(chunk)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 577 def group_by_logGroupId(chunk) begin current = Time.now current_f, current_s = current.to_f, current.strftime("%Y%m%dT%H%M%S%9NZ") records = [] count = 0 invalid_tag_set = Set.new incoming_records_per_tag = Hash.new invalid_records_per_tag = Hash.new tags_per_logGroupId = Hash.new tag_logSet_map = Hash.new tag_metadata_map = Hash.new chunk.each do |time, record| if !record.nil? record_hash = record.keys.map {|x| [x,true]}.to_h is_tag_exists = false if record_hash.has_key?("tag") && is_valid(record["tag"]) is_tag_exists = true end if is_tag_exists && incoming_records_per_tag.has_key?(record["tag"]) incoming_records_per_tag[record["tag"]] += 1 elsif is_tag_exists incoming_records_per_tag[record["tag"]] = 1 end #For any given tag, if one record fails (mandatory fields validation) then all the records from that source will be ignored if is_tag_exists && invalid_tag_set.include?(record["tag"]) invalid_records_per_tag[record["tag"]] += 1 next #This tag is already present in the invalid_tag_set, so ignoring the message. end #Setting tag/default value for oci_la_log_path, when not provided in config file. if !record_hash.has_key?("oci_la_log_path") || !is_valid(record["oci_la_log_path"]) if is_tag_exists record["oci_la_log_path"] = record["tag"] else record["oci_la_log_path"] = 'UNDEFINED' end end #Extracting oci_la_log_set when oci_la_log_set_key and oci_la_log_set_ext_regex is provided. #1) oci_la_log_set param is not provided in config file and above logic not executed. #2) Valid oci_la_log_set_key + No oci_la_log_set_ext_regex #a) Valid key available in record with oci_la_log_set_key corresponding value (oci_la_log_set_key is a key in config file) --> oci_la_log_set #b) No Valid key available in record with oci_la_log_set_key corresponding value --> nil #3) Valid key available in record with oci_la_log_set_key corresponding value + Valid oci_la_log_set_ext_regex #a) Parse success --> parsed oci_la_log_set #b) Parse failure --> nil (as oci_la_log_set value) #4) No oci_la_log_set_key --> do nothing --> nil #Extracting oci_la_log_set when oci_la_log_set and oci_la_log_set_ext_regex is provided. #1) Valid oci_la_log_set + No oci_la_log_set_ext_regex --> oci_la_log_set #2) Valid oci_la_log_set + Valid oci_la_log_set_ext_regex #a) Parse success --> parsed oci_la_log_set #b) Parse failure --> nil (as oci_la_log_set value) #3) No oci_la_log_set --> do nothing --> nil unparsed_logSet = nil processed_logSet = nil if is_tag_exists && tag_logSet_map.has_key?(record["tag"]) record["oci_la_log_set"] = tag_logSet_map[record["tag"]] else if record_hash.has_key?("oci_la_log_set_key") if is_valid(record["oci_la_log_set_key"]) && record_hash.has_key?(record["oci_la_log_set_key"]) if is_valid(record[record["oci_la_log_set_key"]]) unparsed_logSet = record[record["oci_la_log_set_key"]] processed_logSet = get_or_parse_logSet(unparsed_logSet,record, record_hash,is_tag_exists) end end end if !is_valid(processed_logSet) && record_hash.has_key?("oci_la_log_set") if is_valid(record["oci_la_log_set"]) unparsed_logSet = record["oci_la_log_set"] processed_logSet = get_or_parse_logSet(unparsed_logSet,record, record_hash,is_tag_exists) end end record["oci_la_log_set"] = processed_logSet tag_logSet_map[record["tag"]] = processed_logSet end unless is_valid_record(record_hash,record) if is_tag_exists invalid_tag_set.add(record["tag"]) invalid_records_per_tag[record["tag"]] = 1 end next end #This will check for null or empty messages and only that record will be ignored. if !is_valid(record["message"]) if is_tag_exists @@logger.warn {"'message' field has empty value, Skipping records associated with tag : #{record["tag"]}."} if invalid_records_per_tag.has_key?(record["tag"]) invalid_records_per_tag[record["tag"]] += 1 else invalid_records_per_tag[record["tag"]] = 1 end else @@logger.warn {"'message' field has empty value, Skipping record."} end next else record["message"] = json_message_handler(record["message"]) end if record_hash.has_key?("kubernetes") record["oci_la_metadata"] = get_kubernetes_metadata(record["oci_la_metadata"],record) end if tag_metadata_map.has_key?(record["tag"]) record["oci_la_metadata"] = tag_metadata_map[record["tag"]] else if record_hash.has_key?("oci_la_metadata") record["oci_la_metadata"] = get_valid_metadata(record["oci_la_metadata"]) tags_per_logGroupId[record["tag"]] = record["oci_la_metadata"] else tags_per_logGroupId[record["tag"]] = nil end end if is_tag_exists if tags_per_logGroupId.has_key?(record["oci_la_log_group_id"]) if !tags_per_logGroupId[record["oci_la_log_group_id"]].include?(record["tag"]) tags_per_logGroupId[record["oci_la_log_group_id"]] += ", "+record["tag"] end else tags_per_logGroupId[record["oci_la_log_group_id"]] = record["tag"] end end records << record else @@logger.trace {"Record is nil, ignoring the record"} end end @@logger.debug {"records.length:#{records.length}"} lrpes_for_logGroupId = {} records.group_by{|record| oci_la_log_group_id = record['oci_la_log_group_id'] (oci_la_log_group_id) }.map {|oci_la_log_group_id, records_per_logGroupId| lrpes_for_logGroupId[oci_la_log_group_id] = records_per_logGroupId } rescue => ex @@logger.error {"Error occurred while grouping records by oci_la_log_group_id:#{ex.inspect}"} end return incoming_records_per_tag,invalid_records_per_tag,tags_per_logGroupId,lrpes_for_logGroupId end
initialize_loganalytics_client()
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 218 def initialize_loganalytics_client() if is_valid(@config_file_location) @auth_type = "ConfigFile" end case @auth_type when "InstancePrincipal" instance_principals_signer = OCI::Auth::Signers::InstancePrincipalsSecurityTokenSigner.new @@loganalytics_client = OCI::LogAnalytics::LogAnalyticsClient.new(config: OCI::Config.new, signer: instance_principals_signer) when "ConfigFile" my_config = OCI::ConfigFileLoader.load_config(config_file_location: @config_file_location, profile_name: @profile_name) if is_valid(endpoint) @@loganalytics_client = OCI::LogAnalytics::LogAnalyticsClient.new(config:my_config, endpoint:@endpoint) @@logger.info {"loganalytics_client initialised with endpoint: #{@endpoint}"} else @@loganalytics_client = OCI::LogAnalytics::LogAnalyticsClient.new(config:my_config) end else raise Fluent::ConfigError, "Invalid authType @auth_type, authType must be either InstancePrincipal or ConfigFile." abort end if is_valid(@proxy_ip) && is_number(@proxy_port) if is_valid(@proxy_username) && is_valid(@proxy_password) @@loganalytics_client.api_client.proxy_settings = OCI::ApiClientProxySettings.new(@proxy_ip, @proxy_port, @proxy_username, @proxy_password) else @@loganalytics_client.api_client.proxy_settings = OCI::ApiClientProxySettings.new(@proxy_ip, @proxy_port) end end rescue => ex @@logger.error {"Error occurred while initializing LogAnalytics Client: authType: #{@auth_type}, errorMessage: #{ex}"} end
initialize_logger()
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 153 def initialize_logger() filename = nil is_default_log_location = false if is_valid(@plugin_log_location) filename = @plugin_log_location[-1] == '/' ? @plugin_log_location : @plugin_log_location +'/' else is_default_log_location = true end if !is_valid_log_level(@plugin_log_level) @plugin_log_level = @@default_log_level end oci_fluent_output_plugin_log = nil if is_default_log_location oci_fluent_output_plugin_log = 'oci-logging-analytics.log' else oci_fluent_output_plugin_log = filename+'oci-logging-analytics.log' end logger_config = nil if is_valid_number_of_logs(@plugin_log_file_count) && is_valid_log_size(@plugin_log_file_size) # When customer provided valid log_file_count and log_file_size. # logger will rotate with max log_file_count with each file having max log_file_size. # Older logs purged automatically. @@logger = Logger.new(oci_fluent_output_plugin_log, @plugin_log_file_count, @@validated_log_size) logger_config = 'USER_CONFIG' elsif is_valid_log_rotation(@plugin_log_rotation) # When customer provided only log_rotation. # logger will create a new log based on log_rotation (new file everyday if the rotation is daily). # This will create too many logs over a period of time as log purging is not done. @@logger = Logger.new(oci_fluent_output_plugin_log, @plugin_log_rotation) logger_config = 'FALLBACK_CONFIG' else # When customer provided invalid log config, default config is considered. # logger will rotate with max default log_file_count with each file having max default log_file_size. # Older logs purged automatically. @@logger = Logger.new(oci_fluent_output_plugin_log, @@default_number_of_logs, @@default_log_size) logger_config = 'DEFAULT_CONFIG' end logger_set_level(@plugin_log_level) @@logger.info {"Initializing oci-logging-analytics plugin"} if is_default_log_location @@logger.info {"plugin_log_location is not specified. oci-logging-analytics.log will be generated under directory from where fluentd is executed."} end case logger_config when 'USER_CONFIG' @@logger.info {"Logger for oci-logging-analytics.log is initialized with config values log size: #{@plugin_log_file_size}, number of logs: #{@plugin_log_file_count}"} when 'FALLBACK_CONFIG' @@logger.info {"Logger for oci-logging-analytics.log is initialized with log rotation: #{@plugin_log_rotation}"} when 'DEFAULT_CONFIG' @@logger.info {"Logger for oci-logging-analytics.log is initialized with default config values log size: #{@@default_log_size}, number of logs: #{@@default_number_of_logs}"} end if @@logger_config_errors.length > 0 @@logger_config_errors. each {|logger_config_error| @@logger.warn {"#{logger_config_error}"} } end if is_valid_log_age(@plugin_log_age) @@logger.warn {"'plugin_log_age' field is deprecated. Use 'plugin_log_file_size' and 'plugin_log_file_count' instead."} end end
is_number(field)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 426 def is_number(field) true if Integer(field) rescue false end
is_valid(field)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 345 def is_valid(field) if field.nil? || field.empty? then return false else return true end end
is_valid_log_age(param)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 370 def is_valid_log_age(param) if !is_valid(param) return false end case param.downcase when "daily" return true when "weekly" return true when "monthly" return true else return false end end
is_valid_log_level(param)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 386 def is_valid_log_level(param) if !is_valid(param) return false end case param.upcase when "DEBUG" return true when "INFO" return true when "WARN" return true when "ERROR" return true when "FATAL" return true when "UNKNOWN" return true else return false end end
is_valid_log_rotation(log_rotation)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 353 def is_valid_log_rotation(log_rotation) if !is_valid(log_rotation) return false end case log_rotation.downcase when "daily" return true when "weekly" return true when "monthly" return true else @@logger_config_error << "Only 'daily'/'weekly'/'monthly' are supported for 'plugin_log_rotation'." return false end end
is_valid_log_size(log_size)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 430 def is_valid_log_size(log_size) if log_size != nil case log_size.to_s when /([0-9]+)k/i log_size = $~[1].to_i * 1024 when /([0-9]+)m/i log_size = $~[1].to_i * (1024 ** 2) when /([0-9]+)g/i log_size = $~[1].to_i * (1024 ** 3) else @@logger_config_errors << "plugin_log_file_size must be greater than 1KB." return false end @@validated_log_size = log_size return true else return false end end
is_valid_number_of_logs(number_of_logs)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 450 def is_valid_number_of_logs(number_of_logs) if !is_number(number_of_logs) || number_of_logs < 1 @@logger_config_errors << "plugin_log_file_count must be greater than zero" return false end return true end
is_valid_record(record_hash,record)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 505 def is_valid_record(record_hash,record) begin if !record_hash.has_key?("message") if record_hash.has_key?("tag") @@logger.warn {"Invalid records associated with tag : #{record["tag"]}. 'message' field is not present in the record."} else @@logger.info {"InvalidRecord: #{record}"} @@logger.warn {"Invalid record. 'message' field is not present in the record."} end return false elsif !record_hash.has_key?("oci_la_log_group_id") || !is_valid(record["oci_la_log_group_id"]) if record_hash.has_key?("tag") @@logger.warn {"Invalid records associated with tag : #{record["tag"]}.'oci_la_log_group_id' must not be empty. Skipping all the records associated with the tag"} else @@logger.warn {"Invalid record.'oci_la_log_group_id' must not be empty"} end return false elsif !record_hash.has_key?("oci_la_log_source_name") || !is_valid(record["oci_la_log_source_name"]) if record_hash.has_key?("tag") @@logger.warn {"Invalid records associated with tag : #{record["tag"]}.'oci_la_log_source_name' must not be empty. Skipping all the records associated with the tag"} else @@logger.warn {"Invalid record.'oci_la_log_source_name' must not be empty"} end return false else return true end end end
json_message_handler(message)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 567 def json_message_handler(message) if message.is_a?(Hash) return JSON.generate(message) else return message end rescue => ex return message end
logger_set_level(param)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 408 def logger_set_level(param) # DEBUG < INFO < WARN < ERROR < FATAL < UNKNOWN case @plugin_log_level.upcase when "DEBUG" @@logger.level = Logger::DEBUG when "INFO" @@logger.level = Logger::INFO when "WARN" @@logger.level = Logger::WARN when "ERROR" @@logger.level = Logger::ERROR when "FATAL" @@logger.level = Logger::FATAL when "UNKNOWN" @@logger.level = Logger::UNKNOWN end end
mandatory_field_validator()
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 491 def mandatory_field_validator begin if !is_valid(@namespace) return false,'namespace' elsif !is_valid(@config_file_location) && @auth_type == 'ConfigFile' return false,'config_file_location' elsif !is_valid(@profile_name) && @auth_type == 'ConfigFile' return false,'profile_name' else return true,nil end end end
save_zip_to_local(oci_la_log_group_id, zippedstream, current_s)
click to toggle source
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 867 def save_zip_to_local(oci_la_log_group_id, zippedstream, current_s) begin fileName = oci_la_log_group_id+"_"+current_s+'.zip' fileLocation = @zip_file_location+fileName file = File.open(fileLocation, "w") file.write(zippedstream.sysread) rescue => ex @@logger.error {"Error occurred while saving zip file. oci_la_log_group_id: oci_la_log_group_id, fileLocation: @zip_file_location fileName: fileName error message: #{ex}"} ensure file.close unless file.nil? end end
upload_to_oci(oci_la_log_group_id, number_of_records, zippedstream)
click to toggle source
upload zipped stream to oci
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 885 def upload_to_oci(oci_la_log_group_id, number_of_records, zippedstream) begin opts = {payload_type: "ZIP"} response = @@loganalytics_client.upload_log_events_file(namespace_name=@namespace, logGroupId=oci_la_log_group_id , uploadLogEventsFileDetails=zippedstream, opts) if !response.nil? && response.status == 200 then headers = response.headers @@logger.info {"The payload has been successfully uploaded to logAnalytics - oci_la_log_group_id: #{oci_la_log_group_id}, ConsumedRecords: #{number_of_records}, Date: #{headers['date']}, Time: #{headers['timecreated']}, opc-request-id: #{headers['opc-request-id']}, opc-object-id: #{headers['opc-object-id']}"} end rescue OCI::Errors::ServiceError => serviceError case serviceError.status_code when 400 @@logger.error {"oci upload exception : Error while uploading the payload. Invalid/Incorrect/missing Parameter - opc-request-id:#{serviceError.request_id}"} if plugin_retry_on_4xx raise serviceError end when 401 @@logger.error {"oci upload exception : Error while uploading the payload. Not Authenticated. opc-request-id:#{serviceError.request_id} message: #{serviceError.message}"} if plugin_retry_on_4xx raise serviceError end when 404 @@logger.error {"oci upload exception : Error while uploading the payload. Authorization failed for given oci_la_log_group_id against given Tenancy Namespace. oci_la_log_group_id: #{oci_la_log_group_id} Namespace: #{@namespace} opc-request-id: #{serviceError.request_id} message: #{serviceError.message}"} if plugin_retry_on_4xx raise serviceError end when 429 @@logger.error {"oci upload exception : Error while uploading the payload. Too Many Requests - opc-request-id:#{serviceError.request_id}"} raise serviceError when 500 @@logger.error {"oci upload exception : Error while uploading the payload. Internal Server Error - opc-request-id:#{serviceError.request_id}"} raise serviceError when 502 @@logger.error {"oci upload exception : Error while uploading the payload. Bad Gateway - opc-request-id:#{serviceError.request_id}"} raise serviceError when 503 @@logger.error {"oci upload exception : Error while uploading the payload. Service unavailable - opc-request-id:#{serviceError.request_id}"} raise serviceError when 504 @@logger.error {"oci upload exception : Error while uploading the payload. Gateway Timeout - opc-request-id:#{serviceError.request_id}"} raise serviceError when 505 @@logger.error {"oci upload exception : Error while uploading the payload. HTTP Version Not Supported - opc-request-id:#{serviceError.request_id}"} raise serviceError else @@logger.error {"oci upload exception : Error while uploading the payload #{serviceError.message}"} raise serviceError end rescue => ex @@logger.error {"oci upload exception : Error while uploading the payload. #{ex}"} end end
write(chunk)
click to toggle source
main entry point for FluentD’s flush_threads, which get invoked when a chunk is ready for flushing (see chunk limits and flush_intervals)
# File lib/fluent/plugin/out_oci-logging-analytics.rb, line 728 def write(chunk) @@logger.info {"Received new chunk, started processing ..."} metrics = Hash.new metrics["count"] = 0 metrics["event"] = "zipping" metrics["bytes_in"] = chunk.bytesize metrics['records_dropped'] = 0 begin # 1) Create an in-memory zipfile for the given FluentD chunk # 2) Synchronization has been removed. See EMCLAS-28675 begin lrpes_for_logGroupId = {} incoming_records_per_tag,invalid_records_per_tag,tags_per_logGroupId,lrpes_for_logGroupId = group_by_logGroupId(chunk) valid_message_per_tag = Hash.new incoming_records_per_tag.each do |key,value| dropped_messages = (invalid_records_per_tag.has_key?(key)) ? invalid_records_per_tag[key].to_i : 0 valid_messages = value.to_i - dropped_messages valid_message_per_tag[key] = valid_messages if dropped_messages > 0 @@logger.info {"Messages: #{value.to_i} \t Valid: #{valid_messages} \t Invalid: #{dropped_messages} \t tag:#{key}"} end @@logger.debug {"Messages: #{value.to_i} \t Valid: #{valid_messages} \t Invalid: #{dropped_messages} \t tag:#{key}"} end if lrpes_for_logGroupId != nil && lrpes_for_logGroupId.length > 0 lrpes_for_logGroupId.each do |oci_la_log_group_id,records_per_logGroupId| begin tags = tags_per_logGroupId.key(oci_la_log_group_id) @@logger.info {"Generating payload with #{records_per_logGroupId.length} records for oci_la_log_group_id: #{oci_la_log_group_id}"} zippedstream = nil oci_la_log_set = nil logSets_per_logGroupId_map = Hash.new # Only MAX_FILES_PER_ZIP (100) files are allowed, which will be grouped and zipped. # Due to MAX_FILES_PER_ZIP constraint, for a oci_la_log_group_id, we can get more than one zip file and those many api calls will be made. logSets_per_logGroupId_map, oci_la_global_metadata = get_logSets_map_per_logGroupId(oci_la_log_group_id,records_per_logGroupId) if logSets_per_logGroupId_map != nil logSets_per_logGroupId_map.each do |file_count,records_per_logSet_map| zippedstream,number_of_records = get_zipped_stream(oci_la_log_group_id,oci_la_global_metadata,records_per_logSet_map) if zippedstream != nil zippedstream.rewind #reposition buffer pointer to the beginning upload_to_oci(oci_la_log_group_id, number_of_records, zippedstream) end end end ensure zippedstream&.close end end end end end end