lib/fluent/plugin/out_google_cloud.rb
def write(chunk)
      grouped_entries = group_log_entries_by_tag_and_local_resource_id(chunk)

      requests_to_send = []
      grouped_entries.each do |(tag, local_resource_id), arr|
        entries = []
        group_level_resource, group_level_common_labels =
          determine_group_level_monitored_resource_and_labels(
            tag, local_resource_id)

        arr.each do |time, record|
          entry_level_resource, entry_level_common_labels =
            determine_entry_level_monitored_resource_and_labels(
              group_level_resource, group_level_common_labels, record)

          is_json = false
          if @detect_json
            # Save the following fields if available, then clear them out to
            # allow for determining whether we should parse the log or message
            # field.
            preserved_keys = [
              'time',
              'severity',
              @http_request_key,
              @insert_id_key,
              @operation_key,
              @source_location_key,
              @span_id_key,
              @trace_key
            ]

            # If the log is json, we want to export it as a structured log
            # unless there is additional metadata that would be lost.
            record_json = nil
            if (record.keys - preserved_keys).length == 1
              %w(log message msg).each do |field|
                if record.key?(field)
                  record_json = parse_json_or_nil(record[field])
                end
              end
            end
            unless record_json.nil?
              # Propagate these if necessary. Note that we don't want to
              # override these keys in the JSON we've just parsed.
              preserved_keys.each do |key|
                record_json[key] ||= record[key] if record.key?(key)
              end

              record = record_json
              is_json = true
            end
          end

          ts_secs, ts_nanos = compute_timestamp(
            entry_level_resource.type, record, time)
          severity = compute_severity(
            entry_level_resource.type, record, entry_level_common_labels)

          entry = @construct_log_entry.call(entry_level_common_labels,
                                            entry_level_resource,
                                            severity,
                                            ts_secs,
                                            ts_nanos)

          trace = record.delete(@trace_key)
          entry.trace = compute_trace(trace) if trace

          span_id = record.delete(@span_id_key)
          entry.span_id = span_id if span_id
          insert_id = record.delete(@insert_id_key)
          entry.insert_id = insert_id if insert_id

          set_log_entry_fields(record, entry)
          set_payload(entry_level_resource.type, record, entry, is_json)

          entries.push(entry)
        end
        # Don't send an empty request if we rejected all the entries.
        next if entries.empty?

        log_name = "projects/#{@project_id}/logs/#{log_name(
          tag, group_level_resource)}"

        requests_to_send << {
          entries: entries,
          log_name: log_name,
          resource: group_level_resource,
          labels: group_level_common_labels
        }
      end

      if @split_logs_by_tag
        requests_to_send.each do |request|
          @write_request.call(request)
        end
      else
        # Combine all requests into one. The request level "log_name" will be
        # ported to the entry level. The request level "resource" and "labels"
        # are ignored as they should have been folded into the entry level
        # "resource" and "labels" already anyway.
        combined_entries = []
        requests_to_send.each do |request|
          request[:entries].each do |entry|
            # Modify entries in-place as they are not needed later on.
            entry.log_name = request[:log_name]
          end
          combined_entries.concat(request[:entries])
        end
        @write_request.call(entries: combined_entries) unless
          combined_entries.empty?
      end
    end
# Constants for service names, resource types and etc.
Constants for service names, resource types and etc.
Constants for service names resource types and etc .
def write(chunk)
      grouped_entries = group_log_entries_by_tag_and_local_resource_id(chunk)
      requests_to_send = []
      grouped_entries.each do |(tag, local_resource_id), arr|
        entries = []
        group_level_resource, group_level_common_labels =
          determine_group_level_monitored_resource_and_labels(
            tag, local_resource_id)
        arr.each do |time, record|
          entry_level_resource, entry_level_common_labels =
            determine_entry_level_monitored_resource_and_labels(
              group_level_resource, group_level_common_labels, record)
          is_json = false
          if @detect_json
            preserved_keys = [
              'time',
              'severity',
              @http_request_key,
              @insert_id_key,
              @operation_key,
              @source_location_key,
              @span_id_key,
              @trace_key
            ]
            record_json = nil
            if (record.keys - preserved_keys).length == 1
              %w(log message msg).each do |field|
                if record.key?(field)
                  record_json = parse_json_or_nil(record[field])
                end
              end
            end
            unless record_json.nil?
              preserved_keys.each do |key|
                record_json[key] ||= record[key] if record.key?(key)
              end
              record = record_json
              is_json = true
            end
          end
          ts_secs, ts_nanos = compute_timestamp(
            entry_level_resource.type, record, time)
          severity = compute_severity(
            entry_level_resource.type, record, entry_level_common_labels)
          entry = @construct_log_entry.call(entry_level_common_labels,
                                            entry_level_resource,
                                            severity,
                                            ts_secs,
                                            ts_nanos)
          trace = record.delete(@trace_key)
          entry.trace = compute_trace(trace) if trace
          span_id = record.delete(@span_id_key)
          entry.span_id = span_id if span_id
          insert_id = record.delete(@insert_id_key)
          entry.insert_id = insert_id if insert_id
          set_log_entry_fields(record, entry)
          set_payload(entry_level_resource.type, record, entry, is_json)
          entries.push(entry)
        end
        next if entries.empty?
        log_name = "projects/#{@project_id}/logs/#{log_name(
          tag, group_level_resource)}"
        requests_to_send << {
          entries: entries,
          log_name: log_name,
          resource: group_level_resource,
          labels: group_level_common_labels
        }
      end
      if @split_logs_by_tag
        requests_to_send.each do |request|
          @write_request.call(request)
        end
      else
        combined_entries = []
        requests_to_send.each do |request|
          request[:entries].each do |entry|
            entry.log_name = request[:log_name]
          end
          combined_entries.concat(request[:entries])
        end
        @write_request.call(entries: combined_entries) unless
          combined_entries.empty?
      end
    end
def write ( chunk ) grouped_entries = group_log_entries_by_tag_and_local_resource_id ( chunk ) requests_to_send = [ ] grouped_entries . each do | ( tag , local_resource_id ) , arr | entries = [ ] group_level_resource , group_level_common_labels = determine_group_level_monitored_resource_and_labels ( tag , local_resource_id ) arr . each do | time , record | entry_level_resource , entry_level_common_labels = determine_entry_level_monitored_resource_and_labels ( group_level_resource , group_level_common_labels , record ) is_json = false if @detect_json preserved_keys = [ 'time' , 'severity' , @http_request_key , @insert_id_key , @operation_key , @source_location_key , @span_id_key , @trace_key ] record_json = nil if ( record . keys - preserved_keys ) . length == 1 %w( log message msg ) . each do | field | if record . key? ( field ) record_json = parse_json_or_nil ( record [ field ] ) end end end unless record_json . nil? preserved_keys . each do | key | record_json [ key ] ||= record [ key ] if record . key? ( key ) end record = record_json is_json = true end end ts_secs , ts_nanos = compute_timestamp ( entry_level_resource . type , record , time ) severity = compute_severity ( entry_level_resource . type , record , entry_level_common_labels ) entry = @construct_log_entry . call ( entry_level_common_labels , entry_level_resource , severity , ts_secs , ts_nanos ) trace = record . delete ( @trace_key ) entry . trace = compute_trace ( trace ) if trace span_id = record . delete ( @span_id_key ) entry . span_id = span_id if span_id insert_id = record . delete ( @insert_id_key ) entry . insert_id = insert_id if insert_id set_log_entry_fields ( record , entry ) set_payload ( entry_level_resource . type , record , entry , is_json ) entries . push ( entry ) end next if entries . empty? log_name = "projects/#{@project_id}/logs/#{log_name(
          tag, group_level_resource)}" requests_to_send << { entries : entries , log_name : log_name , resource : group_level_resource , labels : group_level_common_labels } end if @split_logs_by_tag requests_to_send . each do | request | @write_request . call ( request ) end else combined_entries = [ ] requests_to_send . each do | request | request [ :entries ] . each do | entry | entry . log_name = request [ :log_name ] end combined_entries . concat ( request [ :entries ] ) end @write_request . call ( entries : combined_entries ) unless combined_entries . empty? end end
Constants for service names, resource types and etc.
Constants for service names resource types and etc .
# Save the following fields if available, then clear them out to # allow for determining whether we should parse the log or message # field. # If the log is json, we want to export it as a structured log # unless there is additional metadata that would be lost. # Propagate these if necessary. Note that we don't want to # override these keys in the JSON we've just parsed. # Don't send an empty request if we rejected all the entries. # Combine all requests into one. The request level "log_name" will be # ported to the entry level. The request level "resource" and "labels" # are ignored as they should have been folded into the entry level # "resource" and "labels" already anyway. # Modify entries in-place as they are not needed later on.
chunk