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_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_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.
- #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.
8 9 10 11 |
# File 'lib/flydata/sync_file_manager.rb', line 8 def initialize(data_entry) @data_entry = data_entry @table_position_files = {} # File objects keyed by table name end |
Instance Method Details
#backup_dir ⇒ Object
300 301 302 |
# File 'lib/flydata/sync_file_manager.rb', line 300 def backup_dir BACKUP_DIR end |
#backup_dump_dir ⇒ Object
292 293 294 295 296 297 298 |
# File 'lib/flydata/sync_file_manager.rb', line 292 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
103 104 105 |
# File 'lib/flydata/sync_file_manager.rb', line 103 def binlog_path File.join(FLYDATA_HOME, @data_entry['name'] + ".binlog.pos") end |
#close ⇒ Object
13 14 15 16 |
# File 'lib/flydata/sync_file_manager.rb', line 13 def close @table_position_files.values.each {|f| f.close } @table_position_files = {} end |
#delete_dump_file ⇒ Object
288 289 290 |
# File 'lib/flydata/sync_file_manager.rb', line 288 def delete_dump_file FileUtils.rm(dump_file_path) if File.exists?(dump_file_path) end |
#delete_table_binlog_pos(table_name) ⇒ Object
255 256 257 258 259 260 261 262 |
# File 'lib/flydata/sync_file_manager.rb', line 255 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
18 19 20 |
# File 'lib/flydata/sync_file_manager.rb', line 18 def dump_file_path File.join(dump_dir, @data_entry['name']) + ".dump" end |
#dump_pos_path ⇒ Object
dump pos file for resume
23 24 25 |
# File 'lib/flydata/sync_file_manager.rb', line 23 def dump_pos_path dump_file_path + ".pos" end |
#get_new_table_list(tables, file_type) ⇒ Object
68 69 70 71 72 73 74 75 |
# File 'lib/flydata/sync_file_manager.rb', line 68 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
218 219 220 221 222 |
# File 'lib/flydata/sync_file_manager.rb', line 218 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.
172 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 |
# File 'lib/flydata/sync_file_manager.rb', line 172 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 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
246 247 248 249 250 251 252 253 |
# File 'lib/flydata/sync_file_manager.rb', line 246 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
273 274 275 276 277 278 279 280 281 282 283 284 285 286 |
# File 'lib/flydata/sync_file_manager.rb', line 273 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
96 97 98 99 100 101 |
# File 'lib/flydata/sync_file_manager.rb', line 96 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
33 34 35 36 37 38 39 40 41 42 |
# File 'lib/flydata/sync_file_manager.rb', line 33 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
44 45 46 47 48 49 50 51 52 53 54 |
# File 'lib/flydata/sync_file_manager.rb', line 44 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_sync_info ⇒ Object
211 212 213 214 215 216 |
# File 'lib/flydata/sync_file_manager.rb', line 211 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
78 79 80 |
# File 'lib/flydata/sync_file_manager.rb', line 78 def mysql_table_marshal_dump_path dump_file_path + ".mysql_table" end |
#reset_table_position_files(tables) ⇒ Object
table files
137 138 139 140 141 142 |
# File 'lib/flydata/sync_file_manager.rb', line 137 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
89 90 91 92 93 94 |
# File 'lib/flydata/sync_file_manager.rb', line 89 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
27 28 29 30 31 |
# File 'lib/flydata/sync_file_manager.rb', line 27 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
56 57 58 59 60 61 62 63 64 65 66 |
# File 'lib/flydata/sync_file_manager.rb', line 56 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
82 83 84 85 86 |
# File 'lib/flydata/sync_file_manager.rb', line 82 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_sent_binlog(binlog_pos) ⇒ Object
sent binlog.pos file
108 109 110 111 112 |
# File 'lib/flydata/sync_file_manager.rb', line 108 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
130 131 132 133 134 |
# File 'lib/flydata/sync_file_manager.rb', line 130 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
205 206 207 208 209 |
# File 'lib/flydata/sync_file_manager.rb', line 205 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
264 265 266 267 268 269 270 271 |
# File 'lib/flydata/sync_file_manager.rb', line 264 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
114 115 116 117 118 119 |
# File 'lib/flydata/sync_file_manager.rb', line 114 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
122 123 124 125 126 127 128 |
# File 'lib/flydata/sync_file_manager.rb', line 122 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 |
#sync_info_file ⇒ Object
201 202 203 |
# File 'lib/flydata/sync_file_manager.rb', line 201 def sync_info_file File.join(dump_dir, "sync.info") end |
#table_binlog_pos_init_paths(*tables) ⇒ Object
163 164 165 166 |
# File 'lib/flydata/sync_file_manager.rb', line 163 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
158 159 160 161 |
# File 'lib/flydata/sync_file_manager.rb', line 158 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
153 154 155 156 |
# File 'lib/flydata/sync_file_manager.rb', line 153 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
148 149 150 151 |
# File 'lib/flydata/sync_file_manager.rb', line 148 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
144 145 146 |
# File 'lib/flydata/sync_file_manager.rb', line 144 def table_positions_dir_path TABLE_POSITIONS_DIR end |
#table_rev(table_name) ⇒ Object
233 234 235 236 237 238 239 240 241 242 243 244 |
# File 'lib/flydata/sync_file_manager.rb', line 233 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
224 225 226 |
# File 'lib/flydata/sync_file_manager.rb', line 224 def table_rev_file_path(table_name) File.join(table_positions_dir_path, table_name + ".rev") end |
#table_rev_file_paths(*tables) ⇒ Object
228 229 230 231 |
# File 'lib/flydata/sync_file_manager.rb', line 228 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 |