summaryrefslogtreecommitdiffstats
path: root/geo-replication/syncdaemon/master.py
diff options
context:
space:
mode:
Diffstat (limited to 'geo-replication/syncdaemon/master.py')
-rw-r--r--geo-replication/syncdaemon/master.py384
1 files changed, 234 insertions, 150 deletions
diff --git a/geo-replication/syncdaemon/master.py b/geo-replication/syncdaemon/master.py
index 98a61bc1d..4301396f9 100644
--- a/geo-replication/syncdaemon/master.py
+++ b/geo-replication/syncdaemon/master.py
@@ -1,25 +1,30 @@
+#
+# Copyright (c) 2011-2014 Red Hat, Inc. <http://www.redhat.com>
+# This file is part of GlusterFS.
+
+# This file is licensed to you under your choice of the GNU Lesser
+# General Public License, version 3 or any later version (LGPLv3 or
+# later), or the GNU General Public License, version 2 (GPLv2), in all
+# cases as published by the Free Software Foundation.
+#
+
import os
import sys
import time
import stat
-import random
-import signal
import json
import logging
import socket
import string
import errno
-from shutil import copyfileobj
-from errno import ENOENT, ENODATA, EPIPE, EEXIST, errorcode
-from threading import currentThread, Condition, Lock
+from errno import ENOENT, ENODATA, EPIPE, EEXIST
+from threading import Condition, Lock
from datetime import datetime
-from libcxattr import Xattr
-
from gconf import gconf
-from tempfile import mkdtemp, NamedTemporaryFile
-from syncdutils import FreeObject, Thread, GsyncdError, boolify, escape, \
- unescape, select, gauxpfx, md5hex, selfkill, entry2pb, \
- lstat, errno_wrap, update_file
+from tempfile import NamedTemporaryFile
+from syncdutils import Thread, GsyncdError, boolify, escape
+from syncdutils import unescape, select, gauxpfx, md5hex, selfkill
+from syncdutils import lstat, errno_wrap
URXTIME = (-1, 0)
@@ -27,18 +32,20 @@ URXTIME = (-1, 0)
# of the DRY principle (no, don't look for elevated or
# perspectivistic things here)
+
def _xtime_now():
t = time.time()
sec = int(t)
nsec = int((t - sec) * 1000000)
return (sec, nsec)
+
def _volinfo_hook_relax_foreign(self):
volinfo_sys = self.get_sys_volinfo()
fgn_vi = volinfo_sys[self.KFGN]
if fgn_vi:
expiry = fgn_vi['timeout'] - int(time.time()) + 1
- logging.info('foreign volume info found, waiting %d sec for expiry' % \
+ logging.info('foreign volume info found, waiting %d sec for expiry' %
expiry)
time.sleep(expiry)
volinfo_sys = self.get_sys_volinfo()
@@ -58,10 +65,14 @@ def gmaster_builder(excrawl=None):
logging.info('setting up %s change detection mode' % changemixin)
modemixin = getattr(this, modemixin.capitalize() + 'Mixin')
crawlmixin = getattr(this, 'GMaster' + changemixin.capitalize() + 'Mixin')
- sendmarkmixin = boolify(gconf.use_rsync_xattrs) and SendmarkRsyncMixin or SendmarkNormalMixin
- purgemixin = boolify(gconf.ignore_deletes) and PurgeNoopMixin or PurgeNormalMixin
+ sendmarkmixin = boolify(
+ gconf.use_rsync_xattrs) and SendmarkRsyncMixin or SendmarkNormalMixin
+ purgemixin = boolify(
+ gconf.ignore_deletes) and PurgeNoopMixin or PurgeNormalMixin
syncengine = boolify(gconf.use_tarssh) and TarSSHEngine or RsyncEngine
- class _GMaster(crawlmixin, modemixin, sendmarkmixin, purgemixin, syncengine):
+
+ class _GMaster(crawlmixin, modemixin, sendmarkmixin,
+ purgemixin, syncengine):
pass
return _GMaster
@@ -71,6 +82,7 @@ def gmaster_builder(excrawl=None):
# sync modes
class NormalMixin(object):
+
"""normal geo-rep behavior"""
minus_infinity = URXTIME
@@ -152,14 +164,18 @@ class NormalMixin(object):
self.slave.server.set_stime(path, self.uuid, mark)
# self.slave.server.set_xtime_remote(path, self.uuid, mark)
+
class PartialMixin(NormalMixin):
+
"""a variant tuned towards operation with a master
that has partial info of the slave (brick typically)"""
def xtime_reversion_hook(self, path, xtl, xtr):
pass
+
class RecoverMixin(NormalMixin):
+
"""a variant that differs from normal in terms
of ignoring non-indexed files"""
@@ -178,11 +194,13 @@ class RecoverMixin(NormalMixin):
# Further mixins for certain tunable behaviors
+
class SendmarkNormalMixin(object):
def sendmark_regular(self, *a, **kw):
return self.sendmark(*a, **kw)
+
class SendmarkRsyncMixin(object):
def sendmark_regular(self, *a, **kw):
@@ -194,19 +212,24 @@ class PurgeNormalMixin(object):
def purge_missing(self, path, names):
self.slave.server.purge(path, names)
+
class PurgeNoopMixin(object):
def purge_missing(self, path, names):
pass
+
class TarSSHEngine(object):
+
"""Sync engine that uses tar(1) piped over ssh(1)
for data transfers. Good for lots of small files.
"""
+
def a_syncdata(self, files):
logging.debug('files: %s' % (files))
for f in files:
pb = self.syncer.add(f)
+
def regjob(se, xte, pb):
rv = pb.wait()
if rv[0]:
@@ -228,13 +251,17 @@ class TarSSHEngine(object):
self.a_syncdata(files)
self.syncdata_wait()
+
class RsyncEngine(object):
+
"""Sync engine that uses rsync(1) for data transfers"""
+
def a_syncdata(self, files):
logging.debug('files: %s' % (files))
for f in files:
logging.debug('candidate for syncing %s' % f)
pb = self.syncer.add(f)
+
def regjob(se, xte, pb):
rv = pb.wait()
if rv[0]:
@@ -258,7 +285,9 @@ class RsyncEngine(object):
self.a_syncdata(files)
self.syncdata_wait()
+
class GMasterCommon(object):
+
"""abstract class impementling master role"""
KFGN = 0
@@ -269,8 +298,9 @@ class GMasterCommon(object):
err out on multiple foreign masters
"""
- fgn_vis, nat_vi = self.master.server.aggregated.foreign_volume_infos(), \
- self.master.server.aggregated.native_volume_info()
+ fgn_vis, nat_vi = (
+ self.master.server.aggregated.foreign_volume_infos(),
+ self.master.server.aggregated.native_volume_info())
fgn_vi = None
if fgn_vis:
if len(fgn_vis) > 1:
@@ -316,15 +346,14 @@ class GMasterCommon(object):
if getattr(gconf, 'state_detail_file', None):
try:
with open(gconf.state_detail_file, 'r+') as f:
- loaded_data= json.load(f)
- diff_data = set(default_data) - set (loaded_data)
+ loaded_data = json.load(f)
+ diff_data = set(default_data) - set(loaded_data)
if len(diff_data):
for i in diff_data:
loaded_data[i] = default_data[i]
return loaded_data
- except (IOError):
- ex = sys.exc_info()[1]
- logging.warn ('Creating new gconf.state_detail_file.')
+ except IOError:
+ logging.warn('Creating new gconf.state_detail_file.')
# Create file with initial data
try:
with open(gconf.state_detail_file, 'wb') as f:
@@ -364,7 +393,8 @@ class GMasterCommon(object):
# - self.turns is the number of turns since start
# - self.total_turns is a limit so that if self.turns reaches it, then
# we exit (for diagnostic purposes)
- # so, eg., if the master fs changes unceasingly, self.turns will remain 0.
+ # so, eg., if the master fs changes unceasingly, self.turns will remain
+ # 0.
self.crawls = 0
self.turns = 0
self.total_turns = int(gconf.turns)
@@ -394,7 +424,7 @@ class GMasterCommon(object):
t.start()
def should_crawl(cls):
- return (gconf.glusterd_uuid in cls.master.server.node_uuid())
+ return gconf.glusterd_uuid in cls.master.server.node_uuid()
def register(self):
self.register()
@@ -416,18 +446,18 @@ class GMasterCommon(object):
volinfo_sys = self.volinfo_hook()
self.volinfo = volinfo_sys[self.KNAT]
inter_master = volinfo_sys[self.KFGN]
- logging.info("%s master with volume id %s ..." % \
- (inter_master and "intermediate" or "primary",
- self.uuid))
+ logging.info("%s master with volume id %s ..." %
+ (inter_master and "intermediate" or "primary",
+ self.uuid))
gconf.configinterface.set('volume_id', self.uuid)
if self.volinfo:
if self.volinfo['retval']:
- logging.warn("master cluster's info may not be valid %d" % \
+ logging.warn("master cluster's info may not be valid %d" %
self.volinfo['retval'])
self.start_checkpoint_thread()
else:
raise GsyncdError("master volinfo unavailable")
- self.total_crawl_stats = self.get_initial_crawl_data()
+ self.total_crawl_stats = self.get_initial_crawl_data()
self.lastreport['time'] = time.time()
logging.info('crawl interval: %d seconds' % self.sleep_interval)
@@ -435,7 +465,7 @@ class GMasterCommon(object):
crawl = self.should_crawl()
while not self.terminate:
if self.start:
- logging.debug("... crawl #%d done, took %.6f seconds" % \
+ logging.debug("... crawl #%d done, took %.6f seconds" %
(self.crawls, time.time() - self.start))
self.start = time.time()
should_display_info = self.start - self.lastreport['time'] >= 60
@@ -443,11 +473,11 @@ class GMasterCommon(object):
logging.info("%d crawls, %d turns",
self.crawls - self.lastreport['crawls'],
self.turns - self.lastreport['turns'])
- self.lastreport.update(crawls = self.crawls,
- turns = self.turns,
- time = self.start)
+ self.lastreport.update(crawls=self.crawls,
+ turns=self.turns,
+ time=self.start)
t1 = time.time()
- if int(t1 - t0) >= 60: #lets hardcode this check to 60 seconds
+ if int(t1 - t0) >= 60: # lets hardcode this check to 60 seconds
crawl = self.should_crawl()
t0 = t1
self.update_worker_remote_node()
@@ -456,11 +486,14 @@ class GMasterCommon(object):
# bring up _this_ brick to the cluster stime
# which is min of cluster (but max of the replicas)
brick_stime = self.xtime('.', self.slave)
- cluster_stime = self.master.server.aggregated.stime_mnt('.', '.'.join([str(self.uuid), str(gconf.slave_id)]))
- logging.debug("Cluster stime: %s | Brick stime: %s" % (repr(cluster_stime), repr(brick_stime)))
+ cluster_stime = self.master.server.aggregated.stime_mnt(
+ '.', '.'.join([str(self.uuid), str(gconf.slave_id)]))
+ logging.debug("Cluster stime: %s | Brick stime: %s" %
+ (repr(cluster_stime), repr(brick_stime)))
if not isinstance(cluster_stime, int):
if brick_stime < cluster_stime:
- self.slave.server.set_stime(self.FLAT_DIR_HIERARCHY, self.uuid, cluster_stime)
+ self.slave.server.set_stime(
+ self.FLAT_DIR_HIERARCHY, self.uuid, cluster_stime)
time.sleep(5)
continue
self.update_worker_health("Active")
@@ -489,13 +522,14 @@ class GMasterCommon(object):
with checkpoint @chkpt"""
if xtimish:
val = cls.serialize_xtime(val)
- gconf.configinterface.set('checkpoint_' + prm, "%s:%s" % (escape(chkpt), val))
+ gconf.configinterface.set(
+ 'checkpoint_' + prm, "%s:%s" % (escape(chkpt), val))
@staticmethod
def humantime(*tpair):
"""format xtime-like (sec, nsec) pair to human readable format"""
ts = datetime.fromtimestamp(float('.'.join(str(n) for n in tpair))).\
- strftime("%Y-%m-%d %H:%M:%S")
+ strftime("%Y-%m-%d %H:%M:%S")
if len(tpair) > 1:
ts += '.' + str(tpair[1])
return ts
@@ -506,7 +540,7 @@ class GMasterCommon(object):
years = int(years)
days = int(days)
- date=""
+ date = ""
m, s = divmod(crawl_time.seconds, 60)
h, m = divmod(m, 60)
@@ -515,7 +549,8 @@ class GMasterCommon(object):
if days != 0:
date += "%s %s " % (days, "day" if days == 1 else "days")
- date += "%s:%s:%s" % (string.zfill(h, 2), string.zfill(m, 2), string.zfill(s, 2))
+ date += "%s:%s:%s" % (string.zfill(h, 2),
+ string.zfill(m, 2), string.zfill(s, 2))
return date
def checkpt_service(self, chan, chkpt):
@@ -540,16 +575,18 @@ class GMasterCommon(object):
if not checkpt_tgt:
checkpt_tgt = self.xtime('.')
if isinstance(checkpt_tgt, int):
- raise GsyncdError("master root directory is unaccessible (%s)",
+ raise GsyncdError("master root directory is "
+ "unaccessible (%s)",
os.strerror(checkpt_tgt))
self._set_checkpt_param(chkpt, 'target', checkpt_tgt)
- logging.debug("checkpoint target %s has been determined for checkpoint %s" % \
+ logging.debug("checkpoint target %s has been determined "
+ "for checkpoint %s" %
(repr(checkpt_tgt), chkpt))
# check if the label is 'now'
chkpt_lbl = chkpt
try:
- x1,x2 = chkpt.split(':')
+ x1, x2 = chkpt.split(':')
if x1 == 'now':
chkpt_lbl = "as of " + self.humantime(x2)
except:
@@ -557,41 +594,46 @@ class GMasterCommon(object):
completed = self._checkpt_param(chkpt, 'completed', xtimish=False)
if completed:
completed = tuple(int(x) for x in completed.split('.'))
- s,_,_ = select([chan], [], [], (not completed) and 5 or None)
+ s, _, _ = select([chan], [], [], (not completed) and 5 or None)
# either request made and we re-check to not
# give back stale data, or we still hunting for completion
- if self.native_xtime(checkpt_tgt) and self.native_xtime(checkpt_tgt) < self.volmark:
+ if (self.native_xtime(checkpt_tgt) and (
+ self.native_xtime(checkpt_tgt) < self.volmark)):
# indexing has been reset since setting the checkpoint
status = "is invalid"
else:
xtr = self.xtime('.', self.slave)
if isinstance(xtr, int):
- raise GsyncdError("slave root directory is unaccessible (%s)",
+ raise GsyncdError("slave root directory is "
+ "unaccessible (%s)",
os.strerror(xtr))
ncompleted = self.xtime_geq(xtr, checkpt_tgt)
- if completed and not ncompleted: # stale data
- logging.warn("completion time %s for checkpoint %s became stale" % \
+ if completed and not ncompleted: # stale data
+ logging.warn("completion time %s for checkpoint %s "
+ "became stale" %
(self.humantime(*completed), chkpt))
completed = None
gconf.configinterface.delete('checkpoint_completed')
- if ncompleted and not completed: # just reaching completion
+ if ncompleted and not completed: # just reaching completion
completed = "%.6f" % time.time()
- self._set_checkpt_param(chkpt, 'completed', completed, xtimish=False)
+ self._set_checkpt_param(
+ chkpt, 'completed', completed, xtimish=False)
completed = tuple(int(x) for x in completed.split('.'))
logging.info("checkpoint %s completed" % chkpt)
status = completed and \
- "completed at " + self.humantime(completed[0]) or \
- "not reached yet"
+ "completed at " + self.humantime(completed[0]) or \
+ "not reached yet"
if s:
conn = None
try:
conn, _ = chan.accept()
try:
- conn.send("checkpoint %s is %s\0" % (chkpt_lbl, status))
+ conn.send("checkpoint %s is %s\0" %
+ (chkpt_lbl, status))
except:
exc = sys.exc_info()[1]
- if (isinstance(exc, OSError) or isinstance(exc, IOError)) and \
- exc.errno == EPIPE:
+ if ((isinstance(exc, OSError) or isinstance(
+ exc, IOError)) and exc.errno == EPIPE):
logging.debug('checkpoint client disconnected')
else:
raise
@@ -602,11 +644,13 @@ class GMasterCommon(object):
def start_checkpoint_thread(self):
"""prepare and start checkpoint service"""
if self.checkpoint_thread or not (
- getattr(gconf, 'state_socket_unencoded', None) and getattr(gconf, 'socketdir', None)
+ getattr(gconf, 'state_socket_unencoded', None) and getattr(
+ gconf, 'socketdir', None)
):
return
chan = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- state_socket = os.path.join(gconf.socketdir, md5hex(gconf.state_socket_unencoded) + ".socket")
+ state_socket = os.path.join(
+ gconf.socketdir, md5hex(gconf.state_socket_unencoded) + ".socket")
try:
os.unlink(state_socket)
except:
@@ -621,9 +665,9 @@ class GMasterCommon(object):
def add_job(self, path, label, job, *a, **kw):
"""insert @job function to job table at @path with @label"""
- if self.jobtab.get(path) == None:
+ if self.jobtab.get(path) is None:
self.jobtab[path] = []
- self.jobtab[path].append((label, a, lambda : job(*a, **kw)))
+ self.jobtab[path].append((label, a, lambda: job(*a, **kw)))
def add_failjob(self, path, label):
"""invoke .add_job with a job that does nothing just fails"""
@@ -644,7 +688,7 @@ class GMasterCommon(object):
ret = j[-1]()
if not ret:
succeed = False
- if succeed and not args[0] == None:
+ if succeed and not args[0] is None:
self.sendmark(path, *args)
return succeed
@@ -657,19 +701,21 @@ class GMasterCommon(object):
self.slave.server.setattr(path, adct)
self.set_slave_xtime(path, mark)
+
class GMasterChangelogMixin(GMasterCommon):
+
""" changelog based change detection and syncing """
# index for change type and entry
IDX_START = 0
- IDX_END = 2
+ IDX_END = 2
- POS_GFID = 0
- POS_TYPE = 1
+ POS_GFID = 0
+ POS_TYPE = 1
POS_ENTRY1 = -1
- TYPE_META = "M "
- TYPE_GFID = "D "
+ TYPE_META = "M "
+ TYPE_GFID = "D "
TYPE_ENTRY = "E "
# flat directory heirarchy for gfid based access
@@ -686,18 +732,19 @@ class GMasterChangelogMixin(GMasterCommon):
def setup_working_dir(self):
workdir = os.path.join(gconf.working_dir, md5hex(gconf.local_path))
logfile = os.path.join(workdir, 'changes.log')
- logging.debug('changelog working dir %s (log: %s)' % (workdir, logfile))
+ logging.debug('changelog working dir %s (log: %s)' %
+ (workdir, logfile))
return (workdir, logfile)
def process_change(self, change, done, retry):
pfx = gauxpfx()
- clist = []
+ clist = []
entries = []
meta_gfid = set()
datas = set()
# basic crawl stats: files and bytes
- files_pending = {'count': 0, 'purge': 0, 'bytes': 0, 'files': []}
+ files_pending = {'count': 0, 'purge': 0, 'bytes': 0, 'files': []}
try:
f = open(change, "r")
clist = f.readlines()
@@ -750,17 +797,19 @@ class GMasterChangelogMixin(GMasterCommon):
elif ty in ['CREATE', 'MKDIR', 'MKNOD']:
entry_update()
# stat information present in the changelog itself
- entries.append(edct(ty, gfid=gfid, entry=en, mode=int(ec[2]),\
+ entries.append(edct(ty, gfid=gfid, entry=en,
+ mode=int(ec[2]),
uid=int(ec[3]), gid=int(ec[4])))
else:
# stat() to get mode and other information
go = os.path.join(pfx, gfid)
st = lstat(go)
if isinstance(st, int):
- if ty == 'RENAME': # special hack for renames...
+ if ty == 'RENAME': # special hack for renames...
entries.append(edct('UNLINK', gfid=gfid, entry=en))
else:
- logging.debug('file %s got purged in the interim' % go)
+ logging.debug(
+ 'file %s got purged in the interim' % go)
continue
if ty == 'LINK':
@@ -771,17 +820,20 @@ class GMasterChangelogMixin(GMasterCommon):
if isinstance(rl, int):
continue
entry_update()
- entries.append(edct(ty, stat=st, entry=en, gfid=gfid, link=rl))
+ entries.append(
+ edct(ty, stat=st, entry=en, gfid=gfid, link=rl))
elif ty == 'RENAME':
entry_update()
- e1 = unescape(os.path.join(pfx, ec[self.POS_ENTRY1 - 1]))
- entries.append(edct(ty, gfid=gfid, entry=e1, entry1=en, stat=st))
+ e1 = unescape(
+ os.path.join(pfx, ec[self.POS_ENTRY1 - 1]))
+ entries.append(
+ edct(ty, gfid=gfid, entry=e1, entry1=en, stat=st))
else:
logging.warn('ignoring %s [op %s]' % (gfid, ty))
elif et == self.TYPE_GFID:
datas.add(os.path.join(pfx, ec[0]))
elif et == self.TYPE_META:
- if ec[1] == 'SETATTR': # only setattr's for now...
+ if ec[1] == 'SETATTR': # only setattr's for now...
meta_gfid.add(os.path.join(pfx, ec[0]))
else:
logging.warn('got invalid changelog type: %s' % (et))
@@ -789,10 +841,10 @@ class GMasterChangelogMixin(GMasterCommon):
if not retry:
self.update_worker_cumilitive_status(files_pending)
# sync namespace
- if (entries):
+ if entries:
self.slave.server.entry_ops(entries)
# sync metadata
- if (meta_gfid):
+ if meta_gfid:
meta_entries = []
for go in meta_gfid:
st = lstat(go)
@@ -814,22 +866,25 @@ class GMasterChangelogMixin(GMasterCommon):
self.skipped_gfid_list = []
self.current_files_skipped_count = 0
- # first, fire all changelog transfers in parallel. entry and metadata
- # are performed synchronously, therefore in serial. However at the end
- # of each changelog, data is synchronized with syncdata_async() - which
- # means it is serial w.r.t entries/metadata of that changelog but
- # happens in parallel with data of other changelogs.
+ # first, fire all changelog transfers in parallel. entry and
+ # metadata are performed synchronously, therefore in serial.
+ # However at the end of each changelog, data is synchronized
+ # with syncdata_async() - which means it is serial w.r.t
+ # entries/metadata of that changelog but happens in parallel
+ # with data of other changelogs.
for change in changes:
logging.debug('processing change %s' % change)
self.process_change(change, done, retry)
if not retry:
- self.turns += 1 # number of changelogs processed in the batch
+ # number of changelogs processed in the batch
+ self.turns += 1
- # Now we wait for all the data transfers fired off in the above step
- # to complete. Note that this is not ideal either. Ideally we want to
- # trigger the entry/meta-data transfer of the next batch while waiting
- # for the data transfer of the current batch to finish.
+ # Now we wait for all the data transfers fired off in the above
+ # step to complete. Note that this is not ideal either. Ideally
+ # we want to trigger the entry/meta-data transfer of the next
+ # batch while waiting for the data transfer of the current batch
+ # to finish.
# Note that the reason to wait for the data transfer (vs doing it
# completely in the background and call the changelog_done()
@@ -837,10 +892,11 @@ class GMasterChangelogMixin(GMasterCommon):
# and prevents a spiraling increase of wait stubs from consuming
# unbounded memory and resources.
- # update the slave's time with the timestamp of the _last_ changelog
- # file time suffix. Since, the changelog prefix time is the time when
- # the changelog was rolled over, introduce a tolerence of 1 second to
- # counter the small delta b/w the marker update and gettimeofday().
+ # update the slave's time with the timestamp of the _last_
+ # changelog file time suffix. Since, the changelog prefix time
+ # is the time when the changelog was rolled over, introduce a
+ # tolerence of 1 second to counter the small delta b/w the
+ # marker update and gettimeofday().
# NOTE: this is only for changelog mode, not xsync.
# @change is the last changelog (therefore max time for this batch)
@@ -856,10 +912,13 @@ class GMasterChangelogMixin(GMasterCommon):
retry = True
tries += 1
if tries == self.MAX_RETRIES:
- logging.warn('changelogs %s could not be processed - moving on...' % \
+ logging.warn('changelogs %s could not be processed - '
+ 'moving on...' %
' '.join(map(os.path.basename, changes)))
- self.update_worker_total_files_skipped(self.current_files_skipped_count)
- logging.warn('SKIPPED GFID = %s' % ','.join(self.skipped_gfid_list))
+ self.update_worker_total_files_skipped(
+ self.current_files_skipped_count)
+ logging.warn('SKIPPED GFID = %s' %
+ ','.join(self.skipped_gfid_list))
self.update_worker_files_syncd()
if done:
xtl = (int(change.split('.')[-1]) - 1, 0)
@@ -873,7 +932,7 @@ class GMasterChangelogMixin(GMasterCommon):
# entry_ops() that failed... so we retry the _whole_ changelog
# again.
# TODO: remove entry retries when it's gets fixed.
- logging.warn('incomplete sync, retrying changelogs: %s' % \
+ logging.warn('incomplete sync, retrying changelogs: %s' %
' '.join(map(os.path.basename, changes)))
time.sleep(0.5)
@@ -884,15 +943,15 @@ class GMasterChangelogMixin(GMasterCommon):
self.sendmark(path, stime)
def get_worker_status_file(self):
- file_name = gconf.local_path+'.status'
+ file_name = gconf.local_path + '.status'
file_name = file_name.replace("/", "_")
- worker_status_file = gconf.georep_session_working_dir+file_name
+ worker_status_file = gconf.georep_session_working_dir + file_name
return worker_status_file
def update_worker_status(self, key, value):
- default_data = {"remote_node":"N/A",
- "worker status":"Not Started",
- "crawl status":"N/A",
+ default_data = {"remote_node": "N/A",
+ "worker status": "Not Started",
+ "crawl status": "N/A",
"files_syncd": 0,
"files_remaining": 0,
"bytes_remaining": 0,
@@ -909,7 +968,7 @@ class GMasterChangelogMixin(GMasterCommon):
f.flush()
os.fsync(f.fileno())
except (IOError, OSError, ValueError):
- logging.info ('Creating new %s' % worker_status_file)
+ logging.info('Creating new %s' % worker_status_file)
try:
with open(worker_status_file, 'wb') as f:
default_data[key] = value
@@ -920,9 +979,9 @@ class GMasterChangelogMixin(GMasterCommon):
raise
def update_worker_cumilitive_status(self, files_pending):
- default_data = {"remote_node":"N/A",
- "worker status":"Not Started",
- "crawl status":"N/A",
+ default_data = {"remote_node": "N/A",
+ "worker status": "Not Started",
+ "crawl status": "N/A",
"files_syncd": 0,
"files_remaining": 0,
"bytes_remaining": 0,
@@ -932,8 +991,8 @@ class GMasterChangelogMixin(GMasterCommon):
try:
with open(worker_status_file, 'r+') as f:
loaded_data = json.load(f)
- loaded_data['files_remaining'] = files_pending['count']
- loaded_data['bytes_remaining'] = files_pending['bytes']
+ loaded_data['files_remaining'] = files_pending['count']
+ loaded_data['bytes_remaining'] = files_pending['bytes']
loaded_data['purges_remaining'] = files_pending['purge']
os.ftruncate(f.fileno(), 0)
os.lseek(f.fileno(), 0, os.SEEK_SET)
@@ -941,11 +1000,11 @@ class GMasterChangelogMixin(GMasterCommon):
f.flush()
os.fsync(f.fileno())
except (IOError, OSError, ValueError):
- logging.info ('Creating new %s' % worker_status_file)
+ logging.info('Creating new %s' % worker_status_file)
try:
with open(worker_status_file, 'wb') as f:
- default_data['files_remaining'] = files_pending['count']
- default_data['bytes_remaining'] = files_pending['bytes']
+ default_data['files_remaining'] = files_pending['count']
+ default_data['bytes_remaining'] = files_pending['bytes']
default_data['purges_remaining'] = files_pending['purge']
json.dump(default_data, f)
f.flush()
@@ -953,24 +1012,24 @@ class GMasterChangelogMixin(GMasterCommon):
except:
raise
- def update_worker_remote_node (self):
+ def update_worker_remote_node(self):
node = sys.argv[-1]
node = node.split("@")[-1]
remote_node_ip = node.split(":")[0]
remote_node_vol = node.split(":")[3]
remote_node = remote_node_ip + '::' + remote_node_vol
- self.update_worker_status ('remote_node', remote_node)
+ self.update_worker_status('remote_node', remote_node)
- def update_worker_health (self, state):
- self.update_worker_status ('worker status', state)
+ def update_worker_health(self, state):
+ self.update_worker_status('worker status', state)
- def update_worker_crawl_status (self, state):
- self.update_worker_status ('crawl status', state)
+ def update_worker_crawl_status(self, state):
+ self.update_worker_status('crawl status', state)
- def update_worker_files_syncd (self):
- default_data = {"remote_node":"N/A",
- "worker status":"Not Started",
- "crawl status":"N/A",
+ def update_worker_files_syncd(self):
+ default_data = {"remote_node": "N/A",
+ "worker status": "Not Started",
+ "crawl status": "N/A",
"files_syncd": 0,
"files_remaining": 0,
"bytes_remaining": 0,
@@ -981,8 +1040,8 @@ class GMasterChangelogMixin(GMasterCommon):
with open(worker_status_file, 'r+') as f:
loaded_data = json.load(f)
loaded_data['files_syncd'] += loaded_data['files_remaining']
- loaded_data['files_remaining'] = 0
- loaded_data['bytes_remaining'] = 0
+ loaded_data['files_remaining'] = 0
+ loaded_data['bytes_remaining'] = 0
loaded_data['purges_remaining'] = 0
os.ftruncate(f.fileno(), 0)
os.lseek(f.fileno(), 0, os.SEEK_SET)
@@ -990,7 +1049,7 @@ class GMasterChangelogMixin(GMasterCommon):
f.flush()
os.fsync(f.fileno())
except (IOError, OSError, ValueError):
- logging.info ('Creating new %s' % worker_status_file)
+ logging.info('Creating new %s' % worker_status_file)
try:
with open(worker_status_file, 'wb') as f:
json.dump(default_data, f)
@@ -999,19 +1058,19 @@ class GMasterChangelogMixin(GMasterCommon):
except:
raise
- def update_worker_files_remaining (self, state):
- self.update_worker_status ('files_remaining', state)
+ def update_worker_files_remaining(self, state):
+ self.update_worker_status('files_remaining', state)
- def update_worker_bytes_remaining (self, state):
- self.update_worker_status ('bytes_remaining', state)
+ def update_worker_bytes_remaining(self, state):
+ self.update_worker_status('bytes_remaining', state)
- def update_worker_purges_remaining (self, state):
- self.update_worker_status ('purges_remaining', state)
+ def update_worker_purges_remaining(self, state):
+ self.update_worker_status('purges_remaining', state)
- def update_worker_total_files_skipped (self, value):
- default_data = {"remote_node":"N/A",
- "worker status":"Not Started",
- "crawl status":"N/A",
+ def update_worker_total_files_skipped(self, value):
+ default_data = {"remote_node": "N/A",
+ "worker status": "Not Started",
+ "crawl status": "N/A",
"files_syncd": 0,
"files_remaining": 0,
"bytes_remaining": 0,
@@ -1029,7 +1088,7 @@ class GMasterChangelogMixin(GMasterCommon):
f.flush()
os.fsync(f.fileno())
except (IOError, OSError, ValueError):
- logging.info ('Creating new %s' % worker_status_file)
+ logging.info('Creating new %s' % worker_status_file)
try:
with open(worker_status_file, 'wb') as f:
default_data['total_files_skipped'] = value
@@ -1057,9 +1116,12 @@ class GMasterChangelogMixin(GMasterCommon):
if changes:
if purge_time:
logging.info("slave's time: %s" % repr(purge_time))
- processed = [x for x in changes if int(x.split('.')[-1]) < purge_time[0]]
+ processed = [x for x in changes
+ if int(x.split('.')[-1]) < purge_time[0]]
for pr in processed:
- logging.info('skipping already processed change: %s...' % os.path.basename(pr))
+ logging.info(
+ 'skipping already processed change: %s...' %
+ os.path.basename(pr))
self.master.server.changelog_done(pr)
changes.remove(pr)
logging.debug('processing changes %s' % repr(changes))
@@ -1080,7 +1142,9 @@ class GMasterChangelogMixin(GMasterCommon):
# control should not reach here
raise
+
class GMasterXsyncMixin(GMasterChangelogMixin):
+
"""
This crawl needs to be xtime based (as of now
it's not. this is beacuse we generate CHANGELOG
@@ -1091,7 +1155,7 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
files, hardlinks and symlinks.
"""
- XSYNC_MAX_ENTRIES = 1<<13
+ XSYNC_MAX_ENTRIES = 1 << 13
def register(self):
self.counter = 0
@@ -1145,7 +1209,8 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
def open(self):
try:
- self.xsync_change = os.path.join(self.tempdir, 'XSYNC-CHANGELOG.' + str(int(time.time())))
+ self.xsync_change = os.path.join(
+ self.tempdir, 'XSYNC-CHANGELOG.' + str(int(time.time())))
self.fh = open(self.xsync_change, 'w')
except IOError:
raise
@@ -1165,7 +1230,7 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
self.put('xsync', self.fname())
self.counter = 0
if not last:
- time.sleep(1) # make sure changelogs are 1 second apart
+ time.sleep(1) # make sure changelogs are 1 second apart
self.open()
def sync_stime(self, stime=None, last=False):
@@ -1207,7 +1272,7 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
xtr_root = self.xtime('.', self.slave)
if isinstance(xtr_root, int):
if xtr_root != ENOENT:
- logging.warn("slave cluster not returning the " \
+ logging.warn("slave cluster not returning the "
"correct xtime for root (%d)" % xtr_root)
xtr_root = self.minus_infinity
xtl = self.xtime(path)
@@ -1216,7 +1281,7 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
xtr = self.xtime(path, self.slave)
if isinstance(xtr, int):
if xtr != ENOENT:
- logging.warn("slave cluster not returning the " \
+ logging.warn("slave cluster not returning the "
"correct xtime for %s (%d)" % (path, xtr))
xtr = self.minus_infinity
xtr = max(xtr, xtr_root)
@@ -1235,7 +1300,8 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
e = os.path.join(path, e)
xte = self.xtime(e)
if isinstance(xte, int):
- logging.warn("irregular xtime for %s: %s" % (e, errno.errorcode[xte]))
+ logging.warn("irregular xtime for %s: %s" %
+ (e, errno.errorcode[xte]))
continue
if not self.need_sync(e, xte, xtr):
continue
@@ -1256,35 +1322,51 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
self.sync_done(self.stimes, False)
self.stimes = []
if stat.S_ISDIR(mo):
- self.write_entry_change("E", [gfid, 'MKDIR', str(mo), str(st.st_uid), str(st.st_gid), escape(os.path.join(pargfid, bname))])
+ self.write_entry_change("E", [gfid, 'MKDIR', str(mo), str(
+ st.st_uid), str(st.st_gid), escape(os.path.join(pargfid,
+ bname))])
self.Xcrawl(e, xtr_root)
self.stimes.append((e, xte))
elif stat.S_ISLNK(mo):
- self.write_entry_change("E", [gfid, 'SYMLINK', escape(os.path.join(pargfid, bname))])
+ self.write_entry_change(
+ "E", [gfid, 'SYMLINK', escape(os.path.join(pargfid,
+ bname))])
elif stat.S_ISREG(mo):
nlink = st.st_nlink
- nlink -= 1 # fixup backend stat link count
- # if a file has a hardlink, create a Changelog entry as 'LINK' so the slave
- # side will decide if to create the new entry, or to create link.
+ nlink -= 1 # fixup backend stat link count
+ # if a file has a hardlink, create a Changelog entry as
+ # 'LINK' so the slave side will decide if to create the
+ # new entry, or to create link.
if nlink == 1:
- self.write_entry_change("E", [gfid, 'MKNOD', str(mo), str(st.st_uid), str(st.st_gid), escape(os.path.join(pargfid, bname))])
+ self.write_entry_change("E",
+ [gfid, 'MKNOD', str(mo),
+ str(st.st_uid),
+ str(st.st_gid),
+ escape(os.path.join(
+ pargfid, bname))])
else:
- self.write_entry_change("E", [gfid, 'LINK', escape(os.path.join(pargfid, bname))])
+ self.write_entry_change(
+ "E", [gfid, 'LINK', escape(os.path.join(pargfid,
+ bname))])
self.write_entry_change("D", [gfid])
if path == '.':
self.stimes.append((path, xtl))
self.sync_done(self.stimes, True)
+
class BoxClosedErr(Exception):
pass
+
class PostBox(list):
+
"""synchronized collection for storing things thought of as "requests" """
def __init__(self, *a):
list.__init__(self, *a)
# too bad Python stdlib does not have read/write locks...
- # it would suffivce to grab the lock in .append as reader, in .close as writer
+ # it would suffivce to grab the lock in .append as reader, in .close as
+ # writer
self.lever = Condition()
self.open = True
self.done = False
@@ -1319,7 +1401,9 @@ class PostBox(list):
self.open = False
self.lever.release()
+
class Syncer(object):
+
"""a staged queue to relay rsync requests to rsync workers
By "staged queue" its meant that when a consumer comes to the