Class: Fluent::GoogleCloudOutput

Inherits:

BufferedOutput

• Object
• BufferedOutput
• Fluent::GoogleCloudOutput

Includes:

selfself::ConfigConstants, selfself::InternalConstants, selfself::ServiceConstants

Defined in:

lib/fluent/plugin/out_google_cloud.rb

Overview

fluentd output plugin for the Stackdriver Logging API

Defined Under Namespace

Modules: ConfigConstants, CredentialsInfo, InternalConstants, Platform, ServiceConstants

Constant Summary

PLUGIN_NAME =

'Fluentd Google Cloud Logging plugin'.freeze

PLUGIN_VERSION =

'0.6.12'.freeze

LOGGING_SCOPE =

Name of the the Google cloud logging write scope.

'https://www.googleapis.com/auth/logging.write'.freeze

METADATA_SERVICE_ADDR =

Address of the metadata service.

'169.254.169.254'.freeze

Instance Attribute Summary

• #common_labels ⇒ Object readonly

  Returns the value of attribute common_labels.

• #project_id ⇒ Object readonly

  Expose attr_readers to make testing of metadata more direct than only testing it indirectly through metadata sent with logs.

• #resource ⇒ Object readonly

  Returns the value of attribute resource.

• #vm_id ⇒ Object readonly

  Returns the value of attribute vm_id.

• #zone ⇒ Object readonly

  Returns the value of attribute zone.

Instance Method Summary

• #configure(conf) ⇒ Object
• #initialize ⇒ GoogleCloudOutput constructor

  A new instance of GoogleCloudOutput.

• #shutdown ⇒ Object
• #start ⇒ Object
• #write(chunk) ⇒ Object

Constructor Details

#initialize ⇒ GoogleCloudOutput

Returns a new instance of GoogleCloudOutput.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 356

def initialize
  super
  # use the global logger
  @log = $log # rubocop:disable Style/GlobalVars
end

Instance Attribute Details

#common_labels ⇒ Object (readonly)

Returns the value of attribute common_labels.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 354

def common_labels
  @common_labels
end

#project_id ⇒ Object (readonly)

Expose attr_readers to make testing of metadata more direct than only testing it indirectly through metadata sent with logs.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 350

def project_id
  @project_id
end

#resource ⇒ Object (readonly)

Returns the value of attribute resource.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 353

def resource
  @resource
end

#vm_id ⇒ Object (readonly)

Returns the value of attribute vm_id.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 352

def vm_id
  @vm_id
end

#zone ⇒ Object (readonly)

Returns the value of attribute zone.

# File 'lib/fluent/plugin/out_google_cloud.rb', line 351

def zone
  @zone
end

Instance Method Details

#configure(conf) ⇒ Object

# File 'lib/fluent/plugin/out_google_cloud.rb', line 362

def configure(conf)
  super

  # TODO(qingling128): Remove this warning after the support is added. Also
  # remove the comment in the description of this configuration.
  if @partial_success && @use_grpc
    @log.warn 'Detected partial_success enabled while use_grpc is also' \
              ' enabled. The support for partial success in the gRPC path' \
              ' is to be added in the near future. For now the ' \
              ' partial_success flag will be ignored.'
  end

  # If monitoring is enabled, register metrics in the default registry
  # and store metric objects for future use.
  if @enable_monitoring
    registry = Monitoring::MonitoringRegistryFactory.create @monitoring_type
    @successful_requests_count = registry.counter(
      :stackdriver_successful_requests_count,
      'A number of successful requests to the Stackdriver Logging API')
    @failed_requests_count = registry.counter(
      :stackdriver_failed_requests_count,
      'A number of failed requests to the Stackdriver Logging API,'\
        ' broken down by the error code')
    @ingested_entries_count = registry.counter(
      :stackdriver_ingested_entries_count,
      'A number of log entries ingested by Stackdriver Logging')
    @dropped_entries_count = registry.counter(
      :stackdriver_dropped_entries_count,
      'A number of log entries dropped by the Stackdriver output plugin')
    @retried_entries_count = registry.counter(
      :stackdriver_retried_entries_count,
      'The number of log entries that failed to be ingested by the'\
        ' Stackdriver output plugin due to a transient error and were'\
        ' retried')
    @ok_code = @use_grpc ? 0 : 200
  end

  # Alert on old authentication configuration.
  unless @auth_method.nil? && @private_key_email.nil? &&
         @private_key_path.nil? && @private_key_passphrase.nil?
    extra = []
    extra << 'auth_method' unless @auth_method.nil?
    extra << 'private_key_email' unless @private_key_email.nil?
    extra << 'private_key_path' unless @private_key_path.nil?
    extra << 'private_key_passphrase' unless @private_key_passphrase.nil?

    raise Fluent::ConfigError,
          "#{PLUGIN_NAME} no longer supports auth_method.\n" \
          "Please remove configuration parameters: #{extra.join(' ')}"
  end

  set_regexp_patterns

  @platform = detect_platform

  # Set required variables: @project_id, @vm_id, @vm_name and @zone.
  set_required_metadata_variables

  # Retrieve monitored resource.
  # Fail over to retrieve monitored resource via the legacy path if we fail
  # to get it from Metadata Agent.
  @resource ||= determine_agent_level_monitored_resource_via_legacy

  # Set regexp that we should match tags against later on. Using a list
  # instead of a map to ensure order. For example, tags will be matched
  # against Cloud Functions first, then GKE.
  @tag_regexp_list = []
  if @resource.type == GKE_CONSTANTS[:resource_type]
    # We only support Cloud Functions logs for GKE right now.
    if fetch_gce_metadata('instance/attributes/'
                         ).split.include?('gcf_region')
      # Fetch this info and store it to avoid recurring
      # metadata server calls.
      @gcf_region = fetch_gce_metadata('instance/attributes/gcf_region')
      @tag_regexp_list << [
        CLOUDFUNCTIONS_CONSTANTS[:resource_type],
        @compiled_cloudfunctions_tag_regexp
      ]
    end
    @tag_regexp_list << [
      GKE_CONSTANTS[:resource_type], @compiled_kubernetes_tag_regexp
    ]
  end

  # Determine the common labels that should be added to all log entries
  # processed by this logging agent.
  @common_labels = determine_agent_level_common_labels

  # The resource and labels are now set up; ensure they can't be modified
  # without first duping them.
  @resource.freeze
  @resource.labels.freeze
  @common_labels.freeze

  # Log an informational message containing the Logs viewer URL
  @log.info 'Logs viewer address: https://console.cloud.google.com/logs/',
            "viewer?project=#{@project_id}&resource=#{@resource.type}/",
            "instance_id/#{@vm_id}"
end

#shutdown ⇒ Object

# File 'lib/fluent/plugin/out_google_cloud.rb', line 469

def shutdown
  super
end

#start ⇒ Object

# File 'lib/fluent/plugin/out_google_cloud.rb', line 462

def start
  super
  init_api_client
  @successful_call = false
  @timenanos_warning = false
end

#write(chunk) ⇒ Object

# File 'lib/fluent/plugin/out_google_cloud.rb', line 473

def write(chunk)
  grouped_entries = group_log_entries_by_tag_and_local_resource_id(chunk)

  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 timestamp and severity if available, then clear it out to
        # allow for determining whether we should parse the log or message
        # field.
        timestamp = record.delete('time')
        severity = record.delete('severity')

        # 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.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?
          record = record_json
          is_json = true
        end
        # Restore timestamp and severity if necessary. Note that we don't
        # want to override these keys in the JSON we've just parsed.
        record['time'] ||= timestamp if timestamp
        record['severity'] ||= severity if severity
      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)

      if @use_grpc
        entry = Google::Logging::V2::LogEntry.new(
          labels: entry_level_common_labels,
          resource: Google::Api::MonitoredResource.new(
            type: entry_level_resource.type,
            labels: entry_level_resource.labels.to_h
          ),
          severity: grpc_severity(severity)
        )
        # If "seconds" is null or not an integer, we will omit the timestamp
        # field and defer the decision on how to handle it to the downstream
        # Logging API. If "nanos" is null or not an integer, it will be set
        # to 0.
        if ts_secs.is_a?(Integer)
          ts_nanos = 0 unless ts_nanos.is_a?(Integer)
          entry.timestamp = Google::Protobuf::Timestamp.new(
            seconds: ts_secs,
            nanos: ts_nanos
          )
        end
      else
        # Remove the labels if we didn't populate them with anything.
        entry_level_resource.labels = nil if
          entry_level_resource.labels.empty?
        entry = Google::Apis::LoggingV2::LogEntry.new(
          labels: entry_level_common_labels,
          resource: entry_level_resource,
          severity: severity,
          timestamp: {
            seconds: ts_secs,
            nanos: ts_nanos
          }
        )
      end

      # Get fully-qualified trace id for LogEntry "trace" field.
      fq_trace_id = record.delete(@trace_key)
      entry.trace = fq_trace_id if fq_trace_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)}"

    # Does the actual write to the cloud logging api.
    client = api_client
    if @use_grpc
      begin
        labels_utf8_pairs = group_level_common_labels.map do |k, v|
          [k.encode('utf-8'), convert_to_utf8(v)]
        end

        entries_count = entries.length
        client.write_log_entries(
          entries,
          log_name: log_name,
          resource: Google::Api::MonitoredResource.new(
            type: group_level_resource.type,
            labels: group_level_resource.labels.to_h
          ),
          labels: labels_utf8_pairs.to_h
        )
        increment_successful_requests_count
        increment_ingested_entries_count(entries_count)

        # Let the user explicitly know when the first call succeeded, to aid
        # with verification and troubleshooting.
        unless @successful_call
          @successful_call = true
          @log.info 'Successfully sent gRPC to Stackdriver Logging API.'
        end

      rescue Google::Gax::GaxError => gax_error
        # GRPC::BadStatus is wrapped in error.cause.
        error = gax_error.cause

        # See the mapping between HTTP status and gRPC status code at:
        # https://github.com/grpc/grpc/blob/master/src/core/lib/transport/status_conversion.cc
        case error
        # Server error, so retry via re-raising the error.
        when \
            # HTTP status 500 (Internal Server Error).
            GRPC::Internal,
            # HTTP status 501 (Not Implemented).
            GRPC::Unimplemented,
            # HTTP status 503 (Service Unavailable).
            GRPC::Unavailable,
            # HTTP status 504 (Gateway Timeout).
            GRPC::DeadlineExceeded
          increment_retried_entries_count(entries_count, error.code)
          @log.debug "Retrying #{entries_count} log message(s) later.",
                     error: error.to_s, error_code: error.code.to_s
          raise error

        # Most client errors indicate a problem with the request itself and
        # should not be retried.
        when \
            # HTTP status 400 (Bad Request).
            GRPC::InvalidArgument,
            # HTTP status 401 (Unauthorized).
            # These are usually solved via a `gcloud auth` call, or by
            # modifying the permissions on the Google Cloud project.
            GRPC::Unauthenticated,
            # HTTP status 403 (Forbidden).
            GRPC::PermissionDenied,
            # HTTP status 404 (Not Found).
            GRPC::NotFound,
            # HTTP status 409 (Conflict).
            GRPC::Aborted,
            # HTTP status 412 (Precondition Failed).
            GRPC::FailedPrecondition,
            # HTTP status 429 (Too Many Requests).
            GRPC::ResourceExhausted,
            # HTTP status 499 (Client Closed Request).
            GRPC::Cancelled,
            # the remaining http codes in both 4xx and 5xx category.
            # It's debatable whether to retry or drop these log entries.
            # This decision is made to avoid retrying forever due to client
            # errors.
            GRPC::Unknown
          increment_failed_requests_count(error.code)
          increment_dropped_entries_count(entries_count, error.code)
          @log.warn "Dropping #{entries_count} log message(s)",
                    error: error.to_s, error_code: error.code.to_s

        else
          # Assume it is a problem with the request itself and don't retry.
          increment_failed_requests_count(error.code)
          increment_dropped_entries_count(entries_count, error.code)
          @log.error "Unknown response code #{error.code} from the "\
                     "server, dropping #{entries_count} log message(s)",
                     error: error.to_s, error_code: error.code.to_s
        end
      end
    else
      begin
        write_request =
          Google::Apis::LoggingV2::WriteLogEntriesRequest.new(
            log_name: log_name,
            resource: group_level_resource,
            labels: group_level_common_labels,
            partial_success: @partial_success,
            entries: entries)
        entries_count = entries.length
        client.write_entry_log_entries(
          write_request,
          options: { api_format_version: '2' }
        )

        increment_successful_requests_count
        increment_ingested_entries_count(entries_count)

        # Let the user explicitly know when the first call succeeded, to aid
        # with verification and troubleshooting.
        unless @successful_call
          @successful_call = true
          @log.info 'Successfully sent to Stackdriver Logging API.'
        end

      rescue Google::Apis::ServerError => error
        # 5xx server errors. Retry via re-raising the error.
        increment_retried_entries_count(entries_count, error.status_code)
        @log.debug "Retrying #{entries_count} log message(s) later.",
                   error: error.to_s, error_code: error.status_code.to_s
        raise error

      rescue Google::Apis::AuthorizationError => error
        # 401 authorization error.
        # These are usually solved via a `gcloud auth` call, or by modifying
        # the permissions on the Google Cloud project.
        increment_failed_requests_count(error.status_code)
        increment_dropped_entries_count(entries_count, error.status_code)
        @log.warn "Dropping #{entries_count} log message(s)",
                  error_class: error.class.to_s, error: error.to_s

      rescue Google::Apis::ClientError => error
        # 4xx client errors. Most client errors indicate a problem with the
        # request itself and should not be retried.
        error_details_map = construct_error_details_map(error)
        if error_details_map.empty?
          increment_failed_requests_count(error.status_code)
          increment_dropped_entries_count(entries_count, error.status_code)
          @log.warn "Dropping #{entries_count} log message(s)",
                    error_class: error.class.to_s, error: error.to_s
        else
          error_details_map.each do |(error_code, error_message), indexes|
            partial_error_count = indexes.length
            increment_dropped_entries_count(partial_error_count, error_code)
            entries_count -= partial_error_count
            @log.warn "Dropping #{partial_error_count} log message(s)",
                      error_code: "google.rpc.Code[#{error_code}]",
                      error: error_message
          end
          # Consider partially successful requests successful.
          increment_successful_requests_count
          increment_ingested_entries_count(entries_count)
        end
      end
    end
  end
end