libexec/daily_afs_backups

   1 #!/usr/bin/python
   2
   3 import os, sys, time
   4 from mitsql import db
   5 from mitsql.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             err = ''
  33             try:
  34                 p0 = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  35                 p1 = subprocess.Popen(['gzip'], stdin=p0.stdout, stdout=file(next[1], 'w+'))
  36                 p1.wait()
  37                 err = p0.stderr.read()
  38             except Exception, e:
  39                 err = str(e)
  40             if len(err):
  41                 log.dump_errnum = p0.returncode
  42                 log.dump_errstr = err
  43             else:
  44                 log.dump_errnum = None
  45                 log.dump_errstr = None
  46             log.save_or_update()
  47             db.session.flush()
  48             #print 'Done'
  49         except (KeyboardInterrupt, SystemExit):
  50             break
  51         except Empty:
  52             if finished:
  53                 break
  54
  55 t_consumer = threading.Thread(target=consumer)
  56 t_consumer.start()
  57
  58 def producer():
  59     c = new_cursor('mysqldump')
  60     for dbname in get_dbs(c):
  61         log = db.Backup.get_by(db=dbname)
  62         if not log:
  63             log = db.Backup(db=dbname)
  64         elif log.skip_date and log.skip_date.timetuple:
  65             if time.mktime(log.skip_date.timetuple()) + 3600 > time.time():
  66                 # never recheck a db skipped in the past hour
  67                 continue
  68         d = db_backup_pre(c, dbname)
  69         if d[0]:
  70             queue.put((dbname, d[1]))
  71             log.skip_reason = None
  72             log.skip_date = None
  73         else:
  74             log.skip_reason = d[1]
  75             log.skip_date = db.func.now()
  76         log.save_or_update()
  77         #db.session.flush()
  78
  79 try:
  80     producer()
  81 except KeyboardInterrupt:
  82     sys.exit(1)
  83 finally:
  84     finished = True