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