Class: Flydata::SyncFileManager
- Inherits:
-
Object
- Object
- Flydata::SyncFileManager
- Defined in:
- lib/flydata/sync_file_manager.rb
Constant Summary collapse
- DUMP_DIR =
ENV['FLYDATA_DUMP'] || File.join(FLYDATA_HOME, 'dump')
- BACKUP_DIR =
ENV['FLYDATA_BACKUP'] || File.join(FLYDATA_HOME, 'backup')
- TABLE_POSITIONS_DIR =
ENV['FLYDATA_TABLE_POSITIONS'] || File.join(FLYDATA_HOME, 'positions')
Instance Method Summary collapse
- #backup_dir ⇒ Object
- #backup_dump_dir ⇒ Object
- #binlog_path ⇒ Object
- #close ⇒ Object
- #delete_dump_file ⇒ Object
- #delete_table_binlog_pos(table_name) ⇒ Object
- #dump_file_path ⇒ Object
-
#dump_pos_path ⇒ Object
dump pos file for resume.
- #get_new_table_list(tables, file_type) ⇒ Object
- #get_table_binlog_pos(table_name) ⇒ Object
-
#increment_and_save_table_position(table_name) ⇒ Object
Read a sequence number from the table’s position file, increment the number and pass the number to a block.
- #increment_table_rev(table_name, base_rev) ⇒ Object
-
#initialize(data_entry) ⇒ SyncFileManager
constructor
A new instance of SyncFileManager.
- #install_table_binlog_files(tables) ⇒ Object
- #load_binlog(file_path = binlog_path) ⇒ Object
- #load_dump_pos ⇒ Object
- #load_generated_ddl(tables) ⇒ Object
- #load_stats ⇒ Object
- #load_sync_info ⇒ Object
-
#mysql_table_marshal_dump_path ⇒ Object
MysqlTable marshal file.
-
#reset_table_position_files(tables) ⇒ Object
table files.
-
#save_binlog(binlog_pos) ⇒ Object
master binlog.pos file.
- #save_dump_pos(status, table_name, last_pos, binlog_pos, state = nil, substate = nil) ⇒ Object
- #save_generated_ddl(tables, contents = "1") ⇒ Object
- #save_mysql_table_marshal_dump(mysql_table) ⇒ Object
- #save_record_count_stat(table, record_count) ⇒ Object
-
#save_sent_binlog(binlog_pos) ⇒ Object
sent binlog.pos file.
- #save_ssl_ca(ssl_ca_content, path = ssl_ca_path) ⇒ Object
- #save_sync_info(initial_sync, tables) ⇒ Object
- #save_table_binlog_pos(tables, binlog_pos) ⇒ Object
- #sent_binlog_path(master_binlog_path = binlog_path) ⇒ Object
-
#ssl_ca_path(master_binlog_path = binlog_path) ⇒ Object
ssl_ca file path.
- #stats_path ⇒ Object
- #sync_info_file ⇒ Object
- #table_binlog_pos_init_paths(*tables) ⇒ Object
- #table_binlog_pos_paths(*tables) ⇒ Object
- #table_ddl_file_paths(*tables) ⇒ Object
- #table_position_file_paths(*tables) ⇒ Object
- #table_positions_dir_path ⇒ Object
- #table_rev(table_name) ⇒ Object
- #table_rev_file_path(table_name) ⇒ Object
- #table_rev_file_paths(*tables) ⇒ Object
Constructor Details
#initialize(data_entry) ⇒ SyncFileManager
Returns a new instance of SyncFileManager.
9 10 11 12 |
# File 'lib/flydata/sync_file_manager.rb', line 9 def initialize(data_entry) @data_entry = data_entry @table_position_files = {} # File objects keyed by table name end |
Instance Method Details
#backup_dir ⇒ Object
302 303 304 |
# File 'lib/flydata/sync_file_manager.rb', line 302 def backup_dir BACKUP_DIR end |
#backup_dump_dir ⇒ Object
294 295 296 297 298 299 300 |
# File 'lib/flydata/sync_file_manager.rb', line 294 def backup_dump_dir backup_dir = BACKUP_DIR.dup FileUtils.mkdir_p(backup_dir) unless Dir.exists?(backup_dir) dest_dir = File.join(backup_dir, Time.now.strftime("%Y%m%d%H%M%S")) FileUtils.mkdir(dest_dir) FileUtils.mv(Dir.glob("#{dump_dir}/*"), dest_dir) end |
#binlog_path ⇒ Object
104 105 106 |
# File 'lib/flydata/sync_file_manager.rb', line 104 def binlog_path File.join(FLYDATA_HOME, @data_entry['name'] + ".binlog.pos") end |
#close ⇒ Object
14 15 16 17 |
# File 'lib/flydata/sync_file_manager.rb', line 14 def close @table_position_files.values.each {|f| f.close } @table_position_files = {} end |
#delete_dump_file ⇒ Object
290 291 292 |
# File 'lib/flydata/sync_file_manager.rb', line 290 def delete_dump_file FileUtils.rm(dump_file_path) if File.exists?(dump_file_path) end |
#delete_table_binlog_pos(table_name) ⇒ Object
257 258 259 260 261 262 263 264 |
# File 'lib/flydata/sync_file_manager.rb', line 257 def delete_table_binlog_pos(table_name) file = File.join(table_positions_dir_path, table_name + ".binlog.pos") if File.exists?(file) FileUtils.rm(file, :force => true) else puts "#{file} does not exist. Something is wrong. Did you delete the file manually when flydata was running?" end end |
#dump_file_path ⇒ Object
19 20 21 |
# File 'lib/flydata/sync_file_manager.rb', line 19 def dump_file_path File.join(dump_dir, @data_entry['name']) + ".dump" end |
#dump_pos_path ⇒ Object
dump pos file for resume
24 25 26 |
# File 'lib/flydata/sync_file_manager.rb', line 24 def dump_pos_path dump_file_path + ".pos" end |
#get_new_table_list(tables, file_type) ⇒ Object
69 70 71 72 73 74 75 76 |
# File 'lib/flydata/sync_file_manager.rb', line 69 def get_new_table_list(tables, file_type) table_positions_dir_path = ENV['FLYDATA_TABLE_POSITIONS'] || File.join(FLYDATA_HOME, 'positions') new_tables = [] tables.each do |table| new_tables << table unless File.exists?(File.join(table_positions_dir_path, "#{table}.#{file_type}")) end new_tables end |
#get_table_binlog_pos(table_name) ⇒ Object
220 221 222 223 224 |
# File 'lib/flydata/sync_file_manager.rb', line 220 def get_table_binlog_pos(table_name) file = File.join(table_positions_dir_path, table_name + ".binlog.pos") return nil unless File.exists?(file) File.open(file, 'r').readline end |
#increment_and_save_table_position(table_name) ⇒ Object
Read a sequence number from the table’s position file, increment the number and pass the number to a block. After executing the block, saves the value to the position file.
173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 |
# File 'lib/flydata/sync_file_manager.rb', line 173 def increment_and_save_table_position(table_name) file = File.join(table_positions_dir_path, table_name + ".pos") retry_count = 0 begin @table_position_files[table_name] ||= File.open(file, "r+") rescue Errno::ENOENT raise if retry_count > 0 # Already retried. Must be a differentfile causing the error # File not exist. Create one with initial value of '0' File.open(file, "w") {|f| f.write('0') } retry_count += 1 retry end f = @table_position_files[table_name] seq = f.read seq = seq.to_i + 1 seq = FlydataCore::QueryJob::SYNC_FIRST_SEQ if seq == 1 begin yield(seq) ensure # when an error happened in yield, the sequence number should remain # as is. For the next call to read the value correctly, the position # must be rewound. f.rewind end f.truncate(0) f.write(seq) f.flush f.rewind end |
#increment_table_rev(table_name, base_rev) ⇒ Object
248 249 250 251 252 253 254 255 |
# File 'lib/flydata/sync_file_manager.rb', line 248 def increment_table_rev(table_name, base_rev) file = table_rev_file_path(table_name) new_rev = base_rev + 1 File.open(file, "w") do |f| f.write(new_rev) end new_rev end |
#install_table_binlog_files(tables) ⇒ Object
275 276 277 278 279 280 281 282 283 284 285 286 287 288 |
# File 'lib/flydata/sync_file_manager.rb', line 275 def install_table_binlog_files(tables) FileUtils.mkdir_p(table_positions_dir_path) unless Dir.exists?(table_positions_dir_path) tables.each do |table_name| file_name = table_name + ".binlog.pos" src_file = File.join(dump_dir, file_name) if ! File.exists?(src_file) raise "#{src_file} does not exist. Error!!" end FileUtils.mv(src_file, table_positions_dir_path) # save the position at initial sync. this is used for repair if # necessary. FileUtils.cp(File.join(table_positions_dir_path, file_name), File.join(table_positions_dir_path, file_name + ".init")) end end |
#load_binlog(file_path = binlog_path) ⇒ Object
97 98 99 100 101 102 |
# File 'lib/flydata/sync_file_manager.rb', line 97 def load_binlog(file_path = binlog_path) return nil unless File.exists?(file_path) f, pos = IO.read(file_path).strip.split("\t") return nil if f.nil? || f.empty? || pos.nil? { binfile: f, pos: pos.to_i } end |
#load_dump_pos ⇒ Object
34 35 36 37 38 39 40 41 42 43 |
# File 'lib/flydata/sync_file_manager.rb', line 34 def load_dump_pos path = dump_pos_path return {} unless File.exists?(path) items = File.open(path, 'r').readline.split("\t") raise "Invalid dump.pos file: #{path}" unless items.length >= 5 && items.length <= 7 mysql_table = load_mysql_table_marshal_dump { status: items[0], table_name: items[1], last_pos: items[2].to_i, binlog_pos: {binfile: items[3], pos: items[4].to_i}, state: items[5], substate: items[6], mysql_table: mysql_table} end |
#load_generated_ddl(tables) ⇒ Object
45 46 47 48 49 50 51 52 53 54 55 |
# File 'lib/flydata/sync_file_manager.rb', line 45 def load_generated_ddl(tables) tables = [ tables ] unless tables.kind_of?(Array) paths = table_ddl_file_paths(*tables) paths.collect{|path| begin File.open(path) {|f| f.read } rescue Errno::ENOENT nil end } end |
#load_stats ⇒ Object
316 317 318 319 |
# File 'lib/flydata/sync_file_manager.rb', line 316 def load_stats return nil unless File.exists?(stats_path) Hash[*File.read(stats_path).split(/\t/)] end |
#load_sync_info ⇒ Object
213 214 215 216 217 218 |
# File 'lib/flydata/sync_file_manager.rb', line 213 def load_sync_info return nil unless File.exists?(sync_info_file) items = File.open(sync_info_file, 'r').readline.split("\t") { initial_sync: (items[0] == 'true'), tables: items[1].split(" ") } end |
#mysql_table_marshal_dump_path ⇒ Object
MysqlTable marshal file
79 80 81 |
# File 'lib/flydata/sync_file_manager.rb', line 79 def mysql_table_marshal_dump_path dump_file_path + ".mysql_table" end |
#reset_table_position_files(tables) ⇒ Object
table files
138 139 140 141 142 143 |
# File 'lib/flydata/sync_file_manager.rb', line 138 def reset_table_position_files(tables) tables.each do |table_name| file = File.join(table_positions_dir_path, table_name + ".pos") File.open(file, "w") {|f| f.write('0') } end end |
#save_binlog(binlog_pos) ⇒ Object
master binlog.pos file
90 91 92 93 94 95 |
# File 'lib/flydata/sync_file_manager.rb', line 90 def save_binlog(binlog_pos) path = binlog_path File.open(path, 'w') do |f| f.write(binlog_content(binlog_pos)) end end |
#save_dump_pos(status, table_name, last_pos, binlog_pos, state = nil, substate = nil) ⇒ Object
28 29 30 31 32 |
# File 'lib/flydata/sync_file_manager.rb', line 28 def save_dump_pos(status, table_name, last_pos, binlog_pos, state = nil, substate = nil) File.open(dump_pos_path, 'w') do |f| f.write(dump_pos_content(status, table_name, last_pos, binlog_pos, state, substate)) end end |
#save_generated_ddl(tables, contents = "1") ⇒ Object
57 58 59 60 61 62 63 64 65 66 67 |
# File 'lib/flydata/sync_file_manager.rb', line 57 def save_generated_ddl(tables, contents = "1") tables = [ tables ] unless tables.kind_of?(Array) table_positions_dir_path = ENV['FLYDATA_TABLE_POSITIONS'] || File.join(FLYDATA_HOME, 'positions') #Create positions if dir does not exist unless File.directory?(table_positions_dir_path) FileUtils.mkdir_p(table_positions_dir_path) end tables.each do |tab| File.open(File.join(table_positions_dir_path, "#{tab}.generated_ddl"), 'w') {|f| f.write(contents) } end end |
#save_mysql_table_marshal_dump(mysql_table) ⇒ Object
83 84 85 86 87 |
# File 'lib/flydata/sync_file_manager.rb', line 83 def save_mysql_table_marshal_dump(mysql_table) File.open(mysql_table_marshal_dump_path, 'w') do |f| f.write Marshal.dump(mysql_table) end end |
#save_record_count_stat(table, record_count) ⇒ Object
310 311 312 313 314 |
# File 'lib/flydata/sync_file_manager.rb', line 310 def save_record_count_stat(table, record_count) stats = load_stats || Hash.new stats[table] = stats[table] ? stats[table].to_i + record_count : record_count save_stats(stats) end |
#save_sent_binlog(binlog_pos) ⇒ Object
sent binlog.pos file
109 110 111 112 113 |
# File 'lib/flydata/sync_file_manager.rb', line 109 def save_sent_binlog(binlog_pos) File.open(sent_binlog_path, 'w') do |f| f.write(binlog_content(binlog_pos)) end end |
#save_ssl_ca(ssl_ca_content, path = ssl_ca_path) ⇒ Object
131 132 133 134 135 |
# File 'lib/flydata/sync_file_manager.rb', line 131 def save_ssl_ca(ssl_ca_content, path = ssl_ca_path) File.open(path, 'w') do |f| f.write(ssl_ca_content) end end |
#save_sync_info(initial_sync, tables) ⇒ Object
207 208 209 210 211 |
# File 'lib/flydata/sync_file_manager.rb', line 207 def save_sync_info(initial_sync, tables) File.open(sync_info_file, "w") do |f| f.write([initial_sync, tables.join(" ")].join("\t")) end end |
#save_table_binlog_pos(tables, binlog_pos) ⇒ Object
266 267 268 269 270 271 272 273 |
# File 'lib/flydata/sync_file_manager.rb', line 266 def save_table_binlog_pos(tables, binlog_pos) tables.each do |table_name| file = File.join(dump_dir, table_name + ".binlog.pos") File.open(file, "w") do |f| f.write(binlog_content(binlog_pos)) end end end |
#sent_binlog_path(master_binlog_path = binlog_path) ⇒ Object
115 116 117 118 119 120 |
# File 'lib/flydata/sync_file_manager.rb', line 115 def sent_binlog_path(master_binlog_path = binlog_path) unless master_binlog_path && master_binlog_path.end_with?('binlog.pos') raise ArgumentError.new("Invalid binlog path. binlog path needs to end with 'binlog.pos'") end "#{master_binlog_path[0..-5]}.sent.pos" end |
#ssl_ca_path(master_binlog_path = binlog_path) ⇒ Object
ssl_ca file path
123 124 125 126 127 128 129 |
# File 'lib/flydata/sync_file_manager.rb', line 123 def ssl_ca_path(master_binlog_path = binlog_path) unless master_binlog_path && master_binlog_path.end_with?('binlog.pos') raise ArgumentError.new("Invalid binlog path. binlog path needs to end with 'binlog.pos'") end # <data-entry-name>.ssl_ca.pem "#{master_binlog_path[0..-12]}.ssl_ca.pem" end |
#stats_path ⇒ Object
306 307 308 |
# File 'lib/flydata/sync_file_manager.rb', line 306 def stats_path File.join(dump_dir, @data_entry['name']) + ".stats" end |
#sync_info_file ⇒ Object
203 204 205 |
# File 'lib/flydata/sync_file_manager.rb', line 203 def sync_info_file File.join(dump_dir, "sync.info") end |
#table_binlog_pos_init_paths(*tables) ⇒ Object
164 165 166 167 |
# File 'lib/flydata/sync_file_manager.rb', line 164 def table_binlog_pos_init_paths(*tables) tables.empty? ? Dir.glob(File.join(table_positions_dir_path, '*.binlog.pos.init')) : tables.map{|table| File.join(table_positions_dir_path, table + '.binlog.pos.init')} end |
#table_binlog_pos_paths(*tables) ⇒ Object
159 160 161 162 |
# File 'lib/flydata/sync_file_manager.rb', line 159 def table_binlog_pos_paths(*tables) tables.empty? ? Dir.glob(File.join(table_positions_dir_path, '*.binlog.pos')) : tables.map{|table| File.join(table_positions_dir_path, table + '.binlog.pos')} end |
#table_ddl_file_paths(*tables) ⇒ Object
154 155 156 157 |
# File 'lib/flydata/sync_file_manager.rb', line 154 def table_ddl_file_paths(*tables) tables.empty? ? Dir.glob(File.join(table_positions_dir_path, '*.generated_ddl')) : tables.map{|table| File.join(table_positions_dir_path, table + '.generated_ddl')} end |
#table_position_file_paths(*tables) ⇒ Object
149 150 151 152 |
# File 'lib/flydata/sync_file_manager.rb', line 149 def table_position_file_paths(*tables) tables.empty? ? Dir.glob(File.join(table_positions_dir_path, '*.pos')) : tables.map{|table| File.join(table_positions_dir_path, table + '.pos')} end |
#table_positions_dir_path ⇒ Object
145 146 147 |
# File 'lib/flydata/sync_file_manager.rb', line 145 def table_positions_dir_path TABLE_POSITIONS_DIR end |
#table_rev(table_name) ⇒ Object
235 236 237 238 239 240 241 242 243 244 245 246 |
# File 'lib/flydata/sync_file_manager.rb', line 235 def table_rev(table_name) file = table_rev_file_path(table_name) return 1 unless File.exists?(file) #default revision is 1 File.open(file, "r+") do |f| seq = f.read if seq.empty? return 1 else return seq.to_i end end end |
#table_rev_file_path(table_name) ⇒ Object
226 227 228 |
# File 'lib/flydata/sync_file_manager.rb', line 226 def table_rev_file_path(table_name) File.join(table_positions_dir_path, table_name + ".rev") end |
#table_rev_file_paths(*tables) ⇒ Object
230 231 232 233 |
# File 'lib/flydata/sync_file_manager.rb', line 230 def table_rev_file_paths(*tables) tables.empty? ? Dir.glob(File.join(table_positions_dir_path, "*.rev")) : tables.map{|table| table_rev_file_path(table)} end |