First cut at proper transactions for new pubd SQL code.

svn path=/branches/tk705/; revision=5922

5 files changed, 129 insertions, 14 deletions

diff --git a/potpourri/show-key-identifiers.py b/potpourri/show-key-identifiers.py
new file mode 100644
index 00000000..fa2bae8b
--- /dev/null
+++ b/potpourri/show-key-identifiers.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+#
+# $Id$
+#
+# Copyright (C) 2014 Dragon Research Labs ("DRL")
+#
+# Permission to use, copy, modify, and/or distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND DRL DISCLAIMS ALL WARRANTIES WITH
+# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+# AND FITNESS.  IN NO EVENT SHALL DRL BE LIABLE FOR ANY SPECIAL, DIRECT,
+# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+# LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+# OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+# PERFORMANCE OF THIS SOFTWARE.
+
+"""
+Print out URIs, resources, and key identifiers.  Yet another twist on
+the same basic tree walk, just different data fields for a different
+research project.
+"""
+
+import os
+import argparse
+import rpki.POW
+import rpki.oids
+
+
+def check_dir(s):
+  if os.path.isdir(s):
+    return os.path.abspath(s)
+  else:
+    raise argparse.ArgumentTypeError("%r is not a directory" % s)
+
+
+def filename_to_uri(filename):
+  if not filename.startswith(args.rcynic_dir):
+    raise ValueError
+  return "rsync://" + filename[len(args.rcynic_dir):].lstrip("/")
+
+
+def get_roa(fn):
+  return rpki.POW.CMS.derReadFile(fn).certs()[0]
+
+def get_crl(fn):
+  return rpki.POW.CRL.derReadFile(fn)
+
+def get_cer(fn):
+  return rpki.POW.X509.derReadFile(fn)
+
+dispatch = dict(roa = get_roa,
+                crl = get_crl,
+                cer = get_cer)
+
+parser = argparse.ArgumentParser(description = __doc__)
+parser.add_argument("rcynic_dir", type = check_dir, help = "rcynic authenticated output directory")
+args = parser.parse_args()
+
+for root, dirs, files in os.walk(args.rcynic_dir):
+  for fn in files:
+    fn = os.path.join(root, fn)
+    fn2 = os.path.splitext(fn)[1][1:]
+    if fn2 not in dispatch:
+      continue
+    obj = dispatch[fn2](fn)
+    uri = filename_to_uri(fn)
+    try:
+      ski = obj.getSKI().encode("hex")
+    except:
+      ski = ""
+    try:
+      aki = obj.getAKI().encode("hex")
+    except:
+      aki = ""
+    try:
+      res = ",".join(",".join("%s-%s" % r2 for r2 in r1) for r1 in obj.getRFC3779() if r1 is not None)
+    except:
+      res = ""
+    print "\t".join((uri, ski, aki, res))
diff --git a/rpki/pubd.py b/rpki/pubd.py
index a2f996b6..c6a2e2d2 100644
--- a/rpki/pubd.py
+++ b/rpki/pubd.py
@@ -115,7 +115,7 @@ class main(object):
     if self.profile:
       logger.info("Running in profile mode with output to %s", self.profile)
 
-    self.sql = rpki.sql.session(self.cfg)
+    self.sql = rpki.sql.session(self.cfg, autocommit = False)
 
     self.bpki_ta   = rpki.x509.X509(Auto_update = self.cfg.get("bpki-ta"))
     self.irbe_cert = rpki.x509.X509(Auto_update = self.cfg.get("irbe-cert"))
@@ -152,9 +152,17 @@ class main(object):
     Process one PDU from the IRBE.
     """
 
+    # This is still structured with callbacks as if it were
+    # asynchronous, because a lot of the grunt work is done by code in
+    # rpki.xml_utils.  If and when we get around to re-writing the
+    # left-right and publication-control protocols to use lxml.etree
+    # directly, most of the rpki.xml_utils code will go away, but for
+    # the moment it's simplest to preserve the weird calling sequences.
+
     def done(r_msg):
-      self.sql.sweep()
+      self.sql.commit()
       request.send_cms_response(rpki.publication_control.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert))
+      self.sql.commit()
 
     try:
       q_cms = rpki.publication_control.cms_msg(DER = q_der)
@@ -218,17 +226,13 @@ class main(object):
             delta.sql_delete()
             self.session.serial -= 1
             self.session.sql_mark_dirty()
-      #
-      # This isn't really right as long as we're using SQL autocommit;
-      # there should be an SQL ROLLBACK somewhere if anything above fails.
-      #
+
       if delta is not None:
         assert not failed
         delta.activate()
         self.sql.sweep()
         self.session.generate_snapshot()
-
-        # Should SQL commit here
+        self.sql.commit()
 
         # These could be merged, and perhaps should be, among other
         # reasons because we need to do something about expiring old
@@ -245,11 +249,14 @@ class main(object):
         # updates from the delta, but that should be straightforward.
 
       request.send_cms_response(rpki.publication.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert, self.pubd_crl))
+      self.sql.commit()
 
     except Exception, e:
       logger.exception("Unhandled exception processing client query, path %r", request.path)
+      self.sql.rollback()
       request.send_error(500, "Could not process PDU: %s" % e)
 
+
   def uri_to_filename(self, uri):
     """
     Convert a URI to a local filename.
diff --git a/rpki/rpkid_tasks.py b/rpki/rpkid_tasks.py
index 8f652fa6..959d4223 100644
--- a/rpki/rpkid_tasks.py
+++ b/rpki/rpkid_tasks.py
@@ -115,6 +115,7 @@ class AbstractTask(object):
     self.completions.append(completion)
 
   def exit(self):
+    self.self.gctx.sql.sweep()
     while self.completions:
       self.completions.pop(0)(self)
     self.clear()
@@ -122,6 +123,7 @@ class AbstractTask(object):
     self.self.gctx.task_next()
 
   def postpone(self, continuation):
+    self.self.gctx.sql.sweep()
     self.continuation = continuation
     self.due_date = None
     self.self.gctx.task_add(self)
diff --git a/rpki/rtr/main.py b/rpki/rtr/main.py
index 29a4873b..12de30cc 100644
--- a/rpki/rtr/main.py
+++ b/rpki/rtr/main.py
@@ -65,9 +65,9 @@ def main():
 
   argparser = argparse.ArgumentParser(description = __doc__)
   argparser.add_argument("--debug", action = "store_true", help = "debugging mode")
-  argparser.add_argument("--log-level", default = logging.DEBUG,
+  argparser.add_argument("--log-level", default = "debug",
                          choices = ("debug", "info", "warning", "error", "critical"),
-                         type = lambda s: int(getattr(logging, s.upper())))
+                         type = lambda s: s.lower())
   argparser.add_argument("--log-to",
                          choices = ("syslog", "stderr"))
   subparsers = argparser.add_subparsers(title = "Commands", metavar = "", dest = "mode")
@@ -89,6 +89,6 @@ def main():
 
   handler.setFormatter(Formatter(args.debug, fmt, "%Y-%m-%dT%H:%M:%SZ"))
   logging.root.addHandler(handler)
-  logging.root.setLevel(args.log_level)
+  logging.root.setLevel(int(getattr(logging, args.log_level.upper())))
 
   return args.func(args)
diff --git a/rpki/sql.py b/rpki/sql.py
index 9e805ad1..55e6f7cb 100644
--- a/rpki/sql.py
+++ b/rpki/sql.py
@@ -56,11 +56,12 @@ class session(object):
 
   ping_threshold = rpki.sundial.timedelta(seconds = 60)
 
-  def __init__(self, cfg):
+  def __init__(self, cfg, autocommit = True):
 
     self.username = cfg.get("sql-username")
     self.database = cfg.get("sql-database")
     self.password = cfg.get("sql-password")
+    self.autocommit = autocommit
 
     self.conv = MySQLdb.converters.conversions.copy()
     self.conv.update({
@@ -78,7 +79,7 @@ class session(object):
                               passwd = self.password,
                               conv   = self.conv)
     self.cur = self.db.cursor()
-    self.db.autocommit(True)
+    self.db.autocommit(self.autocommit)
     self.timestamp = rpki.sundial.now()
 
   def close(self):
@@ -113,6 +114,31 @@ class session(object):
   def lastrowid(self):
     return self.cur.lastrowid
 
+  def commit(self):
+    """
+    Sweep cache, then commit SQL.
+    """
+
+    self.sweep()
+    logger.debug("Executing SQL COMMIT")
+    self.db.commit()
+
+  def rollback(self):
+    """
+    SQL rollback, then clear cache and dirty cache.
+
+    NB: We have no way of clearing other references to cached objects,
+    so if you call this method you MUST forget any state that might
+    cause you to retain such references.  This is probably tricky, and
+    is itself a good argument for switching to something like the
+    Django ORM's @commit_on_success semantics, but we do what we can.
+    """
+
+    logger.debug("Executing SQL ROLLBACK, discarding SQL cache and dirty set")
+    self.db.rollback()
+    self.dirty.clear()
+    self.cache.clear()
+
   def cache_clear(self):
     """
     Clear the SQL object cache.  Shouldn't be necessary now that the
@@ -136,7 +162,6 @@ class session(object):
     """
 
     for s in self.dirty.copy():
-      #if s.sql_cache_debug:
       logger.debug("Sweeping (%s) %r", "deleting" if s.sql_deleted else "storing", s)
       if s.sql_deleted:
         s.sql_delete()