Module: Fluent::FlydataSync
- Included in:
- MysqlBinlogFlydataInput
- Defined in:
- lib/flydata/fluent-plugins/flydata_plugin_ext/flydata_sync.rb
Class Method Summary collapse
Instance Method Summary collapse
Class Method Details
.included(base) ⇒ Object
10 11 12 13 14 15 16 17 18 19 20 21 22 |
# File 'lib/flydata/fluent-plugins/flydata_plugin_ext/flydata_sync.rb', line 10 def self.included(base) base.class_eval do include FlushSupport prepend TransactionSupport config_param :data_entry_name, :string, default: nil # data entry name config_param :data_entry_type, :string, default: nil # data entry type config_param :tables, :string config_param :tables_append_only, :string config_param :tag, :string config_param :position_file, :string, default: 'position.binlog.pos' end end |
Instance Method Details
#build_data_entry(base_object = {}) ⇒ Object
66 67 68 69 70 71 72 |
# File 'lib/flydata/fluent-plugins/flydata_plugin_ext/flydata_sync.rb', line 66 def build_data_entry(base_object = {}) de = base_object || {} de['name'] = @data_entry_name de['type'] = @data_entry_type de.merge!(@data_entry_preferences || {}) de end |
#configure(conf) ⇒ Object
24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
# File 'lib/flydata/fluent-plugins/flydata_plugin_ext/flydata_sync.rb', line 24 def configure(conf) super @source_position_file = self.class::SOURCE_POSITION_FILE_CLASS.new(@position_file) unless @source_position_file.exists? raise "No position file(#{@source_position_file.path}). Initial synchronization is required before starting." end load_custom_conf # preference module needs to be included @data_entry = build_data_entry @source = Flydata::Source.create(@data_entry) @sync_fm = Flydata::SyncFileManager.new(@data_entry, @source) sent_position_file_path = @sync_fm.sent_source_pos_path(@position_file) @sent_position_file = self.class::SOURCE_POSITION_FILE_CLASS.new(sent_position_file_path) # Create positions dir positions_path = @sync_fm.table_positions_dir_path Dir.mkdir positions_path unless File.exists? positions_path @tables = @tables.split(/(?:\s*,\s*|\s+)/) @omit_events = Hash.new @tables_append_only.split(/(?:\s*,\s*|\s+)/).each do |table| @tables << table unless @tables.include?(table) @omit_events[table] = [:delete, :truncate_table] end # Remove tables that do not have pos files new_tables = @sync_fm.get_new_table_list(@tables, "pos") @tables -= new_tables $log.info "Not watching these tables: #{new_tables.join(", ")}" # Set table revisions @table_revs = @tables.inject({}) do |h, table_name| h[table_name] = @sync_fm.table_rev(table_name) h end $log.info("Source position - resume_pos:'#{@source_position_file.read rescue IOError}' " + "sent_pos:'#{@sent_position_file.read rescue nil}'") end |