Merge branch 'master' of skype-git:/git/dba/skytools-3

4 files changed, 29 insertions, 5 deletions

diff --git a/python/a b/python/a
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/python/a
diff --git a/python/londiste/playback.py b/python/londiste/playback.py
index e5fc83a7..aee0c301 100644
--- a/python/londiste/playback.py
+++ b/python/londiste/playback.py
@@ -289,6 +289,8 @@ class Replicator(CascadedWorker):
         #lock_timeout = 10
         # compare: sql to use
         #compare_sql = select count(1) as cnt, sum(hashtext(t.*::text)) as chksum from only _TABLE_ t
+        # workaround for hashtext change between 8.3 and 8.4
+        #compare_sql = select count(1) as cnt, sum(('x'||substr(md5(t.*::text),1,16))::bit(64)::bigint) as chksum from only _TABLE_ t     
         #compare_fmt = %(cnt)d rows, checksum=%(chksum)s
     """
 
diff --git a/python/pgq/consumer.py b/python/pgq/consumer.py
index 77c018c7..b3b45d18 100644
--- a/python/pgq/consumer.py
+++ b/python/pgq/consumer.py
@@ -132,6 +132,10 @@ class Consumer(skytools.DBScript):
 
         # whether to stay behind queue top (postgres interval)
         #pgq_keep_lag = 
+
+        # in how many seconds to write keepalive stats for idle consumers
+        # this stats is used for detecting that consumer is still running
+        #keepalive_stats = 300
     """
 
     # by default, use cursor-based fetch
@@ -157,6 +161,10 @@ class Consumer(skytools.DBScript):
 
     consumer_filter = None
 
+    keepalive_stats = None
+    # statistics: time spent waiting for events
+    idle_start = None
+
     def __init__(self, service_name, db_name, args):
         """Initialize new consumer.
         
@@ -191,6 +199,8 @@ class Consumer(skytools.DBScript):
             raise skytools.UsageError("pgq_autocommit is not compatible with pgq_lazy_fetch")
         self.set_database_defaults(self.db_name, autocommit = self.pgq_autocommit)
 
+        self.idle_start = time.time()
+
     def reload(self):
         skytools.DBScript.reload(self)
 
@@ -209,6 +219,8 @@ class Consumer(skytools.DBScript):
             expr = "ev_extra1 in (%s)" % ','.join(tfilt)
             self.consumer_filter = expr
 
+        self.keepalive_stats = self.cf.getint("keepalive_stats", 300)
+
     def startup(self):
         """Handle commands here.  __init__ does not have error logging."""
         if self.options.register:
@@ -374,11 +386,16 @@ class Consumer(skytools.DBScript):
                     [batch_id, ev_id, retry_time])
 
     def stat_start(self):
-        self.stat_batch_start = time.time()
+        t = time.time()
+        self.stat_batch_start = t
+        if self.stat_batch_start - self.idle_start > self.keepalive_stats:
+            self.stat_put('idle', round(self.stat_batch_start - self.idle_start,4))
+            self.idle_start = t
 
     def stat_end(self, count):
         t = time.time()
         self.stat_put('count', count)
-        self.stat_put('duration', t - self.stat_batch_start)
-
-
+        self.stat_put('duration', round(t - self.stat_batch_start,4))
+        if count > 0: # reset timer if we got some events
+            self.stat_put('idle', round(self.stat_batch_start - self.idle_start,4))
+            self.idle_start = t
\ No newline at end of file
diff --git a/python/walmgr.py b/python/walmgr.py
index 68a75b19..604ba691 100755
--- a/python/walmgr.py
+++ b/python/walmgr.py
@@ -2366,7 +2366,12 @@ STOP TIME: %(stop_time)s
             if fname < last:
                 self.log.debug("deleting %s" % full)
                 if not self.not_really:
-                    os.remove(full)
+                    try:
+                        os.remove(full)
+                    except:
+                        # don't report the errors if the file has been already removed
+                        # happens due to conflicts with pg_archivecleanup for instance.
+                        pass
             cur_last = fname
         return cur_last