libexec/daily_afs_backups

   1 #!/usr/bin/python
   2
   3 import os, sys
   4 import sql.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 = sql.db.Backup.get_by(db=next[0])
  25             if not log:
  26                 log = sql.db.Backup(db=next[0])
  27             log.dump_path = next[1]
  28             log.dump_date = sql.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             sql.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 = sql.db.Backup.get_by(db=db)
  58         if not log:
  59             log = sql.db.Backup(db=db)
  60         d = db_backup_pre(c, db)
  61         if d[0]:
  62             queue.put((db, d[1]))
  63             log.skip_reason = None
  64             log.skip_date = None
  65         else:
  66             log.skip_reason = d[1]
  67             log.skip_date = sql.db.func.now()
  68         log.save_or_update()
  69         sql.db.session.flush()
  70
  71 try:
  72     producer()
  73 except KeyboardInterrupt:
  74     sys.exit(1)
  75 finally:
  76     finished = True