Class: Fluent::GoogleCloudOutput

Inherits:

BufferedOutput

• Object
• BufferedOutput
• Fluent::GoogleCloudOutput

Defined in:

lib/fluent/plugin/out_google_cloud.rb

Overview

fluentd output plugin for the Stackdriver Logging API

Defined Under Namespace

Modules: CredentialsInfo, Platform

Constant Summary

PLUGIN_NAME =

'Fluentd Google Cloud Logging plugin'

PLUGIN_VERSION =

'0.5.5'

APPENGINE_SERVICE =

Constants for service names.

'appengine.googleapis.com'

CLOUDFUNCTIONS_SERVICE =

'cloudfunctions.googleapis.com'

COMPUTE_SERVICE =

'compute.googleapis.com'

CONTAINER_SERVICE =

'container.googleapis.com'

EC2_SERVICE =

'ec2.amazonaws.com'

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.

• #gae_backend_name ⇒ Object readonly

  Returns the value of attribute gae_backend_name.

• #gae_backend_version ⇒ Object readonly

  Returns the value of attribute gae_backend_version.

• #project_id ⇒ Object readonly

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

• #running_on_managed_vm ⇒ Object readonly

  Returns the value of attribute running_on_managed_vm.

• #service_name ⇒ Object readonly

  Returns the value of attribute service_name.

• #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
• #format(tag, time, record) ⇒ Object
• #initialize ⇒ GoogleCloudOutput constructor

  A new instance of GoogleCloudOutput.

• #sanitize_tag(tag) ⇒ Object

  Given a tag, returns the corresponding valid tag if possible, or nil if the tag should be rejected.

• #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 173

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 171

def common_labels
  @common_labels
end

#gae_backend_name ⇒ Object (readonly)

Returns the value of attribute gae_backend_name.

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

def gae_backend_name
  @gae_backend_name
end

#gae_backend_version ⇒ Object (readonly)

Returns the value of attribute gae_backend_version.

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

def gae_backend_version
  @gae_backend_version
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 164

def project_id
  @project_id
end

#running_on_managed_vm ⇒ Object (readonly)

Returns the value of attribute running_on_managed_vm.

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

def running_on_managed_vm
  @running_on_managed_vm
end

#service_name ⇒ Object (readonly)

Returns the value of attribute service_name.

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

def service_name
  @service_name
end

#vm_id ⇒ Object (readonly)

Returns the value of attribute vm_id.

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

def vm_id
  @vm_id
end

#zone ⇒ Object (readonly)

Returns the value of attribute zone.

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

def zone
  @zone
end

Instance Method Details

#configure(conf) ⇒ Object

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

def configure(conf)
  super

  # 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

  # TODO: Send instance tags as labels as well?
  @common_labels = {}
  @common_labels.merge!(@labels) if @labels

  @compiled_kubernetes_tag_regexp = nil
  if @kubernetes_tag_regexp
    @compiled_kubernetes_tag_regexp = Regexp.new(@kubernetes_tag_regexp)
  end

  @cloudfunctions_tag_regexp =
    /\.(?<encoded_function_name>.+)\.\d+-[^-]+_default_worker$/
  @cloudfunctions_log_regexp = /^
    (?:\[(?<severity>.)\])?
    \[(?<timestamp>.{24})\]
    (?:\[(?<execution_id>[^\]]+)\])?
    [ ](?<text>.*)$/x

  # set attributes from metadata (unless overriden by static config)
  @vm_name = Socket.gethostname if @vm_name.nil?
  @platform = detect_platform
  case @platform
  when Platform::GCE
    if @project_id.nil?
      @project_id = fetch_gce_metadata('project/project-id')
    end
    if @zone.nil?
      # this returns "projects/<number>/zones/<zone>"; we only want
      # the part after the final slash.
      fully_qualified_zone = fetch_gce_metadata('instance/zone')
      @zone = fully_qualified_zone.rpartition('/')[2]
    end
    @vm_id = fetch_gce_metadata('instance/id') if @vm_id.nil?
  when Platform::EC2
    metadata = fetch_ec2_metadata
    if @zone.nil? && metadata.key?('availabilityZone')
      @zone = 'aws:' + metadata['availabilityZone']
    end
    if @vm_id.nil? && metadata.key?('instanceId')
      @vm_id = metadata['instanceId']
    end
    if metadata.key?('accountId')
      common_labels["#{EC2_SERVICE}/account_id"] = metadata['accountId']
    end
  when Platform::OTHER
    # do nothing
  else
    fail Fluent::ConfigError, 'Unknown platform ' + @platform
  end

  # If we still don't have a project ID, try to obtain it from the
  # credentials.
  if @project_id.nil?
    @project_id = CredentialsInfo.project_id
    @log.info 'Set Project ID from credentials: ', @project_id unless
      @project_id.nil?
  end

  # all metadata parameters must now be set
  unless @project_id && @zone && @vm_id
    missing = []
    missing << 'project_id' unless @project_id
    missing << 'zone' unless @zone
    missing << 'vm_id' unless @vm_id
    fail Fluent::ConfigError, 'Unable to obtain metadata parameters: ' +
      missing.join(' ')
  end

  # Default this to false; it is only overwritten if we detect Managed VM.
  @running_on_managed_vm = false

  # Default this to false; it is only overwritten if we detect Cloud
  # Functions.
  @running_cloudfunctions = false

  # Set labels, etc. based on the config
  case @platform
  when Platform::GCE
    @service_name = COMPUTE_SERVICE
    if @subservice_name
      @service_name = @subservice_name
    elsif @detect_subservice
      # Check for specialized GCE environments.
      # TODO: Add config options for these to allow for running outside GCE?
      attributes = fetch_gce_metadata('instance/attributes/').split
      # Do nothing, just don't populate other service's labels.
      if attributes.include?('gae_backend_name') &&
         attributes.include?('gae_backend_version')
        # Managed VM
        @running_on_managed_vm = true
        @gae_backend_name =
            fetch_gce_metadata('instance/attributes/gae_backend_name')
        @gae_backend_version =
            fetch_gce_metadata('instance/attributes/gae_backend_version')
        @service_name = APPENGINE_SERVICE
        common_labels["#{APPENGINE_SERVICE}/module_id"] = @gae_backend_name
        common_labels["#{APPENGINE_SERVICE}/version_id"] =
          @gae_backend_version
      elsif attributes.include?('kube-env')
        # Kubernetes/Container Engine
        @service_name = CONTAINER_SERVICE
        common_labels["#{CONTAINER_SERVICE}/instance_id"] = @vm_id
        @raw_kube_env = fetch_gce_metadata('instance/attributes/kube-env')
        @kube_env = YAML.load(@raw_kube_env)
        common_labels["#{CONTAINER_SERVICE}/cluster_name"] =
          cluster_name_from_kube_env(@kube_env)
        detect_cloudfunctions(attributes)
      end
    end
    common_labels["#{COMPUTE_SERVICE}/resource_type"] = 'instance'
    common_labels["#{COMPUTE_SERVICE}/resource_id"] = @vm_id
    common_labels["#{COMPUTE_SERVICE}/resource_name"] = @vm_name
  when Platform::EC2
    @service_name = EC2_SERVICE
    common_labels["#{EC2_SERVICE}/resource_type"] = 'instance'
    common_labels["#{EC2_SERVICE}/resource_id"] = @vm_id
    common_labels["#{EC2_SERVICE}/resource_name"] = @vm_name
  when Platform::OTHER
    # Use COMPUTE_SERVICE as the default environment.
    @service_name = COMPUTE_SERVICE
    common_labels["#{COMPUTE_SERVICE}/resource_type"] = 'instance'
    common_labels["#{COMPUTE_SERVICE}/resource_id"] = @vm_id
    common_labels["#{COMPUTE_SERVICE}/resource_name"] = @vm_name
  end

  # Log an informational message containing the Logs viewer URL
  @log.info 'Logs viewer address: ',
            'https://console.developers.google.com/project/', @project_id,
            '/logs?service=', @service_name, '&key1=instance&key2=', @vm_id
end

#format(tag, time, record) ⇒ Object

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

def format(tag, time, record)
  [tag, time, record].to_msgpack
end

#sanitize_tag(tag) ⇒ Object

Given a tag, returns the corresponding valid tag if possible, or nil if the tag should be rejected. If ‘require_valid_tags’ is false, non-string tags are converted to strings, and invalid characters are sanitized; otherwise such tags are rejected.

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

def sanitize_tag(tag)
  if @require_valid_tags &&
     (!tag.is_a?(String) || tag == '' || convert_to_utf8(tag) != tag)
    return nil
  end
  tag = convert_to_utf8(tag.to_s)
  tag = '_' if tag == ''
  tag
end

#shutdown ⇒ Object

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

def shutdown
  super
end

#start ⇒ Object

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

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 356

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 = []
    labels = @common_labels.clone

    if @running_cloudfunctions
      # If the current group of entries is coming from a Cloud Functions
      # function, the function name can be extracted from the tag.
      match_data = @cloudfunctions_tag_regexp.match(tag)
      if match_data
        # Service name is set to Cloud Functions only for logs actually
        # coming from a function.
        @service_name = CLOUDFUNCTIONS_SERVICE
        labels["#{CLOUDFUNCTIONS_SERVICE}/region"] = @gcf_region
        labels["#{CLOUDFUNCTIONS_SERVICE}/function_name"] =
          decode_cloudfunctions_function_name(
            match_data['encoded_function_name'])
      else
        # Other logs are considered as coming from the Container Engine
        # service.
        @service_name = CONTAINER_SERVICE
      end
    end
    if @service_name == CONTAINER_SERVICE && @compiled_kubernetes_tag_regexp
      # Container logs in Kubernetes are tagged based on where they came
      # from, so we can extract useful metadata from the tag.
      # Do this here to avoid having to repeat it for each record.
      match_data = @compiled_kubernetes_tag_regexp.match(tag)
      if match_data
        %w(namespace_name pod_name container_name).each do |field|
          labels["#{CONTAINER_SERVICE}/#{field}"] = match_data[field]
        end
      end
    end

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

      if @use_grpc
        entry = Google::Logging::V1::LogEntry.new(
          metadata: Google::Logging::V1::LogEntryMetadata.new(
            service_name: convert_to_utf8(@service_name),
            project_id: convert_to_utf8(@project_id),
            zone: convert_to_utf8(@zone),
            labels: {}
          ))
      else
        entry = Google::Apis::LoggingV1beta3::LogEntry.new(
          metadata: Google::Apis::LoggingV1beta3::LogEntryMetadata.new(
            service_name: @service_name,
            project_id: @project_id,
            zone: @zone,
            labels: {}
          ))
      end

      if @service_name == CLOUDFUNCTIONS_SERVICE && record.key?('log')
        @cloudfunctions_log_match =
          @cloudfunctions_log_regexp.match(record['log'])
      end
      if @service_name == CONTAINER_SERVICE
        # Move the stdout/stderr annotation from the record into a label.
        field_to_label(record, 'stream', entry.metadata.labels,
                       "#{CONTAINER_SERVICE}/stream")
        # If the record has been annotated by the kubernetes_metadata_filter
        # plugin, then use that metadata. Otherwise, rely on commonLabels
        # populated at the grouped_entries level from the group's tag.
        if record.key?('kubernetes')
          handle_container_metadata(record, entry)
        end

        # 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(record, time)
      if @use_grpc
        # 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.metadata.timestamp = Google::Protobuf::Timestamp.new(
            seconds: ts_secs,
            nanos: ts_nanos
          )
        end

        entry.metadata.severity =
          grpc_severity(compute_severity(record, entry))

        set_http_request_grpc(record, entry) # FIXME
      else
        entry.metadata.timestamp = {
          seconds: ts_secs,
          nanos: ts_nanos
        }

        entry.metadata.severity =
          compute_severity(record, entry)

        set_http_request(record, entry)
      end

      # If a field is present in the label_map, send its value as a label
      # (mapping the field name to label name as specified in the config)
      # and do not send that field as part of the payload.
      if @label_map
        @label_map.each do |field, label|
          field_to_label(record, field, entry.metadata.labels, label)
        end
      end

      if @service_name == CLOUDFUNCTIONS_SERVICE &&
         @cloudfunctions_log_match &&
         @cloudfunctions_log_match['execution_id']
        entry.metadata.labels['execution_id'] =
          @cloudfunctions_log_match['execution_id']
      end

      if @use_grpc
        set_payload_grpc(record, entry, is_json)
      else
        set_payload(record, entry, is_json)
        entry.metadata.labels = nil if entry.metadata.labels.empty?
      end

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

    log_name = log_name(tag, labels)

    if @use_grpc
      begin
        # Does the actual write to the cloud logging api.

        client = api_client

        labels_utf8_pairs = labels.map do |k, v|
          [k.encode('utf-8'), convert_to_utf8(v)]
        end

        write_request = Google::Logging::V1::WriteLogEntriesRequest.new(
          log_name: "projects/#{@project_id}/logs/#{log_name}",
          common_labels: Hash[labels_utf8_pairs],
          entries: entries
        )

        client.write_log_entries(write_request)

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

      rescue GRPC::BadStatus => error
        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
          @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
          @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
          @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
        # Does the actual write to the cloud logging api.

        client = api_client

        # The URI of the write is constructed by the Google::Api request;
        # it is equivalent to this URL:
        # 'https://logging.googleapis.com/v1beta3/projects/'          #   "#{@project_id}/logs/#{log_name}/entries:write"
        write_request = \
          Google::Apis::LoggingV1beta3::WriteLogEntriesRequest.new(
            common_labels: labels,
            entries: entries)

        # TODO: RequestOptions
        client.write_log_entries(@project_id, log_name, write_request)

        # 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
        @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
        @log.warn "Dropping #{dropped} log message(s)",
                  error_class: error.class.to_s, error: error.to_s
      end
    end
  end
end