libexec/daily_afs_backups

   1 #!/usr/bin/python
   2
   3 import os, sys, time
   4 from sql import db
   5 from sql.util import new_cursor, get_dbs, db_backup_pre, db_backup_mkdir
   6 from Queue import Queue, Empty
   7 import threading
   8 import subprocess
   9
  10 MYSQLDUMP_ARGS = ['--compact', '--add-drop-table', '--add-locks',
  11     '--create-options', '--disable-keys', '--dump-date',
  12     '--extended-insert', '--quick',
  13     '--no-autocommit', '--quote-names', '--routines',
  14     '--single-transaction', '--triggers', '--force']
  15
  16 finished = False
  17 queue = Queue()
  18
  19 def consumer():
  20     while True:
  21         try:
  22             next = queue.get(timeout=3)
  23             print next[0] + ':',
  24             log = db.Backup.get_by(db=next[0])
  25             if not log:
  26                 log = db.Backup(db=next[0])
  27             log.dump_path = next[1]
  28             log.dump_date = db.func.now()
  29             db_backup_mkdir(next[1])
  30             args = ['mysqldump', next[0]]
  31             args.extend(MYSQLDUMP_ARGS)
  32             p0 = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  33             p1 = subprocess.Popen(['gzip'], stdin=p0.stdout, stdout=file(next[1], 'w+'))
  34             p1.wait()
  35             err = p0.stderr.read()
  36             if len(err):
  37                 log.dump_errnum = p0.returncode
  38                 log.dump_errstr = err
  39             else:
  40                 log.dump_errnum = None
  41                 log.dump_errstr = None
  42             log.save_or_update()
  43             db.session.flush()
  44             print 'Done'
  45         except (KeyboardInterrupt, SystemExit):
  46             break
  47         except Empty:
  48             if finished:
  49                 break
  50
  51 t_consumer = threading.Thread(target=consumer)
  52 t_consumer.start()
  53
  54 def producer():
  55     c = new_cursor('mysqldump')
  56     for db in get_dbs(c):
  57         log = db.Backup.get_by(db=db)
  58         if not log:
  59             log = db.Backup(db=db)
  60         elif log.skip_date:
  61             if time.mktime(log.skip_date.timetuple()) + 3600 > time.time():
  62                 # never recheck a db skipped in the past hour
  63                 continue
  64         d = db_backup_pre(c, db)
  65         if d[0]:
  66             queue.put((db, d[1]))
  67             log.skip_reason = None
  68             log.skip_date = None
  69         else:
  70             log.skip_reason = d[1]
  71             log.skip_date = db.func.now()
  72         log.save_or_update()
  73         #db.session.flush()
  74
  75 try:
  76     producer()
  77 except KeyboardInterrupt:
  78     sys.exit(1)
  79 finally:
  80     finished = True