Class: Fluent::BigQueryOutput

Inherits:

BufferedOutput

• Object
• BufferedOutput
• Fluent::BigQueryOutput

Defined in:

lib/fluent/plugin/out_bigquery.rb

Overview

TODO: error classes for each api error responses class BigQueryAPIError < StandardError end

Defined Under Namespace

Classes: BooleanFieldSchema, FieldSchema, FloatFieldSchema, IntegerFieldSchema, RecordSchema, StringFieldSchema, TimestampFieldSchema

Instance Method Summary

• #client ⇒ Object
• #configure(conf) ⇒ Object
• #format_stream(tag, es) ⇒ Object
• #initialize ⇒ BigQueryOutput constructor

  Table types https://developers.google.com/bigquery/docs/tables.

• #insert(table_id, rows) ⇒ Object
• #load ⇒ Object
• #shutdown ⇒ Object
• #start ⇒ Object
• #write(chunk) ⇒ Object

Constructor Details

#initialize ⇒ BigQueryOutput

Table types https://developers.google.com/bigquery/docs/tables

type - The following data types are supported; see Data Formats for details on each data type: STRING INTEGER FLOAT BOOLEAN RECORD A JSON object, used when importing nested records. This type is only available when using JSON source files.

mode - Whether a field can be null. The following values are supported: NULLABLE - The cell can be null. REQUIRED - The cell cannot be null. REPEATED - Zero or more repeated simple or nested subfields. This mode is only supported when using JSON source files.

# File 'lib/fluent/plugin/out_bigquery.rb', line 119

def initialize
  super
  require 'json'
  require 'google/api_client'
  require 'google/api_client/client_secrets'
  require 'google/api_client/auth/installed_app'
  require 'google/api_client/auth/compute_service_account'
end

Instance Method Details

#client ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 208

def client
  return @cached_client if @cached_client && @cached_client_expiration > Time.now

  client = Google::APIClient.new(
    :application_name => 'Fluentd BigQuery plugin',
    :application_version => Fluent::BigQueryPlugin::VERSION
  )

  case @auth_method
  when 'private_key'
    key = Google::APIClient::PKCS12.load_key( @private_key_path, @private_key_passphrase )
    asserter = Google::APIClient::JWTAsserter.new(
      @email,
      "https://www.googleapis.com/auth/bigquery",
      key
    )
    # refresh_auth
    client.authorization = asserter.authorize

  when 'compute_engine'
    auth = Google::APIClient::ComputeServiceAccount.new
    auth.fetch_access_token!
    client.authorization = auth

  else
    raise ConfigError, "Unknown auth method: #{@auth_method}"
  end

  @cached_client_expiration = Time.now + 1800
  @cached_client = client
end

#configure(conf) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 128

def configure(conf)
  super

  case @auth_method
  when 'private_key'
    if !@email || !@private_key_path
      raise Fluent::ConfigError, "'email' and 'private_key_path' must be specified if auth_method == 'private_key'"
    end
  when 'compute_engine'
    # Do nothing
  else
    raise Fluent::ConfigError, "unrecognized 'auth_method': #{@auth_method}"
  end

  if (!@table && !@tables) || (@table && @tables)
    raise Fluent::ConfigError, "'table' or 'tables' must be specified, and both are invalid"
  end

  @tablelist = @tables ? @tables.split(',') : [@table]

  @fields = RecordSchema.new
  if @schema_path
    @fields.load_schema(JSON.parse(File.read(@schema_path)))
  end
  if @field_string
    @field_string.split(',').each do |fieldname|
      @fields.register_field fieldname, :string
    end
  end
  if @field_integer
    @field_integer.split(',').each do |fieldname|
      @fields.register_field fieldname, :integer
    end
  end
  if @field_float
    @field_float.split(',').each do |fieldname|
      @fields.register_field fieldname, :float
    end
  end
  if @field_boolean
    @field_boolean.split(',').each do |fieldname|
      @fields.register_field fieldname, :boolean
    end
  end

  if @localtime.nil?
    if @utc
      @localtime = false
    end
  end
  @timef = TimeFormatter.new(@time_format, @localtime)

  if @time_field
    keys = @time_field.split('.')
    last_key = keys.pop
    @add_time_field = lambda {|record, time|
      keys.inject(record) { |h, k| h[k] ||= {} }[last_key] = @timef.format(time)
      record
    }
  else
    @add_time_field = lambda {|record, time| record }
  end
end

#format_stream(tag, es) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 275

def format_stream(tag, es)
  super
  buf = ''
  es.each do |time, record|
    row = @fields.format(@add_time_field.call(record, time))
    buf << {"json" => row}.to_msgpack unless row.empty?
  end
  buf
end

#insert(table_id, rows) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 240

def insert(table_id, rows)
  res = client().execute(
    :api_method => @bq.tabledata.insert_all,
    :parameters => {
      'projectId' => @project,
      'datasetId' => @dataset,
      'tableId' => table_id,
    },
    :body_object => {
      "rows" => rows
    }
  )
  unless res.success?
    # api_error? -> client cache clear
    @cached_client = nil

    message = res.body
    if res.body =~ /^\{/
      begin
        res_obj = JSON.parse(res.body)
        message = res_obj['error']['message'] || res.body
      rescue => e
        $log.warn "Parse error: google api error response body", :body => res.body
      end
    end
    $log.error "tabledata.insertAll API", :project_id => @project_id, :dataset => @dataset_id, :table => table_id, :code => res.status, :message => message
    raise "failed to insert into bigquery" # TODO: error class
  end
end

#load ⇒ Object

Raises:

• (NotImplementedError)

# File 'lib/fluent/plugin/out_bigquery.rb', line 270

def load
  # https://developers.google.com/bigquery/loading-data-into-bigquery#loaddatapostrequest
  raise NotImplementedError # TODO
end

#shutdown ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 203

def shutdown
  super
  # nothing to do
end

#start ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 192

def start
  super

  @bq = client.discovered_api("bigquery", "v2") # TODO: refresh with specified expiration
  @cached_client = nil
  @cached_client_expiration = nil

  @tables_queue = @tablelist.dup.shuffle
  @tables_mutex = Mutex.new
end

#write(chunk) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 285

def write(chunk)
  rows = []
  chunk.msgpack_each do |row_object|
    # TODO: row size limit
    rows << row_object
  end

  # TODO: method

  insert_table = @tables_mutex.synchronize do
    t = @tables_queue.shift
    @tables_queue.push t
    t
  end
  insert(insert_table, rows)
end