Class: Fluent::Plugin::MysqlBulkOutput

Inherits:
Output
  • Object
show all
Defined in:
lib/fluent/plugin/out_mysql_bulk.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initializeMysqlBulkOutput

Returns a new instance of MysqlBulkOutput.



59
60
61
62
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 59

def initialize
  super
  require 'mysql2-cs-bind'
end

Instance Attribute Details

#handlerObject

Returns the value of attribute handler.



57
58
59
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 57

def handler
  @handler
end

Instance Method Details

#check_table_schema(database: @database, table: @table) ⇒ Object



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 107

def check_table_schema(database: @database, table: @table)
  _client = client(database)
  result = _client.xquery("SHOW COLUMNS FROM #{table}")
  max_lengths = []
  @column_names.each do |column|
    info = result.select { |x| x['Field'] == column }.first
    r = /(char|varchar)\(([\d]+)\)/
    begin
      max_length = info['Type'].scan(r)[0][1].to_i
    rescue
      max_length = nil
    end
    max_lengths << max_length
  end
  max_lengths
ensure
  if not _client.nil? then _client.close end
end

#client(database) ⇒ Object



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 139

def client(database)
  Mysql2::Client.new(
      host: @host,
      port: @port,
      username: @username,
      password: @password,
      database: database,
      sslkey: @sslkey,
      sslcert: @sslcert,
      sslca: @sslca,
      sslcapath: @sslcapath,
      sslcipher: @sslcipher,
      sslverify: @sslverify,
      flags: Mysql2::Client::MULTI_STATEMENTS
    )
end

#configure(conf) ⇒ Object



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 64

def configure(conf)
  compat_parameters_convert(conf, :buffer, :inject)
  super

  if @column_names.nil?
    fail Fluent::ConfigError, 'column_names MUST specified, but missing'
  end

  if @on_duplicate_key_update
    if @on_duplicate_update_keys.nil?
      fail Fluent::ConfigError, 'on_duplicate_key_update = true , on_duplicate_update_keys nil!'
    end
    @on_duplicate_update_keys = @on_duplicate_update_keys.split(',')

    if !@on_duplicate_update_custom_values.nil?
      @on_duplicate_update_custom_values = @on_duplicate_update_custom_values.split(',')
      if @on_duplicate_update_custom_values.length != @on_duplicate_update_keys.length
        fail Fluent::ConfigError, <<-DESC
on_duplicate_update_keys and on_duplicate_update_custom_values must be the same length
DESC
      end
    end

    @on_duplicate_key_update_sql = ' ON DUPLICATE KEY UPDATE '
    updates = []
    @on_duplicate_update_keys.each_with_index do |update_column, i|
      if @on_duplicate_update_custom_values.nil? || @on_duplicate_update_custom_values[i] == "#{update_column}"
        updates << "#{update_column} = VALUES(#{update_column})"
      else
        value = @on_duplicate_update_custom_values[i].to_s.match(/\${(.*)}/)[1]
        escape_value = Mysql2::Client.escape(value)
        updates << "#{update_column} = #{escape_value}"
      end
    end
    @on_duplicate_key_update_sql += updates.join(',')
  end

  @column_names = @column_names.split(',').collect(&:strip)
  @key_names = @key_names.nil? ? @column_names : @key_names.split(',').collect(&:strip)
  @json_key_names = @json_key_names.split(',') if @json_key_names
  @unixtimestamp_key_names = @unixtimestamp_key_names.split(',') if @unixtimestamp_key_names
end

#expand_placeholders(metadata) ⇒ Object



156
157
158
159
160
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 156

def expand_placeholders()
  database = extract_placeholders(@database, ).gsub('.', '_')
  table = extract_placeholders(@table, ).gsub('.', '_')
  return database, table
end

#format(tag, time, record) ⇒ Object



126
127
128
129
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 126

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

#formatted_to_msgpack_binaryObject



131
132
133
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 131

def formatted_to_msgpack_binary
  true
end

#multi_workers_ready?Boolean

Returns:

  • (Boolean)


135
136
137
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 135

def multi_workers_ready?
  true
end

#write(chunk) ⇒ Object



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
# File 'lib/fluent/plugin/out_mysql_bulk.rb', line 162

def write(chunk)
  database, table = expand_placeholders(chunk.)
  max_lengths = check_table_schema(database: database, table: table)
  @handler = client(database)
  values = []
  values_template = "(#{ @column_names.map { |key| '?' }.join(',') })"
  chunk.msgpack_each do |tag, time, data|
    data = format_proc.call(tag, time, data, max_lengths)
    values << Mysql2::Client.pseudo_bind(values_template, data)
  end
  sql = "INSERT INTO #{table} (#{@column_names.map{|x| "`#{x.to_s.gsub('`', '``')}`"}.join(',')}) VALUES #{values.join(',')}"
  sql += @on_duplicate_key_update_sql if @on_duplicate_key_update

  log.info "bulk insert values size (table: #{table}) => #{values.size}"
  @handler.query("SET SESSION TRANSACTION ISOLATION LEVEL #{transaction_isolation_level}") if @transaction_isolation_level
  @handler.xquery(sql)
  @handler.close
end