Class: Fluent::Plugin::OutOracleOCILogAnalytics

Inherits:
Output
  • Object
show all
Defined in:
lib/fluent/plugin/out_oci-logging-analytics.rb

Constant Summary collapse

MAX_FILES_PER_ZIP = 
100
@@logger = 
nil
@@loganalytics_client = 
nil
@@logger_config_errors = 
[]
@@default_log_level = 
'info'
@@default_log_rotation = 
'daily'
@@validated_log_size = 
nil
@@default_log_size =

1MB

1 * 1024 * 1024
@@default_number_of_logs = 
10

Instance Method Summary collapse

Constructor Details

#initialize ⇒ OutOracleOCILogAnalytics

Returns a new instance of OutOracleOCILogAnalytics.



111
112
113
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 111

def initialize
  super
end

Instance Method Details

#configure(conf) ⇒ Object

Raises:

  • (Fluent::ConfigError) 


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
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 215

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) ⇒ Object 



499
500
501
502
503
504
505
506
507
508
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 499

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) ⇒ Object 



510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 510

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) ⇒ Object

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.



748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 748

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) ⇒ Object 



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 265

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) ⇒ Object 



420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 420

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) ⇒ Object

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()



786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 786

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) ⇒ Object 



539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 539

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 ⇒ Object 



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 180

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 ⇒ Object 



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 115

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) ⇒ Object 



388
389
390
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 388

def is_number(field)
  true if Integer(field) rescue false
end

#is_valid(field) ⇒ Object 



307
308
309
310
311
312
313
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 307

def is_valid(field)
  if field.nil? || field.empty? then
    return false
  else
    return true
  end
end

#is_valid_log_age(param) ⇒ Object 



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 332

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) ⇒ Object 



348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 348

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) ⇒ Object 



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 315

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) ⇒ Object 



392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 392

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) ⇒ Object 



412
413
414
415
416
417
418
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 412

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) ⇒ Object 



467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 467

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) ⇒ Object 



529
530
531
532
533
534
535
536
537
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 529

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) ⇒ Object 



370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 370

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 ⇒ Object 



453
454
455
456
457
458
459
460
461
462
463
464
465
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 453

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) ⇒ Object 



829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 829

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) ⇒ Object

upload zipped stream to oci



847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 847

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) ⇒ Object

main entry point for FluentD’s flush_threads, which get invoked when a chunk is ready for flushing (see chunk limits and flush_intervals)



690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
 
# File 'lib/fluent/plugin/out_oci-logging-analytics.rb', line 690

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