X-Git-Url: https://git.adam-barratt.org.uk/?a=blobdiff_plain;f=modules%2Froles%2Ffiles%2Fstatic-mirroring%2Fstatic-master-run;h=95c9355a5ffc6172f550f8489a09c1ebbae47fd7;hb=5dc5997952b5754e99c246f2887057de4ad4b126;hp=c82d68849e1003c1b30c9f2dc58d1b7069cd0a2a;hpb=fd2dbf1eb15d270f6a9f4f8467ba819a91f4d32f;p=mirror%2Fdsa-puppet.git diff --git a/modules/roles/files/static-mirroring/static-master-run b/modules/roles/files/static-mirroring/static-master-run index c82d68849..95c9355a5 100755 --- a/modules/roles/files/static-mirroring/static-master-run +++ b/modules/roles/files/static-mirroring/static-master-run @@ -9,25 +9,36 @@ import sys import tempfile import time -base='/home/staticsync/static-master' -subdirs = { 'master': 'master', # where updates from off-site end up going, the source of everything we do here - 'cur': 'current-push', # where clients rsync from during a mirror push - 'live': 'current-live'} # what is currently on the mirrors, and what they rsync from when they come back from being down serialname = '.serial' +had_warnings = False -clients = [] +conffile = '/etc/staticsync.conf' +config={} + +with open(conffile) as f: + for line in f: + line = line.rstrip() + if not line or line.startswith("#"): continue + (name, value) = line.split("=") + config[name] = value + +for key in ('base',): + if not key in config: + raise Exception("Configuration element '%s' not found in config file %s"%(key, conffile)) + +allclients = set() with open('/etc/static-clients.conf') as f: for line in f: line = line.strip() if line == "": continue if line.startswith('#'): continue - clients.append(line) + allclients.add(line) def log(m): t = time.strftime("[%Y-%m-%d %H:%M:%S]", time.gmtime()) print t, m -def stage1(pipes, status): +def stage1(pipes, status, clients): for c in clients: p = pipes[c] while 1: @@ -73,7 +84,7 @@ def count_statuses(status): else: cnt[v] += 1 return cnt -def stage2(pipes, status, command): +def stage2(pipes, status, command, clients): for c in clients: if status[c] != 'waiting': continue log("%s << %s"%(c, command)) @@ -88,54 +99,101 @@ def stage2(pipes, status, command): log("%s >> %s"%(c, l)) log("%s: returned %d"%(c, p.returncode)) -def callout(serial): +def callout(component, serial, clients): log("Calling clients...") pipes = {} status = {} for c in clients: - args = ['ssh', '-o', 'BatchMode=yes', c, 'mirror', "%d"%(serial,)] + args = ['ssh', '-o', 'BatchMode=yes', c, 'mirror', component, "%d"%(serial,)] p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) pipes[c] = p status[c] = 'in-progress' log("Stage 1...") - stage1(pipes, status) + stage1(pipes, status, clients) log("Stage 1 done.") cnt = count_statuses(status) - if 'failed' in cnt > 0: - log("Some clients failed, aborting...") - stage2(pipes, status, 'abort') + if 'failed' in cnt and cnt['failed'] >= 2: + log("%d clients failed, aborting..."%(cnt['failed'],)) + stage2(pipes, status, 'abort', clients) return False - elif 'waiting' in cnt > 0: + + failedmirrorsfile = os.path.join(config['base'], 'master', component + "-failedmirrors") + if 'failed' in cnt: + log("WARNING: %d clients failed! Continuing anyway!"%(cnt['failed'],)) + global had_warnings + had_warnings = True + f = open(failedmirrorsfile, "w") + for c in status: + if status[c] == 'failed': f.write(c+"\n") + f.close() + else: + if os.path.exists(failedmirrorsfile): os.unlink(failedmirrorsfile) + + if 'waiting' in cnt: log("Committing...") - stage2(pipes, status, 'go') + stage2(pipes, status, 'go', clients) return True else: log("All clients up to date.") return True +def load_component_info(component): + with open('/etc/static-components.conf') as f: + for line in f: + if line.startswith('#'): continue + field = line.strip().split() + if len(field) < 4: continue + if field[1] != component: continue + meta = {} + meta['master'] = field[0] + meta['sourcehost'] = field[2] + meta['sourcedir'] = field[3] + meta['extrapushhosts'] = set(field[4].split(',')) if len(field) > 4 else set() + meta['extraignoreclients'] = set(field[5].split(',')) if len(field) > 5 else set() + return meta + else: + return None cleanup_dirs = [] -def run_mirror(): +def run_mirror(component): + meta = load_component_info(component) + if meta is None: + log("Component %s not found."%(component,)) + return False + clients = allclients - meta['extraignoreclients'] + # setup - master = os.path.join(base, subdirs['master']) - cur = os.path.join(base, subdirs['cur']) - live = os.path.join(base, subdirs['live']) - tmpdir_new = tempfile.mkdtemp(prefix='live.new-', dir=base); cleanup_dirs.append(tmpdir_new); - tmpdir_old = tempfile.mkdtemp(prefix='live.old-', dir=base); cleanup_dirs.append(tmpdir_old); + basemaster = os.path.join(config['base'], 'master') + componentdir = os.path.join(basemaster, component) + cur = componentdir + '-current-push' + live = componentdir + '-current-live' + tmpdir_new = tempfile.mkdtemp(prefix=component+'-live.new-', dir=basemaster); cleanup_dirs.append(tmpdir_new); + tmpdir_old = tempfile.mkdtemp(prefix=component+'-live.old-', dir=basemaster); cleanup_dirs.append(tmpdir_old); os.chmod(tmpdir_new, 0755) locks = [] - for p in (master, live, tmpdir_new): - if not os.path.exists(p): os.mkdir(p, 0755) + lockfiles = [ os.path.join(basemaster, component + ".lock") ] + for p in lockfiles: fd = os.open(p, os.O_RDONLY) log("Acquiring lock for %s(%d)."%(p,fd)) fcntl.flock(fd, fcntl.LOCK_EX) locks.append(fd) log("All locks acquired.") - serialfile = os.path.join(master, serialname) + for p in (live, ): + if not os.path.exists(p): os.mkdir(p, 0755) + + #for p in (componentdir, live, tmpdir_new): + # if not os.path.exists(p): os.mkdir(p, 0755) + # fd = os.open(p, os.O_RDONLY) + # log("Acquiring lock for %s(%d)."%(p,fd)) + # fcntl.flock(fd, fcntl.LOCK_EX) + # locks.append(fd) + #log("All locks acquired.") + + serialfile = os.path.join(componentdir, serialname) try: with open(serialfile) as f: serial = int(f.read()) except: @@ -144,7 +202,7 @@ def run_mirror(): log("Serial is %s."%(serial,)) log("Populating %s."%(tmpdir_new,)) - subprocess.check_call(['cp', '-al', os.path.join(master, '.'), tmpdir_new]) + subprocess.check_call(['cp', '-al', os.path.join(componentdir, '.'), tmpdir_new]) if os.path.exists(cur): log("Removing existing %s."%(cur,)) @@ -153,7 +211,7 @@ def run_mirror(): log("Renaming %s to %s."%(tmpdir_new, cur)) os.rename(tmpdir_new, cur) - proceed = callout(serial) + proceed = callout(component, serial, clients) if proceed: log("Moving %s aside."%(live,)) @@ -162,21 +220,27 @@ def run_mirror(): os.rename(cur, live) log("Cleaning up.") shutil.rmtree(tmpdir_old) - log("Done.") + if had_warnings: log("Done, with warnings.") + else: log("Done.") ret = True else: log("Aborted.") ret = False for fd in locks: - fd.close() + os.close(fd) return ret +if len(sys.argv) != 2: + print >> sys.stderr, "Usage: %s "%(sys.argv[0],) + sys.exit(1) +component = sys.argv[1] + ok = False try: - ok = run_mirror() + ok = run_mirror(component) finally: for p in cleanup_dirs: if os.path.exists(p): shutil.rmtree(p)