Class: Fluent::GoogleCloudOutput

Inherits:

BufferedOutput

• Object
• BufferedOutput
• Fluent::GoogleCloudOutput

Includes:

selfself::Constants

Defined in:

lib/fluent/plugin/out_google_cloud.rb

Overview

fluentd output plugin for the Stackdriver Logging API

Defined Under Namespace

Modules: Constants, CredentialsInfo, Platform

Constant Summary

PLUGIN_NAME =

'Fluentd Google Cloud Logging plugin'

PLUGIN_VERSION =

'0.6.6'

LOGGING_SCOPE =

Name of the the Google cloud logging write scope.

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

METADATA_SERVICE_ADDR =

Address of the metadata service.

'169.254.169.254'

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 305

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 303

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 299

def project_id
  @project_id
end

#resource ⇒ Object (readonly)

Returns the value of attribute resource.

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

def resource
  @resource
end

#vm_id ⇒ Object (readonly)

Returns the value of attribute vm_id.

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

def vm_id
  @vm_id
end

#zone ⇒ Object (readonly)

Returns the value of attribute zone.

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

def zone
  @zone
end

Instance Method Details

#configure(conf) ⇒ Object

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

def configure(conf)
  super

  # 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')
  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?

    fail 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 by
  # making some requests to metadata server.
  #
  # Note: Once we support metadata injection at Logging API side, we might
  # no longer need to require all these metadata in logging agent. But for
  # now, they are still required.
  #
  # TODO(qingling128): After Metadata Agent support is added, try extracting
  # these info from responses from Metadata Agent first.
  set_required_metadata_variables

  # Retrieve monitored resource.
  #
  # TODO(qingling128): After Metadata Agent support is added, try retrieving
  # the monitored resource from Metadata Agent first.
  @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 == CONTAINER_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 << [
      CONTAINER_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 413

def shutdown
  super
end

#start ⇒ Object

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

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 417

def write(chunk)
  # Group the entries since we have to make one call per tag.
  grouped_entries = {}
  chunk.msgpack_each do |tag, *arr|
    sanitized_tag = sanitize_tag(tag)
    if sanitized_tag.nil?
      @log.warn "Dropping log entries with invalid tag: '#{tag}'. " \
                'A tag should be a string with utf8 characters.'
      next
    end
    grouped_entries[sanitized_tag] ||= []
    grouped_entries[sanitized_tag].push(arr)
  end

  grouped_entries.each do |tag, arr|
    entries = []
    group_resource, group_common_labels =
      determine_group_level_monitored_resource_and_labels(tag)

    arr.each do |time, record|
      next unless record.is_a?(Hash)

      extracted_resource_labels, extracted_common_labels = \
        determine_entry_level_labels(group_resource, record)
      entry_resource = group_resource.dup
      entry_resource.labels.merge!(extracted_resource_labels)
      entry_common_labels = \
        group_common_labels.merge(extracted_common_labels)

      if entry_resource.type == CONTAINER_CONSTANTS[:resource_type]
        # Save the timestamp if available, then clear it out to allow for
        # determining whether we should parse the log or message field.
        timestamp = record.key?('time') ? record['time'] : nil
        record.delete('time')
        # If the log is json, we want to export it as a structured log
        # unless there is additional metadata that would be lost.
        is_json = false
        if record.length == 1 && record.key?('log')
          record_json = parse_json_or_nil(record['log'])
        end
        if record.length == 1 && record.key?('message')
          record_json = parse_json_or_nil(record['message'])
        end
        unless record_json.nil?
          record = record_json
          is_json = true
        end
        # Restore timestamp if necessary.
        unless record.key?('time') || timestamp.nil?
          record['time'] = timestamp
        end
      end

      ts_secs, ts_nanos = compute_timestamp(
        entry_resource.type, record, time)
      severity = compute_severity(
        entry_resource.type, record, entry_common_labels)

      ts_secs = begin
                  Integer ts_secs
                rescue ArgumentError, TypeError
                  ts_secs
                end
      ts_nanos = begin
                   Integer ts_nanos
                 rescue ArgumentError, TypeError
                   ts_nanos
                 end
      if @use_grpc
        entry = Google::Logging::V2::LogEntry.new(
          labels: entry_common_labels,
          resource: Google::Api::MonitoredResource.new(
            type: entry_resource.type,
            labels: entry_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_resource.labels = nil if entry_resource.labels.empty?
        entry = Google::Apis::LoggingV2beta1::LogEntry.new(
          labels: entry_common_labels,
          resource: entry_resource,
          severity: severity,
          timestamp: {
            seconds: ts_secs,
            nanos: ts_nanos
          }
        )
      end

      # Get fully-qualified trace id for LogEntry "trace" field per config.
      fq_trace_id = record.delete(@trace_key)
      entry.trace = fq_trace_id if fq_trace_id

      set_log_entry_fields(record, entry)

      if @use_grpc
        set_payload_grpc(entry_resource.type, record, entry, is_json)
      else
        set_payload(entry_resource.type, record, entry, is_json)
      end

      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_resource)}"

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

        write_request = Google::Logging::V2::WriteLogEntriesRequest.new(
          log_name: log_name,
          resource: Google::Api::MonitoredResource.new(
            type: group_resource.type,
            labels: group_resource.labels.to_h
          ),
          labels: labels_utf8_pairs.to_h,
          entries: entries
        )

        client.write_log_entries(write_request)
        increment_successful_requests_count
        increment_ingested_entries_count(entries.length)

        # 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 GRPC::Cancelled => error
        increment_failed_requests_count(GRPC::Core::StatusCodes::CANCELLED)
        # RPC cancelled, so retry via re-raising the error.
        raise error

      rescue GRPC::BadStatus => error
        increment_failed_requests_count(error.code)
        case error.code
        when GRPC::Core::StatusCodes::CANCELLED,
             GRPC::Core::StatusCodes::UNAVAILABLE,
             GRPC::Core::StatusCodes::DEADLINE_EXCEEDED,
             GRPC::Core::StatusCodes::INTERNAL,
             GRPC::Core::StatusCodes::UNKNOWN
          # TODO
          # Server error, so retry via re-raising the error.
          raise error
        when GRPC::Core::StatusCodes::UNIMPLEMENTED,
             GRPC::Core::StatusCodes::RESOURCE_EXHAUSTED
          # Most client errors indicate a problem with the request itself
          # and should not be retried.
          dropped = entries.length
          increment_dropped_entries_count(dropped)
          @log.warn "Dropping #{dropped} log message(s)",
                    error: error.to_s, error_code: error.code.to_s
        when GRPC::Core::StatusCodes::UNAUTHENTICATED
          # Authorization error.
          # These are usually solved via a `gcloud auth` call, or by
          # modifying the permissions on the Google Cloud project.
          dropped = entries.length
          increment_dropped_entries_count(dropped)
          @log.warn "Dropping #{dropped} log message(s)",
                    error: error.to_s, error_code: error.code.to_s
        else
          # Assume this is a problem with the request itself
          # and don't retry.
          dropped = entries.length
          increment_dropped_entries_count(dropped)
          @log.error "Unknown response code #{error.code} from the "\
                     "server, dropping #{dropped} log message(s)",
                     error: error.to_s, error_code: error.code.to_s
        end
      end
    else
      begin
        write_request = \
          Google::Apis::LoggingV2beta1::WriteLogEntriesRequest.new(
            log_name: log_name,
            resource: group_resource,
            labels: group_common_labels,
            entries: entries)

        # TODO: RequestOptions
        begin
          client.write_entry_log_entries(write_request)
        rescue Google::Apis::Error => error
          increment_failed_requests_count(error.status_code)
          raise error
        end
        increment_successful_requests_count
        increment_ingested_entries_count(entries.length)

        # 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
        # Server error, so retry via re-raising the error.
        raise error

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

      rescue Google::Apis::ClientError => error
        # Most ClientErrors indicate a problem with the request itself and
        # should not be retried.
        dropped = entries.length
        increment_dropped_entries_count(dropped)
        @log.warn "Dropping #{dropped} log message(s)",
                  error_class: error.class.to_s, error: error.to_s
      end
    end
  end
end