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 =

begin
  # Extract plugin version from file path.
  match_data = __FILE__.match(
    %r{fluent-plugin-google-cloud-(?<version>[^/]*)/})
  if match_data
    match_data['version']
  else
    # Extract plugin version by finding the spec this file was loaded from.
    dependency = Gem::Dependency.new('fluent-plugin-google-cloud')
    all_specs, = Gem::SpecFetcher.fetcher.spec_for_dependency(dependency)
    matching_version, = all_specs.grep(
      proc { |spec,| __FILE__.include?(spec.full_gem_path) }) do |spec,|
        spec.version.to_s
      end
    matching_version
  end
end.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.

• #multi_workers_ready? ⇒ Boolean
• #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 472

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

  @failed_requests_count = nil
  @successful_requests_count = nil
  @dropped_entries_count = nil
  @ingested_entries_count = nil
  @retried_entries_count = nil

  @ok_code = nil
end

Instance Attribute Details

#common_labels ⇒ Object (readonly)

Returns the value of attribute common_labels.

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

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 466

def project_id
  @project_id
end

#resource ⇒ Object (readonly)

Returns the value of attribute resource.

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

def resource
  @resource
end

#vm_id ⇒ Object (readonly)

Returns the value of attribute vm_id.

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

def vm_id
  @vm_id
end

#zone ⇒ Object (readonly)

Returns the value of attribute zone.

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

def zone
  @zone
end

Instance Method Details

#configure(conf) ⇒ Object

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

def configure(conf)
  super

  # TODO(qingling128): Remove this warning after the support is added. Also
  # remove the comment in the description of this configuration.
  unless @logging_api_url == DEFAULT_LOGGING_API_URL || @use_grpc
    @log.warn 'Detected customized logging_api_url while use_grpc is not' \
              ' enabled. Customized logging_api_url for the non-gRPC path' \
              ' is not supported. The logging_api_url option will be' \
              ' ignored.'
  end

  # 1. If @metadata_agent_url is customized (aka not nil), use that.
  # 2. Otherwise check the presence of the environment variable
  #    STACKDRIVER_METADATA_AGENT_URL and use that if set.
  # 3. Fall back to the default if neither is set.
  if @enable_metadata_agent
    # Convert to string to capture empty string.
    @metadata_agent_url ||=
      if ENV[METADATA_AGENT_URL_ENV_VAR].to_s.empty?
        DEFAULT_METADATA_AGENT_URL
      else
        ENV[METADATA_AGENT_URL_ENV_VAR]
      end
  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

  # Treat an empty setting of the credentials file path environment variable
  # as unset. This way the googleauth lib could fetch the credentials
  # following the fallback path.
  ENV.delete(CREDENTIALS_PATH_ENV_VAR) if
    ENV[CREDENTIALS_PATH_ENV_VAR] == ''

  # 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

  # If monitoring is enabled, register metrics in the default registry
  # and store metric objects for future use.
  if @enable_monitoring
    unless Monitoring::MonitoringRegistryFactory.supports_monitoring_type(
      @monitoring_type)
      @log.warn "monitoring_type '#{@monitoring_type}' is unknown; "\
                'there will be no metrics'
    end
    @registry = Monitoring::MonitoringRegistryFactory
                .create(@monitoring_type, @project_id, @resource)
    # Export metrics every 60 seconds.
    timer_execute(:export_metrics, 60) { @registry.export }
    @successful_requests_count = @registry.counter(
      :stackdriver_successful_requests_count,
      [:grpc, :code],
      'A number of successful requests to the Stackdriver Logging API')
    @failed_requests_count = @registry.counter(
      :stackdriver_failed_requests_count,
      [:grpc, :code],
      '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,
      [:grpc, :code],
      'A number of log entries ingested by Stackdriver Logging')
    @dropped_entries_count = @registry.counter(
      :stackdriver_dropped_entries_count,
      [:grpc, :code],
      'A number of log entries dropped by the Stackdriver output plugin')
    @retried_entries_count = @registry.counter(
      :stackdriver_retried_entries_count,
      [:grpc, :code],
      '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 ? GRPC::Core::StatusCodes::OK : 200
  end

  # Set regexp that we should match tags against later on. Using a list
  # instead of a map to ensure order.
  @tag_regexp_list = []
  if @resource.type == GKE_CONSTANTS[:resource_type]
    @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

  if @use_grpc
    @construct_log_entry = method(:construct_log_entry_in_grpc_format)
    @write_request = method(:write_request_via_grpc)
  else
    @construct_log_entry = method(:construct_log_entry_in_rest_format)
    @write_request = method(:write_request_via_rest)
  end

  if [Platform::GCE, Platform::EC2].include?(@platform)
    # 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
end

#multi_workers_ready? ⇒ Boolean

Returns:

• (Boolean)

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

def multi_workers_ready?
  true
end

#shutdown ⇒ Object

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

def shutdown
  super
  # Export metrics on shutdown. This is a best-effort attempt, and it might
  # fail, for instance if there was a recent write to the same time series.
  @registry.export unless @registry.nil?
end

#start ⇒ Object

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

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

  if @statusz_port > 0
    @log.info "Starting statusz server on port #{@statusz_port}"
    server_create(:out_google_cloud_statusz,
                  @statusz_port,
                  bind: '127.0.0.1') do |data, conn|
      if data.split(' ')[1] == '/statusz'
        write_html_response(data, conn, 200, Statusz.response(self))
      else
        write_html_response(data, conn, 404, "Not found\n")
      end
    end
  end
end

#write(chunk) ⇒ Object

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

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.
        # This list should be in sync with
        # https://cloud.google.com/logging/docs/agent/configuration#special-fields.
        preserved_keys = [
          'time',
          'timeNanos',
          'timestamp',
          'timestampNanos',
          'timestampSeconds',
          'severity',
          @http_request_key,
          @insert_id_key,
          @labels_key,
          @operation_key,
          @source_location_key,
          @span_id_key,
          @trace_key,
          @trace_sampled_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) && !record_json.key?(key)
          end

          record = record_json
          is_json = true
        end
      end

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

      dynamic_labels_from_payload = parse_labels(record)

      entry_level_common_labels = entry_level_common_labels.merge!(
        dynamic_labels_from_payload) if dynamic_labels_from_payload

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

      insert_id = record.delete(@insert_id_key)
      entry.insert_id = insert_id if insert_id
      span_id = record.delete(@span_id_key)
      entry.span_id = span_id if span_id
      trace = record.delete(@trace_key)
      entry.trace = compute_trace(trace) if trace
      trace_sampled = record.delete(@trace_sampled_key)
      entry.trace_sampled = parse_bool(trace_sampled) unless
        trace_sampled.nil?

      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