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