DBScript: Simplify looping and docstring handling

- looping: remove .do_single_loop and .looping variables, let only .loop_delay
  control looping.  If loop_delay is missing from confing or 0, then
  instead sleep, the script will exit.

- docstr: print docstring fragments recursively, so each class
  needs to only document its own parameters.

- londiste.py: use Replicator class to print default config as it has
  proper class inheritance.

5 files changed, 110 insertions, 61 deletions

diff --git a/python/londiste.py b/python/londiste.py
index 745968ea..11b45fbc 100755
--- a/python/londiste.py
+++ b/python/londiste.py
@@ -1,25 +1,6 @@
 #! /usr/bin/env python
 
 """Londiste launcher.
-
-Config template::
-
-    [londiste]
-    job_name = somedb_worker
-
-    db = dbname=somedb host=127.0.0.1
-
-    queue_name = some_queue
-
-    logfile = ~/log/%(job_name)s.log
-    pidfile = ~/pid/%(job_name)s.pid
-
-    # how many tables can be copied in parallel
-    #parallel_copies = 1
-
-    # sleep time between work loops
-    #loop_delay = 1.0
-
 """
 
 import sys, os, os.path, optparse
@@ -74,7 +55,6 @@ cmd_handlers = (
 )
 
 class Londiste(skytools.DBScript):
-    __doc__ = __doc__
     def __init__(self, args):
         skytools.DBScript.__init__(self, 'londiste', args)
 
@@ -94,6 +74,10 @@ class Londiste(skytools.DBScript):
     def start(self):
         self.script.start()
 
+    def print_ini(self):
+        """Let the Replicator print the default config."""
+        londiste.Replicator(['--ini'])
+
     def init_optparse(self, parser=None):
         p = skytools.DBScript.init_optparse(self, parser)
         p.set_usage(command_usage.strip())
diff --git a/python/londiste/playback.py b/python/londiste/playback.py
index a9d90702..5eed8ae3 100644
--- a/python/londiste/playback.py
+++ b/python/londiste/playback.py
@@ -234,7 +234,18 @@ class TableState(object):
             self.change_snapshot(None)
 
 class Replicator(CascadedWorker):
-    """Replication core."""
+    """Replication core.
+
+    Config options::
+
+        ## Parameters for Londiste ##
+
+        # target database
+        db = dbname=somedb host=127.0.0.1
+
+        # how many tables can be copied in parallel
+        #parallel_copies = 1
+    """
 
     sql_command = {
         'I': "insert into %s %s;",
diff --git a/python/pgq/cascade/worker.py b/python/pgq/cascade/worker.py
index babf7c6e..1c0154b5 100644
--- a/python/pgq/cascade/worker.py
+++ b/python/pgq/cascade/worker.py
@@ -70,6 +70,16 @@ class WorkerState:
 
 class CascadedWorker(CascadedConsumer):
     """CascadedWorker base class.
+
+    Config fragment::
+
+        ## Parameters for pgq.CascadedWorker ##
+
+        # how often the root node should push wm downstream (seconds)
+        #global_wm_publish_period = 300
+
+        # how often the nodes should report their wm upstream (seconds)
+        #local_wm_publish_period = 300
     """
 
     global_wm_publish_time = 0
diff --git a/python/pgq/consumer.py b/python/pgq/consumer.py
index 68382d4a..ef6283d6 100644
--- a/python/pgq/consumer.py
+++ b/python/pgq/consumer.py
@@ -101,8 +101,33 @@ class _BatchWalker(object):
 
 class Consumer(skytools.DBScript):
     """Consumer base class.
+
+    Config template::
+
+        ## Parameters for pgq.Consumer ##
+
+        # queue name to read from
+        queue_name =
+
+        # override consumer name
+        #consumer_name = %(job_name)s
+
+        # whether to use cursor to fetch events (0 disables)
+        #pgq_lazy_fetch = 300
+
+        # whether to wait for specified number of events, before
+        # assigning a batch (0 disables)
+        #pgq_batch_collect_events = 0
+
+        # whether to wait specified amount of time,
+        # before assigning a batch (postgres interval)
+        #pgq_batch_collect_interval =
+
+        # whether to stay behind queue top (postgres interval)
+        #pgq_keep_lag = 
     """
 
+    # by default, use cursor-based fetch
     default_lazy_fetch = 300
 
     # proper variables
@@ -146,9 +171,9 @@ class Consumer(skytools.DBScript):
 
         self.pgq_lazy_fetch = self.cf.getint("pgq_lazy_fetch", self.default_lazy_fetch)
         # set following ones to None if not set
-        self.pgq_min_count = self.cf.getint("pgq_min_count", 0) or None
-        self.pgq_min_interval = self.cf.get("pgq_min_interval", '') or None
-        self.pgq_min_lag = self.cf.get("pgq_min_lag", '') or None
+        self.pgq_min_count = self.cf.getint("pgq_batch_collect_events", 0) or None
+        self.pgq_min_interval = self.cf.get("pgq_batch_collect_interval", '') or None
+        self.pgq_min_lag = self.cf.get("pgq_keep_lag", '') or None
 
     def startup(self):
         """Handle commands here.  __init__ does not have error logging."""
diff --git a/python/skytools/scripting.py b/python/skytools/scripting.py
index 78e725b3..02f00c79 100644
--- a/python/skytools/scripting.py
+++ b/python/skytools/scripting.py
@@ -1,22 +1,6 @@
 
 """Useful functions and classes for database scripts.
 
-Config template::
-
-    [service_type]
-    # generic ini file example from Skytools framework
-    # if job name is omitted then ini file name without extension is used
-    job_name = %(config_name)s
-
-    # loop delay in seconds
-    loop_delay = 1.0
-
-    # location of log and pid files usually kept in ~/log and ~/pid
-    logfile = ~/log/%(job_name)s.log
-    pidfile = ~/pid/%(job_name)s.pid
-
-    # should centralized logging be used
-    use_skylog = 0
 """
 
 import sys, os, signal, optparse, time, errno
@@ -277,14 +261,36 @@ class DBScript(object):
     """Base class for database scripts.
 
     Handles logging, daemonizing, config, errors.
+
+    Config template::
+
+        ## Parameters for skytools.DBScript ##
+
+        # how many seconds to sleep between work loops
+        # if missing or 0, then instead sleeping, the script will exit
+        loop_delay = 1.0
+
+        # where to log
+        logfile = ~/log/%(job_name)s.log
+
+        # where to write pidfile
+        pidfile = ~/pid/%(job_name)s.pid
+
+        # per-process name to use in logging
+        #job_name = %(config_name)s
+
+        # whether centralized logging should be used (loaded from skylog.ini)
+        #use_skylog = 0
+
+        # default lifetime for database connections (in seconds)
+        #connection_lifetime = 1200
     """
     service_name = None
     job_name = None
     cf = None
     log = None
     pidfile = None
-    loop_delay = 1
-    doc_string = None
+    loop_delay = 0
 
     # result from last work() call:
     #  1 - there is probably more work, don't sleep
@@ -308,8 +314,6 @@ class DBScript(object):
         self.service_name = service_name
         self.db_cache = {}
         self.go_daemon = 0
-        self.do_single_loop = 0
-        self.looping = 1
         self.need_reload = 1
         self.stat_dict = {}
         self.log_level = logging.INFO
@@ -350,14 +354,29 @@ class DBScript(object):
     def print_ini(self):
         """Prints out ini file from doc string of the script of default for dbscript
         
-        Used by --ini option on command line
+        Used by --ini option on command line.
         """
 
+        # current service name
+        print("[%s]\n" % self.service_name)
+
+        # walk class hierarchy
+        bases = [self.__class__]
+        while len(bases) > 0:
+            parents = []
+            for c in bases:
+                for p in c.__bases__:
+                    if p not in parents:
+                        parents.append(p)
+                doc = c.__doc__
+                if doc:
+                    self._print_ini_frag(doc)
+            bases = parents
+
+    def _print_ini_frag(self, doc):
         # use last '::' block as config template
-        doc = self.__doc__
         pos = doc and doc.rfind('::') or -1
         if pos < 0:
-            print 'No ini template defined.'
             return
         doc = doc[pos+2 : ].rstrip()
 
@@ -371,9 +390,10 @@ class DBScript(object):
                 pfx = ln[ : wslen]
             if pfx:
                 if ln.startswith(pfx):
-                    print ln[ len(pfx) : ]
+                    print(ln[ len(pfx) : ])
                 else:
-                    print ln
+                    print(ln)
+        print('')
 
     def load_config(self):
         """Loads and returns skytools.Config instance.
@@ -440,7 +460,8 @@ class DBScript(object):
 
     def set_single_loop(self, do_single_loop):
         """Changes whether the script will loop or not."""
-        self.do_single_loop = do_single_loop
+        if do_single_loop:
+            self.loop_delay = 0
 
     def _boot_daemon(self):
         run_single_process(self, self.go_daemon, self.pidfile)
@@ -455,7 +476,7 @@ class DBScript(object):
 
     def stop(self):
         """Safely stops processing loop."""
-        self.looping = 0
+        self.loop_delay = 0
 
     def reload(self):
         "Reload config."
@@ -557,7 +578,7 @@ class DBScript(object):
         # run startup, safely
         self.run_func_safely(self.startup)
 
-        while self.looping:
+        while 1:
             # reload config, if needed
             if self.need_reload:
                 self.reload()
@@ -573,16 +594,14 @@ class DBScript(object):
             for dbc in self.db_cache.values():
                 dbc.refresh()
 
-            # exit if needed
-            if self.do_single_loop:
-                self.log.debug("Only single loop requested, exiting")
-                break
-
             # remember work state
             self.work_state = work
             # should sleep?
             if not work:
-                time.sleep(self.loop_delay)
+                if self.loop_delay:
+                    time.sleep(self.loop_delay)
+                else:
+                    break
 
     def run_once(self):
         return self.run_func_safely(self.work, True)
@@ -598,13 +617,13 @@ class DBScript(object):
             sys.exit(1)
         except SystemExit, d:
             self.send_stats()
-            if prefer_looping and not self.do_single_loop:
+            if prefer_looping and self.loop_delay:
                 self.log.info("got SystemExit(%s), exiting" % str(d))
             self.reset()
             raise d
         except KeyboardInterrupt, d:
             self.send_stats()
-            if prefer_looping and not self.do_single_loop:
+            if prefer_looping and self.loop_delay:
                 self.log.info("got KeyboardInterrupt, exiting")
             self.reset()
             sys.exit(1)
@@ -631,7 +650,7 @@ class DBScript(object):
         # reset and sleep
         self.reset()
         self.exception_hook(d, emsg, cname)
-        if prefer_looping and self.looping and not self.do_single_loop:
+        if prefer_looping and self.loop_delay:
             time.sleep(20)
             return -1
         sys.exit(1)